LangChain Agent 工具调用循环里那些 demo 不会告诉你的事

create_react_agent 一行起步的 demo 大家都看过，简洁优雅、五分钟跑通——博文里看着像神器。

可真正接上业务工具之后，麻烦事一桩接一桩：工具抛个异常，直接报错回去；模型在关键参数还没拿到时就开始瞎调；连续失败的时候它会”我不信邪”地一直重试；偶尔还甩你一句”抱歉我无法完成”，把球踢回来。

下面是我后来把这个 ReAct 拆开重写时的几个改动点。

为啥不用 create_react_agent

预制版本本身没啥不好，作为起点也合适：

from langgraph.prebuilt import create_react_agent

agent = create_react_agent(model, tools, prompt=SystemMessage(content=system_prompt))
result = await agent.ainvoke({"messages": [HumanMessage(content=user_message)]})

但它把”model 决定 → 调用工具 → 把结果丢回去”这个循环写死了，跟个铁盒子似的。

我想在工具调用前后插自己的逻辑：参数校验、错误归一化、强制回到工具调用、连续失败熔断——这些活儿，都得自己拼图。

用 StateGraph 自己拼

LangGraph 的 StateGraph 提供了节点和条件边，把上面这些需求拆成节点就行：

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

def _build_agent_graph(self, model, tools, system_prompt):
    tool_node = ToolNode(tools, awrap_tool_call=self._wrap_tool_call)
    finalize_model = model.bind(tool_choice="none")

    async def agent_node(state): ...
    async def precheck_node(state): ...
    async def enforce_tool_node(state): ...
    async def finalize_node(state): ...
    async def update_flags_node(state): ...

    graph = StateGraph(AgentState)
    graph.add_node("agent", agent_node)
    graph.add_node("precheck", precheck_node)
    graph.add_node("run_tools", tool_node)
    graph.add_node("enforce_tool", enforce_tool_node)
    graph.add_node("finalize", finalize_node)
    graph.add_node("update_flags", update_flags_node)

    graph.add_conditional_edges("agent", after_agent, {
        "precheck": "precheck",
        "enforce_tool": "enforce_tool",
        "finalize": "finalize",
    })
    graph.add_conditional_edges("precheck", after_precheck, {
        "run_tools": "run_tools",
        "agent": "agent",
    })
    graph.add_edge("run_tools", "update_flags")
    graph.add_conditional_edges("update_flags", after_update_flags, {
        "agent": "agent",
        "finalize": "finalize",
    })
    graph.add_edge("enforce_tool", "agent")
    graph.add_edge("finalize", END)
    graph.set_entry_point("agent")

    return graph.compile()

各个节点的分工大概是：

节点	做的事
agent	调用 LLM，决定下一步是回话还是调工具
precheck	工具调用前检查参数 / 依赖
run_tools	真正执行工具（LangGraph 自带的 ToolNode）
enforce_tool	检测到模型”偷懒”时，强制注入指令让它必须调工具
finalize	用 bind(tool_choice="none") 强制产出自然语言
update_flags	工具执行后更新失败计数等状态

工具别抛异常

这条是最关键的一条改动，划重点。

原始做法是工具失败直接 raise，LangGraph 默认行为就把异常包成错误终止图——用户那边看到的是一句没头没尾的报错，体验直接拉到底。

更聪明的方式，是工具内部把异常吞了，返回一个结构化的失败消息——让模型自己看见、理解、然后自己修正：

async def _wrap_tool_call(self, request, execute):
    response = await execute(request)
    if not isinstance(response, ToolMessage):
        return response

    parsed = self._parse_tool_output(getattr(response, "content", ""))
    success = getattr(response, "status", "") != "error"

    error_text = None
    if isinstance(parsed, dict):
        if "success" in parsed:
            success = bool(parsed.get("success"))
        if parsed.get("error"):
            error_text = str(parsed.get("error"))

    normalized_payload = {
        "success": success,
        "data": parsed.get("data", parsed) if isinstance(parsed, dict) else parsed,
    }
    if error_text:
        normalized_payload["error"] = error_text

    return ToolMessage(
        content=json.dumps(normalized_payload, ensure_ascii=False),
        name=getattr(response, "name", ""),
        tool_call_id=getattr(response, "tool_call_id", ""),
        status="error" if not success else "success",
    )

这样模型看到的东西是这副样子：

{
  "success": false,
  "data": null,
  "error": "设备 ID 不存在，请先调用 list_devices 获取有效设备列表"
}

模型一看就懂——“哦那我先 list 一下”，自己把这个圈给闭环了。这种自我纠正能力，是大模型最值钱的本事之一，但前提是你得给它看懂的信息。直接甩个 Python traceback 过去？它要么乱猜参数硬重试，要么干脆撂挑子放弃。

precheck：工具调用之前先瞅一眼

有些工具调用，注定要失败——比如 create_automation 需要 device_id，但模型还没查过设备列表，那它肯定是瞎填一个。

与其让它失败再纠正、白白浪费一轮 token，不如在执行前拦下：

async def precheck_node(self, state):
    last_msg = state["messages"][-1]
    if not (isinstance(last_msg, AIMessage) and last_msg.tool_calls):
        return {"_precheck_result": None}

    tool_name = last_msg.tool_calls[0].get("name", "")
    tool_args = last_msg.tool_calls[0].get("args", {})

    precheck_err = self._precheck_tool_call(
        tool_name=tool_name,
        tool_args=tool_args,
        user_message=state.get("user_message"),
    )

    if precheck_err is None:
        return {"_precheck_result": None}

    resolved = await self._resolve_precheck_with_auto_dependency(
        precheck_error=precheck_err,
        tool_name=tool_name,
        tool_args=tool_args,
    )

    if resolved.get("dependency_tool_result") and not resolved["precheck_error"]:
        # 自动执行了依赖工具，把结果塞进消息历史
        return {"messages": [ToolMessage(
            content=str(resolved["dependency_tool_result"]),
            tool_call_id="auto_dep",
        )]}
    else:
        return {"messages": [ToolMessage(
            content=json.dumps(resolved["precheck_error"]),
            tool_call_id=tool_call_id,
        )]}

自动调用依赖工具是个偷懒做法，但确实少绕一圈。要是你担心黑盒过头不可控，可以只生成提示让模型自己去调——灵活度更高，但响应也会慢一点。

enforce_tool：不让模型偷懒

用户说”帮我创建一个温度大于 30 度就开风扇的自动化”——这意图明明白白吧？

可模型偶尔就会回一句”我可以帮你创建，请告诉我设备 ID 和阈值”。诶，你需要的信息其实已经在用户问题里了啊！

这种情况下，注入一个比较硬气的提示，强制让它走工具路径：

def _tool_enforcement_instruction(self, routing, user_message, available_tool_names):
    return f"""你必须调用工具来完成用户的请求，不要直接回复。

用户问题: {user_message}

可用工具: {', '.join(available_tool_names)}

请立即调用合适的工具。"""

注意别无限循环用这招——万一是真的需要追问呢？反复鬼打墙就尴尬了。所以用一个 forced_tool_rounds 计数器卡次数。

状态里要塞的几个标志

熔断和重入这些防御逻辑，全靠状态里几个 flag 撑着：

class AgentState(TypedDict):
    tool_calls_made: int
    tool_failure_streak: int
    last_failed_tool: Optional[str]
    last_tool_error: Optional[str]
    forced_tool_rounds: int

def after_update_flags(self, state):
    # 连续失败 2 次以上，别再试了
    if state.get("tool_failure_streak", 0) >= 2:
        return "finalize"

    # 工具调用次数超过 10 次，强制收尾
    if state.get("tool_calls_made", 0) >= 10:
        return "finalize"

    return "agent"

为啥用”连续”失败而不是”累计”失败？因为同一会话里出错很正常啊，谁还没失手的时候。但连续两次大概率就说明模型把方向带歪了，继续硬试只会烧 token、烧到肉疼。

tool_calls_made >= 10 是终极兜底——正常对话最多走 3-4 轮工具，跑到 10 就说明它陷进迷宫了，得拽出来。

search_tools：让模型自己找工具

工具按意图分组，每次只给模型暴露相关的一小撮（这块在路由那篇文章里讲过）。

问题来了：偶尔会有跨域请求——分类成”设备查询”，但用户实际想问的是设备的历史趋势（这属于”数据分析”工具组）。这时候模型手头没合适的工具，就抓瞎了。

办法是注册一个”元工具” search_tools，让模型自己去搜需要但当前没加载的工具：

@_tool(args_schema=SearchToolsArgs)
async def search_tools(query: str, tags: list = None) -> dict:
    """搜索当前未加载的工具"""
    results = registry.search_tools_by_keyword(query, max_results=10)

    if not results:
        return {
            "success": True,
            "data": {
                "tools": [],
                "message": f"未找到与 '{query}' 匹配的工具",
            },
        }

    return {
        "success": True,
        "data": {
            "tools": results,
            "message": f"找到 {len(results)} 个匹配工具，已动态加载",
        },
    }

search_tools 永远在工具集里，相当于给模型一条”逃生通道”——分类错了它能自己救回来。

代价是 prompt 里得一直占着一个工具描述位。但这账算得很值——路由分类错虽然是低概率事件，可一旦发生影响就大，留个后门更安心。