工具与 Agent

本章深入 create_agent 的工具定义、调用循环与流式/异步 invocation。

定义工具

`@tool` 装饰器

from langchain.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two integers."""
    return a * b

Docstring 与类型注解会进入 JSON Schema，供模型选择参数。

普通 callable

任何带类型注解和 docstring 的 Python 函数也可传入 tools=[...]。

访问运行时上下文

工具可读取 Agent State（如 user_id），通过 ToolRuntime 等机制（见官方 Tools 文档）。典型模式：在 Middleware 中写入 state，工具内读取。

创建 Agent

from langchain.agents import create_agent

agent = create_agent(
    model="openai:gpt-4.1-mini",
    tools=[multiply, search],
    system_prompt="Use tools for math and search. Cite sources.",
    name="assistant",  # 多 Agent 图中作为节点名
)

参数	说明
`model`	字符串或模型实例
`tools`	callable / `@tool` / Tool 对象列表
`system_prompt`	字符串或 SystemMessage
`response_format`	Pydantic 模型，结构化最终输出
`middleware`	中间件列表
`checkpointer`	持久化后端
`name`	子图节点名

invoke

result = agent.invoke({
    "messages": [{"role": "user", "content": "What is 12 * 7?"}]
})

for msg in result["messages"]:
    print(type(msg).__name__, msg.content[:80] if msg.content else msg.tool_calls)

返回的 State 含完整 messages 轨迹，便于调试与审计。

stream

for mode, data in agent.stream(
    {"messages": [{"role": "user", "content": "Research LangGraph."}]},
    stream_mode=["messages", "updates"],
):
    print(mode, data)

messages：token 级流
updates：每个图节点的状态增量（含 tool 开始/结束）

async

result = await agent.ainvoke({"messages": [...]})
async for chunk in agent.astream(...):
    ...

FastAPI / asyncio 服务应使用 async API。

多工具与并行

LangGraph 工具节点可对同一轮多个 tool_calls 并行执行（视工具是否副作用安全）。设计工具时：

读操作可并行
写操作需幂等或加锁
长 IO 考虑超时与重试 Middleware

错误处理

工具抛异常时，框架通常将错误信息作为 ToolMessage 回传模型，由模型决定是否重试或换策略。可在 Middleware 中统一 try/except 与日志。

与 LangGraph ToolNode

自定义 LangGraph 图时，可用预置 ToolNode 执行工具（见 LangGraph 工作流）。create_agent 内部已封装等价逻辑。

反模式

反模式	问题
工具 docstring 模糊	模型乱调参数
巨型万能工具	难调试，易幻觉
无 idempotency 的写工具	重试导致重复写入

#工具与 Agent

#定义工具

#@tool 装饰器

#普通 callable

#访问运行时上下文

#创建 Agent

#invoke

#stream

#async

#多工具与并行

#错误处理

#与 LangGraph ToolNode

#反模式

#下一步