LangGraph 工具调用代理

QuibblerAgentQuibblerAgent 2026-11-04 约 21 分钟 3 次阅读

第 4 篇收尾预告过:条件边讲完,官方预制的 Agent 图就该登场了。前四篇一直在从零搭图——五步建图、reducer 定合并语义、条件边管路由——每个零件都亲手摸过一遍。本篇把零件拼回官方成品:create_agent,一个函数直接返回完整的工具调用代理。第 1 篇手搓了 15 行的"模型-工具"循环图,这里一行调用拿下,而且因为机制已经全部拆开讲过,这一行没有黑盒。

版本口径先交代:create_agent 的家在 langchain(1.x 统一入口),从 langchain.agents 导入;它底层搭图用的正是本系列主角 langgraph(PyPI 上 1.2.x),安装 langchain 时会作为依赖一并装上,接口细节以官方 Agents 文档为准。全文按"行数对比 → 最小示例 → 内部图拆解 → system_prompt 与调试 → State 与 reducer → 手搓的保留地 → 实战 → 总结"展开。


1、从手搓图到一行成品

先把第 1 篇那张手搓图原样请回来——模型节点、路由函数、ToolNode、条件边、回边,一样不少:

from langgraph.graph import MessagesState, StateGraph, START, END
from langgraph.prebuilt import ToolNode

def agent(state):
    return {"messages": [model.invoke(state["messages"])]}

def route(state):  # 有工具调用去 tools,否则收束
    return "tools" if state["messages"][-1].tool_calls else END

builder = StateGraph(MessagesState)
builder.add_node("agent", agent)
builder.add_node("tools", ToolNode(TOOLS))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", route)  # 条件边:去或停
builder.add_edge("tools", "agent")             # 回边:循环在此
graph = builder.compile()

这 16 行里,第 2 篇的建图五步、第 3 篇的消息合并、第 4 篇的条件边路由全部在场。同一张图,换成 create_agent 的写法如下:

from langchain.agents import create_agent

graph = create_agent(model="openai:gpt-6", tools=TOOLS)

两行导入加一行调用,16 行变 1 行。从裸 SDK 到现在,工具循环的完整演进汇总如下:

阶段 循环的写法 行数
裸 SDK(第 1 篇开场) while 手写分发与终止 约 10 行
StateGraph 手搓 双节点图加条件边回环 约 15 行
create_agent 一个函数返回成品图 1 行

关键论断先立住:create_agent 不是新机制,而是前四篇机制的官方打包。封装度一路上升,但每一层的原理都已在前面拆开看过,所以"一行"是红利,不是魔法。它内部到底长什么样,第 3 节亲手拆。

2、最小的 create_agent

安装一步:pip install langchain,PyPI 上 1.x 是当前主线,langgraph 与 langchain-core 作为依赖自动装上。第 1 篇交代过这段历史:0.x 时代住在 langgraph.prebuilt 里的 create_react_agent,1.0 起推荐入口收拢到 langchain 侧的 create_agent,本篇直接用新入口。最小的可用代理只靠 model 与 tools 两个参数,完整示例如下:

from langchain.agents import create_agent
from langchain.tools import tool

@tool
def count_words(text: str) -> int:
    """统计一段英文文本的词数"""
    return len(text.split())

agent = create_agent(
    model="openai:gpt-6",      # 字符串形式:提供方冒号模型名
    tools=[count_words],       # @tool 函数直接进列表
)

result = agent.invoke(
    {"messages": [{"role": "user",
                   "content": "统计一下 langgraph is a graph 有几个词"}]}
)
print(result["messages"][-1].content)

输出如下:

这句话共有 4 个词。

解读三件事。其一,model 参数接受两种形态:"openai:gpt-6" 这样的字符串(框架按"提供方冒号模型名"自动初始化),或 init_chat_model 初始化好的模型实例——要精细配置采样参数时用后者。其二,tools 里的元素就是普通函数:@tool 装饰器把函数的 docstring 与类型注解翻译成模型可见的工具描述,细节第 6 篇专讲。其三,返回值是编译好的 LangGraph 图,invoke 喂初始 State、取最终黑板,与第 2 篇的用法一字不差——输出取 result["messages"][-1].content,黑板上最后一条就是模型的最终回答,中间的 tool_calls 与工具结果都按第 3 篇的追加语义躺在列表前段。

3、拆开看内部图

一行调用的背后,create_agent 内部用 StateGraph 搭了一张双节点循环图。以官方实现为准(源码见 GitHub 仓库):模型节点(注册名 model)负责调模型,工具节点(注册名 tools)的本体就是一个 ToolNode;模型节点跑完经条件边路由——最后一条消息带 tool_calls 就去工具节点,不带就到 END;工具节点跑完沿普通边回到模型节点。图画出来是这样:

flowchart LR S([START]) --> A[模型节点] A -->|有 tool_calls| B[工具节点] A -->|无 tool_calls| E([END]) B -->|结果写回| A

解读:这张图与第 4 篇的条件边模型逐点对得上——"有无 tool_calls"就是路由函数读的判断依据,工具节点指回模型节点的那条边就是回边,循环依旧只是图上一条线。步数兜底也没换人:超限照样抛第 2 篇的 GraphRecursionError,recursion_limit 1.0.6 起默认 1000,工具循环不会真的无限转。内部元素与前几篇概念的对应关系整理如下:

内部图元素 对应机制 讲解篇目
模型节点 普通节点函数,读全量写增量 第 2 篇
按 tool_calls 路由 条件边的路由函数 第 4 篇
工具节点 langgraph.prebuilt 的 ToolNode 第 6 篇
tools 回指 model 回边表达循环 第 4 篇
messages 合并 add_messages reducer 第 3 篇

五个元素全部来自前四篇,成品没有任何私货。不信官方打包,可以自己复刻一遍——ToolNode 与现成的路由函数 tools_condition 都从 langgraph.prebuilt 导入:

from langgraph.prebuilt import ToolNode, tools_condition

builder.add_node("model", model_with_tools)   # 模型节点
builder.add_node("tools", ToolNode(TOOLS))    # 工具节点
builder.add_conditional_edges("model", tools_condition)
builder.add_edge("tools", "model")            # 回环

解读:tools_condition 是官方预制的路由函数,语义与第 4 篇手写的 route 完全相同——最后一条 AI 消息带 tool_calls 就返回 tools,否则返回 END。用它有一条约定:工具节点的注册名必须是 tools,因为路由返回值是写死的字符串。这也反过来解释了第 4 节的执行轨迹里片段键为什么是 model 与 tools——stream 吐的正是注册名。手搓与成品完全同构,差别只在这五行由谁来写。

4、system_prompt 与执行轨迹

不加说明的代理没有人设。system_prompt 参数负责这一层,接受字符串或 SystemMessage,框架在每次调模型时把它前置到消息列表头部。同时用 stream 把执行轨迹拉出来看,两个改动合在一起:

agent = create_agent(
    model="openai:gpt-6",
    tools=[count_words],
    system_prompt="你是严谨的中文助手,计数必须调用工具,不得口算。",
)

for chunk in agent.stream(
    {"messages": [{"role": "user",
                   "content": "统计一下 hello langgraph world 有几个词"}]},
    stream_mode="updates",
):
    print(chunk)

输出如下(消息体较长,摘其骨架):

{'model': {'messages': [AIMessage(..., tool_calls=[{'name': 'count_words', ...}])]}}
{'tools': {'messages': [ToolMessage(..., content='3')]}}
{'model': {'messages': [AIMessage(..., content='这句话共有 3 个词。')]}}

解读:三个片段就是三个超步——模型要工具、工具交结果、模型给结论,与第 3 节的内部图一一对应;片段的键正是内部节点名 model 与 tools,第 2 篇的 updates 模式原样适用。system_prompt 之外,签名里的常用可选参数一并交代:

参数 作用
system_prompt 系统提示,字符串或 SystemMessage
state_schema 自定义 State,第 5 节展开
checkpointer 持久化,第 7 篇展开
middleware 官方扩展点,拦截模型调用等环节

调试代理不用加日志:它本来就是一张图,stream 就是它的执行轨迹。

顺带说明:上面看到的是节点级轨迹;token 级的打字机效果要用 stream 事件的另一套模式,第 9 篇流式输出专讲。

5、默认 State 与 reducer

create_agent 的默认 State 官方叫 AgentState,核心字段只有一个 messages,挂的正是第 3 篇的 add_messages reducer——与 MessagesState 同构。循环每一圈都靠它把消息攒进黑板:模型节点带回 AIMessage(含 tool_calls),追加;工具节点交回 ToolMessage,追加;第 3 篇的同 ID 替换规则也全程在场,要改写历史时按原 ID 返回修正版即可。

黑板不止消息一个字段时,用 state_schema 传入 AgentState 的子类,自定义字段照第 3 篇的规则挂 reducer:

import operator
from typing import Annotated
from langchain.agents import AgentState, create_agent

class SearchState(AgentState):
    search_count: Annotated[int, operator.add]   # 检索次数,累加

agent = create_agent(
    model="openai:gpt-6", tools=TOOLS,
    state_schema=SearchState,                    # 换上自定义黑板
)

解读:子类自动继承内置的 messages,新增 search_count 计数字段挂 operator.add,任何节点返回 {"search_count": 1} 都会被累加进黑板。至于谁来写这个字段——普通 @tool 只能返回工具消息,想让工具顺手更新计数,官方姿势是让工具返回 Command(第 4 篇提过的原语),实战篇再展开。第 3 篇的知识在这里原样迁移,成品代理的黑板语义就是它定的。

6、什么时候仍要手搓

成品这么香,手搓是否退场?判断准则一句话:流程由模型决定下一步就用 create_agent,由业务规则决定下一步就回 StateGraph。两类典型场景仍以手搓为主。其一,循环里要插非模型节点:审校回炉(第 3 篇实战)、双模型互评、结果裁决——拓扑不再是"模型-工具"二节点来回,条件边得自己画。其二,非标循环形状:多路分发、Send 驱动的 map-reduce、子图嵌套组网,这些是第 4 篇与第 10 篇的领地,预制图没有对应开关。

折中地带也在生长:create_agent 的 middleware 参数提供官方钩子,能在不动图结构的前提下拦截模型调用、改写输入输出,人审门这类轻量定制够用;重拓扑改动仍建议下沉手搓。选型速查如下:

场景 建议
标准工具循环、问答助手 create_agent 直接连
循环内轻量拦截与人审 create_agent 加 middleware
业务回环与多节点裁决 StateGraph 手搓
多 Agent 组网 图作节点,第 10 篇展开

一句话收束:create_agent 是默认起点,不是天花板;不合身的时候,前四篇的手艺随时接得上。

7、文献检索代理实战

把本篇能力合起来做一个够用的真东西:文献检索代理。两个工具——按关键词检索论文编号、按编号读摘要,数据用字典模拟,骨架与真数据源完全同构。工具定义如下:

from langchain.tools import tool

PAPERS = {
    "p1": "Sparse Attention:稀疏注意力,推理成本降四成",
    "p2": "Graph Agent:工具循环建成显式图,可恢复可中断",
    "p3": "RAG Survey:检索增强生成的评测口径综述",
}
INDEX = {"attention": ["p1"], "agent": ["p2"], "rag": ["p3"]}

@tool
def search_papers(keyword: str) -> list[str]:
    """按关键词检索论文,返回论文编号列表"""
    return INDEX.get(keyword.lower(), [])

@tool
def read_paper(paper_id: str) -> str:
    """按编号读取一篇论文的摘要"""
    return PAPERS.get(paper_id, "未找到该论文")

代理组装与过程打印如下:

from langchain.agents import create_agent

agent = create_agent(
    model="openai:gpt-6",
    tools=[search_papers, read_paper],
    system_prompt="你是文献助理:先检索再阅读,结论必须注明论文编号。",
)

for chunk in agent.stream(
    {"messages": [{"role": "user",
                   "content": "agent 方向有什么论文?核心结论是什么"}]},
    stream_mode="updates",
):
    for name, update in chunk.items():
        print(f"[{name}] 新增 {len(update.get('messages', []))} 条消息")

输出如下:

[model] 新增 1 条消息
[tools] 新增 1 条消息
[model] 新增 1 条消息

解读:三个超步的轨迹——模型先要 search_papers,工具节点交回编号列表;模型接着要 read_paper,拿到 p2 的摘要;最后模型组织答案并注明编号。两把工具接力,中间全靠 add_messages 把对话攒在黑板上,全文没有一行循环代码。顺带留意:如果模型一次要了两个工具,tools 片段会带回两条 ToolMessage,计数变成 2,并行调用第 6 篇细说。想知道每步具体说了什么,把 print 换成逐条打印消息类型与内容即可;接到真数据源只改两个工具函数体,代理一行不动。整段代码三十行出头,其中图结构零行——这就是预制图的红利。

8、总结

create_agent 一个函数返回完整的工具调用代理:model 与 tools 两个参数起步,system_prompt 定人设,返回的是编译好的 LangGraph 图,invoke 与 stream 都是前几篇的老动作。内部图与手搓版完全同构——模型节点、工具节点、按 tool_calls 路由的条件边、tools 回指模型的回边;默认 State 的 messages 字段挂 add_messages,state_schema 可扩展自定义字段。选型口径记一句:模型驱动的标准循环用成品,业务规则驱动的回环回手搓,轻量拦截交给 middleware。

练习两道。练习一:给实战代理加第三个工具 compare(paper_a, paper_b),返回两篇摘要的拼接,让模型对比 p2 与 p3 的适用场景,用 stream 观察工具的接力顺序。练习二:把实战代理的打印改成逐条输出消息类型与内容,给轨迹补全血肉;再故意问一个 INDEX 里没有的关键词(如 quantum),观察模型拿到空结果后如何收场。对于想继续深入的读者,工具本身还有很多可讲究的——参数校验、错误处理、并行调用,建议下一站直接进第 6 篇:ToolNode 与工具循环的细节全在那里。

相关推荐

获取系统SDK版本、判断手机ROM
精选Code

获取系统SDK版本、判断手机ROM

Build获取系统SDK版本Android中部分API的使用,需要在特定的SDK版本之后才能使用,因此在兼容老版本SDK的时候,经常需要判断API的版本。各种Android版本的对应关系参考《Android各版本对应的SDK版本》判断手机ROM有时候需要判断手机系统的ROM,检测ROM是MIUI、EMUI还是Flyme,可以使用getprop命令,去系统build.prop文件(关于build.p

4.2k
AndroidStudio中各种中文乱码问题
精选Code

AndroidStudio中各种中文乱码问题

AndroidStudio中各种中文乱码问题1、编译Java错误信息乱码在出现这个Annotation processors must be explicitly declared now...问题的时候,正好也发现AndroidStudio Build Output错误信息都乱码。经常遇到各种问题,习惯了,现在遇到的问题,都是以后的答案。 1.1、修改项目build.gradle(无效)在整个p

2.8k
PydanticAI 详解:类型安全的 Python AI SDK
精选AI

PydanticAI 详解:类型安全的 Python AI SDK

PydanticAI 详解:类型安全的 Python AI SDK写 LLM 应用最疼的两件事:模型返回的 JSON 不合预期、换一家模型商要改一串代码。pydantic/pydantic-ai(PydanticAI)用 Pydantic 团队的老本行回答了第一件——输出用 BaseModel 定义、运行前先校验,错误从运行时提前到编码时;第二件则用"模型即字符串"回答:换个厂商只改一个参数。如今

75