


在 LangChain/LangGraph 生态里,Function Calling 和 Tool Calling 指的是同一能力——都是让 LLM 输出结构化请求、由后端执行后再把结果喂回模型的一段协议。差别主要在历史演进和工程封装层级上:OpenAI 在 2023 年 6 月先用 functions / function_call 参数推出这个功能,同年 11 月升级为 tools / tool_choice / tool_calls,旧参数已废弃,现代代码统一用 tools。“Tool Calling” 是更通用的演进形态,它不仅涵盖函数,还包括 Web 搜索、代码解释器、MCP 服务等更广泛的工具类型,并支持并行调用。 一、核心概念:Function Calling 与 Tool Calling 的真实关系 在 LangChain 语境里: Function Calling 是 OpenAI 2023 年 6 月首次推出时的原始 API 形态,参数为 functions + function_call。Tool Calling 是 2023 年 11 月之后的统一演进形态,参数为 tools + tool_choice,tool_calls 字段支持一次返回多个调用(并行)。 两者在 2026 年的现代用法中完全同义,OpenAI 官方文档明确写了 “Function calling (also known as tool calling)”。各家的对应叫法: 平台术语控制参数OpenAIFunction calling / Tool callingtool_choiceAnthropic ClaudeTool usetool_choiceGoogle GeminiFunction callingtool_configAzure OpenAITool callingtool_choice一次 Tool Call 的完整生命周期(5 步循环): 开发者用 JSON Schema 定义工具,放进 tools 参数用户提问,模型根据上下文决定要不要调工具模型返回 tool_calls:[{id, type:"function", function:{name, arguments}}],finish_reason 为 tool_calls后端真实执行函数,拿到结果把结果包装成 role:"tool"、tool_call_id 对应的消息回传模型,模型生成最终回复 💡 关键点:模型本身不执行函数,它只产生结构化调用请求。真正的执行、副作用控制、错误处理都在后端。 在 LangChain 里,"工具"被抽象成 BaseTool:包含可调用函数 + 输入 schema + 描述元数据。模型靠 description 选择工具,靠 schema 约束参数。 二、LangChain 侧的 Tool 抽象 LangChain 把工具统一封装为 BaseTool,核心能力由 @tool 装饰器提供: from langchain_core.tools import tool from pydantic import BaseModel, Field from typing import Literal # 简单工具:docstring 自动成为 description,类型提示推断 schema @tool def search_database(query: str, limit: int = 10) -> str: """在客户数据库中搜索匹配查询的记录。 Args: query: 要查找的搜索词 limit: 返回结果的最大数量 """ return f"找到 {limit} 个关于 '{query}' 的结果" 更复杂的场景用 Pydantic 模型定义 args_schema: class WeatherInput(BaseModel): """天气查询的输入。""" location: str = Field(description="城市名称或坐标") units: Literal["celsius", "fahrenheit"] = Field( default="celsius", description="温度单位偏好" ) include_forecast: bool = Field(default=False, description="是否包含5天预报") @tool(args_schema=WeatherInput) def get_weather(location: str, units: str = "celsius", include_forecast: bool = False) -> str: """获取当前天气及可选预报。""" temp = 22 if units == "celsius" else 72 result = f"{location} 当前天气: {temp} 度 {units[0].upper()}" if include_forecast: result += "\n未来5天: 晴天" return result @tool 装饰器的关键参数: description:覆盖 docstring,作为给模型的工具说明args_schema:用 Pydantic 或 JSON Schema 精确控制输入return_direct=True:短路 Agent 循环,工具输出直接作为最终答案返回,不再过一遍 LLMparse_docstring=True:从 docstring 的 Args: 段解析字段描述 InjectedToolCallId 用于在工具内拿到本次调用的 ID,方便返回 ToolMessage: from typing import Annotated from langchain_core.messages import ToolMessage from langchain_core.tools import tool, InjectedToolCallId @tool def foo(x: int, tool_call_id: Annotated[str, InjectedToolCallId]) -> ToolMessage: """Return x.""" return ToolMessage(str(x), artifact=x, name="foo", tool_call_id=tool_call_id) 抛 ToolException 可以让工具有控制地把错误回传给 Agent,而不是中断流程: from langchain_core.tools import ToolException @tool def risky_tool(param: str) -> str: raise ToolException("服务暂时不可用,请稍后重试") 三、用 LangChain 快速搭一个 Tool-Calling Agent 最简单的做法是用 LangGraph 预构建的 create_react_agent: from langchain_openai import ChatOpenAI from langgraph.prebuilt import create_react_agent from langchain_core.tools import tool @tool def get_weather(city: str) -> str: """获取指定城市的当前天气。""" # 实际应用中调用天气 API return f"{city}: 22°C, 晴" @tool def send_email(to: str, subject: str, body: str) -> str: """发送邮件。""" # 实际应用中调用邮件服务 return f"邮件已发送至 {to}" # 1. 初始化支持 tool calling 的模型 model = ChatOpenAI(model="gpt-4o", temperature=0) # 2. 把工具交给 Agent —— 内部就是 LangGraph 图 tools = [get_weather, send_email] agent = create_react_agent(model, tools) # 3. 调用 result = agent.invoke({ "messages": [{"role": "user", "content": "北京天气怎么样?如果低于 25 度就提醒我穿外套"}] }) print(result["messages"][-1].content) create_react_agent 在底层构建的就是一个 LangGraph 图:LLM 节点 → 条件路由 → ToolNode → 回到 LLM,形成 ReAct 循环。 四、用 LangGraph 显式编排(生产级) 当工具调用涉及重试、分支、人工确认、长任务恢复时,应该显式写图: from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode from langchain_openai import ChatOpenAI from langchain_core.messages import ToolMessage from langchain_core.tools import tool # ---------- 1. 定义状态 ---------- class AgentState(TypedDict): messages: Annotated[list, operator.add] # 消息累积 error: Annotated[list, operator.add] # 错误累积 # ---------- 2. 定义工具 ---------- @tool def get_weather(city: str) -> str: """获取天气。""" return f"{city}: 22°C" tools = [get_weather] tool_node = ToolNode(tools, handle_tool_errors=True) # ---------- 3. 定义 LLM 节点 ---------- model = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools) def call_model(state: AgentState): msgs = state["messages"] response = model.invoke(msgs) return {"messages": [response]} # ---------- 4. 路由函数:判断是否继续调工具 ---------- def should_continue(state: AgentState) -> Literal["tools", END]: last = state["messages"][-1] # 如果模型产生了 tool_calls,且不是错误兜底,则去执行工具 if getattr(last, "tool_calls", None): return "tools" return END # ---------- 5. 组装图 ---------- builder = StateGraph(AgentState) builder.add_node("agent", call_model) builder.add_node("tools", tool_node) builder.add_edge(START, "agent") builder.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END}) builder.add_edge("tools", "agent") # 工具结果回传模型,形成循环 # 6. 编译时挂载 checkpointer,支持断点恢复 from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() graph = builder.compile(checkpointer=checkpointer) # 7. 调用(带 thread_id 以支持恢复) config = {"configurable": {"thread_id": "session-001"}} result = graph.invoke( {"messages": [{"role": "user", "content": "北京天气怎么样?"}]}, config=config ) 这张图和 create_react_agent 的本质区别:状态、路由、循环完全在掌控中。你可以插入: 重试逻辑:在 should_continue 里检查 state["error"] 长度,超过阈值走兜底人工确认:在 tools 节点前插一个 human_approval 节点,敏感操作暂停等人类输入(LangGraph 的 interrupt)分支:不同工具结果路由到不同下游节点Checkpoint:graph.get_state(config) 可以取回状态,graph.invoke(None, config) 可以从断点恢复 五、并行工具调用 现代模型支持一次返回多个 tool_calls。在 LangGraph 里 ToolNode 会自动并行执行这些调用。如果要在自己的图里手动并行: from langgraph.graph import StateGraph, START, END # 让 plan 节点分叉到多个独立检索节点 builder.add_edge("plan", "search_A") builder.add_edge("plan", "search_B") # 汇合节点前,State 里要用 reducer 合并 class State(TypedDict): results: Annotated[list, operator.add] # 多个分支 append 到这里 ⚠️ 并行陷阱: 结果顺序:并行节点的返回顺序不确定,不要在汇合节点依赖下标,要用带标识的数据结构部分失败:一个分支失败可能导致整个 fan-in 卡住,要做超时和降级状态合并:共享字段必须有 reducer 函数,否则后面的写入覆盖前面的 六、生产环境的关键坑点 坑 1:工具名和 schema 的跨模型兼容性 不同模型对工具名、参数格式的要求不同。一些模型提供商对包含空格或特殊字符的名称会有问题或拒绝。 # ✅ 推荐:snake_case,字母数字+下划线/连字符 @tool("web_search") def search(query: str) -> str: ... # ❌ 避免:"Web Search"、 "get-weather!" 等 坑 2:参数校验必须由你来做 模型可能产出不符合 schema 的参数,甚至 JSON 解析失败。务必在工具入口做防御: @tool def transfer_money(from_account: str, to_account: str, amount: float) -> str: # 1. 类型与范围校验 if amount <= 0: raise ToolException("转账金额必须大于 0") # 2. 业务校验 if not is_valid_account(from_account): raise ToolException(f"账户 {from_account} 不存在") # 3. 权限校验 if not has_permission(context.user_id, "transfer"): raise ToolException("无转账权限") # 4. 执行(带幂等键) return execute_transfer(from_account, to_account, amount, idempotency_key=...) 坑 3:副作用工具必须幂等 + 人工确认 发邮件、删数据、下订单、退款——这些动作不能让模型随意触发。模式: def should_continue(state): last = state["messages"][-1] if not last.tool_calls: return END # 敏感工具走人工确认节点 sensitive = {"send_email", "delete_record", "refund"} if any(tc["name"] in sensitive for tc in last.tool_calls): return "human_approval" return "tools" 配合 LangGraph 的 interrupt: from langgraph.types import interrupt def human_approval(state): decision = interrupt({ "question": "确认执行敏感操作?", "tool_calls": state["messages"][-1].tool_calls }) if decision == "approve": return {"messages": [], "approved": True} else: return {"messages": [ToolMessage(content="用户拒绝了操作", tool_call_id=...)]} 坑 4:错误处理的三个层级 层级策略实现瞬时错误(超时/429)指数退避重试@retry(stop_after_attempt(3), wait=wait_exponential(...))工具失败Fallback 到备用数据源在 tool_node 外包一层 try/except,调用备用工具不可恢复人工介入 / 明确报错LangGraph 的 interrupt 或返回 ToolExceptionfrom tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) def call_unreliable_api(ticker: str): response = requests.get(f"https://api.example.com/quote/{ticker}") response.raise_for_status() return response.json() def tool_node_with_fallback(state): try: data = call_unreliable_api(state["ticker"]) except Exception as e: # 降级:用缓存价格 data = get_cached_price(state["ticker"]) if not data: # 升级:人工介入 send_slack_alert(f"Price check failed for {state['ticker']}") state["needs_human"] = True return state 坑 5:空结果导致幻觉 工具返回空字符串 "" 时,模型可能"脑补"数据。解决方案是返回显式的 NO_RESULTS_FOUND 信号: @tool def search_kb(query: str) -> str: results = kb.search(query) if not results: return "NO_RESULTS_FOUND: 知识库中未检索到相关内容" return "\n".join(results) 坑 6:流式输出的解析 LangGraph 支持多种 stream mode:updates、values、messages、custom、checkpoints 等。 # 推荐新应用使用 event streaming(LangGraph v1.2+) for chunk in graph.stream( {"messages": [...]}, stream_mode=["updates", "messages"], version="v2" ): if chunk["type"] == "messages": # 处理 token-by-token 的 LLM 输出 print(chunk["data"].content, end="") elif chunk["type"] == "updates": # 节点更新 for node, state_update in chunk["data"].items(): print(f"Node {node} updated") ⚠️ 流式下解析 tool_calls 要特别小心:模型可能分多个 chunk 返回一个 tool_call,需要累积拼接。ToolCallChunk 的合并要求 index 相等且非 None。 坑 7:tool_call_id 必须正确回传 模型返回的每一个 tool_call 都有一个唯一 id。你的 ToolMessage 必须带上对应的 tool_call_id,否则模型无法把结果和调用对应起来: # ❌ 错误:漏了 tool_call_id ToolMessage(content="22°C", name="get_weather") # ✅ 正确 ToolMessage(content="22°C", name="get_weather", tool_call_id="call_abc123") 坑 8:工具数量爆炸 给模型的工具越多,选择错误的几率越高。建议控制在 10 个以内。工具多时用 tool_search 动态加载(仅 gpt-5.4+ 支持),或者按业务域拆分多个 Agent。 坑 9:Checkpoint 与长期记忆 生产环境的长任务必须持久化状态。开发用 MemorySaver,生产要实现 BaseCheckpointSaver 写到 PostgreSQL / Redis / S3: from langgraph.checkpoint.postgres import PostgresSaver with PostgresSaver.from_conn_string(conn_string) as checkpointer: graph = builder.compile(checkpointer=checkpointer) # 即使容器重启,也能从 checkpoint 恢复 state = graph.get_state(config) if state.next: # 还有后续节点要执行 graph.invoke(None, config) # 从断点继续 坑 10:可观测性 复杂 Agent 必须接入 tracing(如 LangSmith),记录每个节点的输入输出、状态变化、工具返回、失败点。没有 tracing 的 Agent 等于盲人摸象。 七、LangChain 还是 LangGraph? 维度LangChain(LCEL/AgentExecutor)LangGraph控制流线性 DAG,拓扑固定循环状态机,运行时路由状态管理基础上下文传递显式 State + Checkpoint重试/分支难实现条件边天然支持人工介入不支持interrupt 原生支持适用场景RAG、简单链、快速原型ReAct Agent、多 Agent、长任务、生产系统判断标准: 线性流程(检索→提示→回答)→ LangChain 足够需要"思考→行动→观察"循环、分支、重试、人工确认 → 必须用 LangGraph工具调用有副作用(发邮件、删库、下订单)→ 强烈建议 LangGraph 📌 注意:坊间有说法称"LangChain 1.0 统一为 create_agent",这是不准确的。正确的预构建入口是 langgraph.prebuilt.create_react_agent,LangGraph 并非退居幕后,而是核心显式依赖。 八、最终总结 Function Calling 与 Tool Calling 本质是同一样东西——都是模型输出结构化调用请求、后端执行的协议。前者是 OpenAI 2023 年 6 月的原始叫法(functions),后者是同年 11 月后的统一演进(tools),支持并行调用和更多工具类型。现代代码一律用 tools。 LangChain 负责"工具抽象":@tool 装饰器 + BaseTool 把 Python 函数封装成模型可理解的、带 schema 和描述的工具。ToolNode 提供高级工具执行控制。 LangGraph 负责"流程编排":用 State + Node + Edge + 条件路由把工具调用组织成可循环、可分支、可恢复的图。create_react_agent 是封装好的 ReAct 图,显式写图则获得完全控制。 生产落地的 10 个核心坑:工具名兼容、参数校验、副作用幂等、三层错误处理、空结果防幻觉、tool_call_id 回传、流式解析、工具数量控制、Checkpoint 持久化、可观测性接入。 架构选型:简单工具调用 LangChain 足够;工具调用一旦有副作用或需要重试/分支/人工确认,必须用 LangGraph。两者不是替代关系——LangGraph 是架在 LangChain 组件之上的流程管理层。 构建一个生产级 Agent 的正确姿势是:用 LangChain 的 @tool 把每个能力封装好,用 LangGraph 的图把这些工具编排成可控、可恢复、有状态的系统,并全程接入 tracing、retry、fallback、human-in-the-loop。