LangChain Core Workflow B: エージェント & ツール

概要

LangChainのエージェント フレームワークを使用して、ツールを利用し、判断を下し、複数ステップのタスクを実行できる自律型エージェントを構築します。

前提条件

• langchain-core-workflow-a(チェーン)の完了
• 関数/ツール呼び出しの概念の理解
• 非同期プログラミングの知識

手順

ステップ 1: ツールを定義する

from langchain_core.tools import tool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="The search query")

@tool(args_schema=SearchInput)
def search_web(query: str) -> str:
    """Search the web for information."""
    # Implement actual search logic
    return f"Search results for: {query}"

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    try:
        result = eval(expression)  # Use safer alternative in production
        return str(result)
    except Exception as e:
        return f"Error: {e}"

@tool
def get_current_time() -> str:
    """Get the current date and time."""
    from datetime import datetime
    return datetime.now().isoformat()

tools = [search_web, calculate, get_current_time]

ステップ 2: ツール付きエージェントを作成する

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

llm = ChatOpenAI(model="gpt-4o-mini")

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant with access to tools."),
    MessagesPlaceholder(variable_name="chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

agent = create_tool_calling_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10,
    handle_parsing_errors=True
)

ステップ 3: エージェントを実行する

# Simple invocation
result = agent_executor.invoke({
    "input": "What's 25 * 4 and what time is it?"
})
print(result["output"])

# With chat history
from langchain_core.messages import HumanMessage, AIMessage

history = [
    HumanMessage(content="Hi, I'm Alice"),
    AIMessage(content="Hello Alice! How can I help you?")
]

result = agent_executor.invoke({
    "input": "What's my name?",
    "chat_history": history
})

ステップ 4: エージェント出力をストリーミングする

async def stream_agent():
    async for event in agent_executor.astream_events(
        {"input": "Search for LangChain news"},
        version="v2"
    ):
        if event["event"] == "on_chat_model_stream":
            print(event["data"]["chunk"].content, end="", flush=True)
        elif event["event"] == "on_tool_start":
            print(f"\n[Using tool: {event['name']}]")

出力

• Pydanticスキーマ付きの型付きツール定義
• エラーハンドリング機能を備えた設定済みエージェント エクゼキューター
• 推論を行いツールを使用できる動作するエージェント
• リアルタイム フィードバック用のストリーミング出力

高度なパターン

非同期サポート付きカスタム ツール

from langchain_core.tools import StructuredTool

async def async_search(query: str) -> str:
    """Async search implementation."""
    import aiohttp
    async with aiohttp.ClientSession() as session:
        # Implement async search
        return f"Async results for: {query}"

search_tool = StructuredTool.from_function(
    func=lambda q: "sync fallback",
    coroutine=async_search,
    name="search",
    description="Search the web"
)

メモリ機能付きエージェント

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

message_history = ChatMessageHistory()

agent_with_memory = RunnableWithMessageHistory(
    agent_executor,
    lambda session_id: message_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)

result = agent_with_memory.invoke(
    {"input": "Remember, I prefer Python"},
    config={"configurable": {"session_id": "user123"}}
)

エラーハンドリング

エラー	原因	解決策
Tool Not Found	ツール名が一致しない	プロンプト内のツール名を確認する
Max Iterations	エージェントがループに陥っている	制限を増やすか、プロンプトを改善する
Parse Error	無効なツール呼び出し形式	handle_parsing_errorsを有効にする
Tool Error	ツール実行が失敗した	ツール関数にtry/exceptを追加する

リソース

• Agents Conceptual Guide
• Tool Calling
• Agent Types

次のステップ

デバッグガイダンスについては、langchain-common-errorsに進んでください。