在工业界和大产品（像 ChatGPT、Claude、OpenAI Agents、企业级 Agent 服务）真正投入生产环境时，他们对“LLM 调工具 / Agent 模式”的实现和设计，远远不是简单的 ReAct 或 bind_tools 组合，而是一套完整的工程级架构，覆盖工具发现、路由、执行、治理、并发、失败恢复、成本控制等多个层面。

下面是生产级 AI Agent 工程实践的关键点总结：

---

1. LLM 只负责意图描述，不执行逻辑

在生产系统中（比如 ChatGPT、OpenAI Agents SDK），模型输出的不是实际代码执行，而是：

{"tool": "...", "arguments": {...}}

然后应用层负责实际调用、参数校验、数据获取和安全审计。模型从结构化输出 → 变成真实操作，是通过中间执行层来做的。

---

2. 工具治理：统一的 Tool Registry（安全边界）

生产级系统不会随便把所有函数或 API 给模型看：

📌 所有工具都有元数据：

• 名称、描述
• 输入参数 JSON Schema
• RBAC 权限配置
• 调用超时/频率限制
• 安全等级 & 访问权限

这个“工具注册表（Tool Registry）”是安全和审计的第一道防线。

---

3. 工具路由 & 动态筛选

当你的工具数量很大时（几十、上百甚至更多），全量注入到模型 prompt 会失效或耗费太多 token，模型也难以从大量工具中准确选择。OpenAI 官方文档和业界最佳实践都推荐先做：

🧠 工具路由层

• 使用向量搜索或关键词匹配，从大工具库里检索出与当前请求最相关的工具
• 动态抽取出一个较小子集（比如 5–10 个）注入模型上下文
• 再让模型决定要不调用这些

这样可以显著提升工具选择准确率，并避免 prompt 膨胀。

---

4. 并行、异步、多路执行

实际生产环境性能要求严苛：

✅ 多个独立工具可以并行执行
(比如同时从不同 API 拉数据、走多个数据源并发执行)

❗ 但需要控制依赖关系
如果 B 工具依赖 A 的结果，不能盲目并发。

并发执行往往通过 async / promise /任务队列来做。

---

5. 错误分类 & 城市恢复（Retry、Fallback）

生产环境对失败的处理比错误结果更重要：

✔ 捕获异常
✔ 超时处理
✔ 识别故障类型（502/网络超时/参数错误）
✔ 按失败类型选择策略

• 重新调用
• 模型反馈错误信息让其换另一种方案
• 退回降级回答
• 人工介入/UI提示

这些处理逻辑不是由模型自动完成，而是 API 层做状态机控制和错误分类。

---

6. 模式解耦：规划 vs 执行

成熟 Agent 系统会分两层：

🧩 规划层

模型做任务分解、序列化计划
不是边做边想，而是：

计划 = [
  {tool: "搜索", args: {...}},
  {tool: "数据库查询", ...},
  {tool: "分析计算", ...}
]

这种方式比 ReAct 更可靠、可审计、可串行化执行。

🔧 执行层

• 串行执行计划
• 跟踪每次调用结果
• 异常回流到规划层调整

---

7. 连续记忆 + Context 管道

生产 Agent 不只是处理一句话，它们需要：

📍 上下文积累
📍 对话状态管理
📍 历史执行记录
📍 结果上下文回传

这保证 Agent 在“长任务、多轮执行、多工具调用”的情境下不丢失上下文状态。ChatGPT 就是在这样一个循环执行机制下工作。

---

8. 安全 & 市场约束

大厂内部还会在工具调用：

🔐 安全沙箱（特别是 Code Exec）
🔒 API 费率限流
🧾 审计日志
📊 监控报警
📈 SLA 保障

这些工程级约束也是生产系统不可或缺的一部分。

---

生产级实践总结

一个成熟的生产环境 Agent 系统通常包括：

1. 工具元数据 & 注册表
2. 意图识别 + 工具路由层
3. 模型结构化输出（纯描述，不执行）
4. 执行层（运行工具 + 并发 +错误治理）
5. 计划/执行分离 + 失败回环逻辑
6. 上下文与状态管理系统
7. 安全、审计、监控、可控性保障层

---

📌 现实案例

• ChatGPT / OpenAI Agents
  模型通过 function calling 产生结构化 tool call，然后由平台统一执行并返回结果，同时注入上下文和安全策略。
• Anthropic MCP + 工具生态
  使用模型上下文协议，使模型能更标准化地与工具、数据库、APIs 安全交互。
• 企业级 SaaS Agent 平台 (如 AskTable)
  内置 Agent 循环机制 + 状态机 + 错误校验 + Output Parser 设计。

Python + LangGraph 生产级骨架：不是简单 ReAct，而是“工具路由 → 动态 bind_tools → 安全执行 → 错误回流 → 最终回答”的结构。

生产级建议用：自定义 LangGraph，不直接全量 ReAct。

核心结构：

User
 ↓
意图识别 / Router
 ↓
从 200+ 工具中筛 Top-K
 ↓
LLM.bind_tools(Top-K tools)
 ↓
Tool Executor
 ↓
错误分类 / 重试 / 降级
 ↓
回到 LLM
 ↓
Final Answer

LangGraph 官方的 create_react_agent 本质是“模型节点调用工具，循环直到停止”；ToolNode 用来执行工具，tools_condition 根据最后一条 AIMessage 是否有 tool calls 来路由。

下面是推荐骨架。

from typing import TypedDict, Annotated, Literal
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage, SystemMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
import traceback
import time


class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    selected_tools: list[str]
    tool_errors: list[dict]
    step_count: int


# 1. 示例工具
@tool
def search_docs(query: str) -> str:
    """Search internal documents by query."""
    return f"docs result for: {query}"


@tool
def query_db(sql: str) -> str:
    """Run a readonly SQL query."""
    if "delete" in sql.lower():
        raise ValueError("Only readonly SQL is allowed")
    return f"db result for: {sql}"


ALL_TOOLS = [search_docs, query_db]
TOOL_MAP = {t.name: t for t in ALL_TOOLS}


# 2. 工具路由：生产环境这里可换成向量检索 / BM25 / 标签召回
def route_tools(state: AgentState):
    last_user = state["messages"][-1].content

    selected = []
    if "数据库" in last_user or "sql" in last_user.lower():
        selected.append("query_db")
    if "文档" in last_user or "搜索" in last_user:
        selected.append("search_docs")

    if not selected:
        selected = ["search_docs"]

    return {
        "selected_tools": selected[:10],
        "step_count": state.get("step_count", 0),
        "tool_errors": state.get("tool_errors", []),
    }


# 3. LLM 节点：只绑定 Top-K 工具，不绑定 200+ 全量工具
def call_model(state: AgentState):
    selected_tools = [TOOL_MAP[name] for name in state["selected_tools"]]

    llm_with_tools = llm.bind_tools(selected_tools)

    system = SystemMessage(content="""
你是生产级 Agent。
规则：
1. 只在必要时调用工具。
2. 工具失败后，不要重复相同错误参数。
3. 如果工具不可用，给出降级回答。
4. 最多调用工具 5 轮。
""")

    response = llm_with_tools.invoke([system] + state["messages"])

    return {
        "messages": [response],
        "step_count": state.get("step_count", 0) + 1,
    }


# 4. 自定义工具执行器：比直接 ToolNode 更适合生产级错误处理
def execute_tools(state: AgentState):
    last_msg = state["messages"][-1]
    tool_messages = []
    errors = state.get("tool_errors", [])

    for call in last_msg.tool_calls:
        name = call["name"]
        args = call["args"]

        try:
            tool_obj = TOOL_MAP[name]

            start = time.time()
            result = tool_obj.invoke(args)
            latency = time.time() - start

            tool_messages.append(
                ToolMessage(
                    tool_call_id=call["id"],
                    name=name,
                    content=str({
                        "ok": True,
                        "result": result,
                        "latency": latency,
                    }),
                )
            )

        except Exception as e:
            err = {
                "tool": name,
                "args": args,
                "error_type": type(e).__name__,
                "error": str(e),
            }
            errors.append(err)

            tool_messages.append(
                ToolMessage(
                    tool_call_id=call["id"],
                    name=name,
                    content=str({
                        "ok": False,
                        "error_type": type(e).__name__,
                        "error": str(e),
                        "hint": "请修正参数，或选择其他工具，或降级回答。",
                    }),
                )
            )

    return {
        "messages": tool_messages,
        "tool_errors": errors,
    }


# 5. 路由条件
def should_continue(state: AgentState) -> Literal["tools", "end"]:
    last_msg = state["messages"][-1]

    if state.get("step_count", 0) >= 5:
        return "end"

    if isinstance(last_msg, AIMessage) and last_msg.tool_calls:
        return "tools"

    return "end"


# 6. 构建 Graph
graph = StateGraph(AgentState)

graph.add_node("route_tools", route_tools)
graph.add_node("agent", call_model)
graph.add_node("tools", execute_tools)

graph.set_entry_point("route_tools")
graph.add_edge("route_tools", "agent")

graph.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        "end": END,
    },
)

graph.add_edge("tools", "agent")

app = graph.compile()


# 7. 调用
result = app.invoke({
    "messages": [HumanMessage(content="帮我查一下文档里关于订单系统的说明")],
    "selected_tools": [],
    "tool_errors": [],
    "step_count": 0,
})

print(result["messages"][-1].content)

生产级关键点：

不要把 200+ 工具一次性 bind_tools
先做工具召回，只给模型 Top-K 工具

不要完全依赖 create_react_agent
复杂业务用自定义 LangGraph

不要让 ToolNode 裸跑
生产环境建议自定义 executor，加入：
- timeout
- retry
- permission check
- 参数校验
- 日志
- fallback
- 最大步数

实际项目里推荐：

工具少于 20 个：create_react_agent 可以
工具 20-100 个：Router + bind_tools
工具 100+ 个：Tool Registry + Tool Retriever + LangGraph 自定义流程

一句话：ChatGPT / Codex 这类产品不是“一个 ReAct 跑到底”，而是模型负责决策，平台负责路由、执行、安全、状态和恢复。

工具调用失败不要只 try/except，生产级要分层处理：

1. 参数错误 → 反馈给 LLM 修正参数
2. 权限错误 → 直接拒绝 / 要授权
3. 超时错误 → 重试 / 降级
4. 网络错误 → 指数退避重试
5. 业务错误 → 返回结构化错误给 LLM
6. 连续失败 → 停止循环，给用户可解释结果

核心做法：

def execute_tools(state):
    last_msg = state["messages"][-1]
    tool_messages = []
    errors = state.get("tool_errors", [])

    for call in last_msg.tool_calls:
        name = call["name"]
        args = call["args"]

        try:
            result = TOOL_MAP[name].invoke(args)

            tool_messages.append(
                ToolMessage(
                    tool_call_id=call["id"],
                    name=name,
                    content={
                        "ok": True,
                        "result": result,
                    },
                )
            )

        except TimeoutError as e:
            error_payload = {
                "ok": False,
                "error_type": "timeout",
                "message": "工具调用超时，可重试或降级回答",
                "retryable": True,
            }

        except PermissionError as e:
            error_payload = {
                "ok": False,
                "error_type": "permission_denied",
                "message": "没有权限调用该工具",
                "retryable": False,
            }

        except ValueError as e:
            error_payload = {
                "ok": False,
                "error_type": "bad_arguments",
                "message": str(e),
                "retryable": True,
                "hint": "请修正工具参数后重新调用",
            }

        except Exception as e:
            error_payload = {
                "ok": False,
                "error_type": type(e).__name__,
                "message": str(e),
                "retryable": False,
            }

        if "error_payload" in locals():
            errors.append({
                "tool": name,
                "args": args,
                **error_payload,
            })

            tool_messages.append(
                ToolMessage(
                    tool_call_id=call["id"],
                    name=name,
                    content=str(error_payload),
                )
            )

            del error_payload

    return {
        "messages": tool_messages,
        "tool_errors": errors,
    }

然后在 LLM system prompt 里明确告诉它：

如果工具返回 ok=false：
- bad_arguments：修正参数后最多重试 1 次
- timeout/network：可重试或换工具
- permission_denied：不要重试，向用户说明权限问题
- 连续失败 2 次：停止工具调用，给出降级回答

最重要的是加最大轮数：

def should_continue(state):
    if state["step_count"] >= 5:
        return "end"

    last_msg = state["messages"][-1]
    if isinstance(last_msg, AIMessage) and last_msg.tool_calls:
        return "tools"

    return "end"

生产级推荐返回给模型的错误格式：

{
  "ok": false,
  "error_type": "bad_arguments",
  "retryable": true,
  "message": "missing required field: user_id",
  "hint": "请补充 user_id 后重新调用"
}

一句话：工具失败不要让程序崩，也不要静默吞掉；要把错误结构化成 ToolMessage 回给 LLM，让它有机会修正，但必须限制重试次数和最大循环。