糖豆家

一、整体流程概览

┌─────────────────────────────────────────────────────────────────┐
│                    Tool Call 生成流程                           │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  用户消息 ──▶ LLM 推理 ──▶ 工具调用决策 ──▶ tool_calls 生成     │
│       │                                                      │
│       ▼                                                      │
│  ┌─────────────────────────────────────────────────────────┐  │
│  │  1. model.bind_tools(tools)                             │  │
│  │     ↓                                                   │  │
│  │  2. model.ainvoke(messages)                             │  │
│  │     ↓                                                   │  │
│  │  3. LLM 输出带有 tool_calls 的 AIMessage                │  │
│  └─────────────────────────────────────────────────────────┘  │
│       │                                                      │
│       ▼                                                      │
│  tool_calls 数组 ──▶ guardrail_node 处理 ──▶ execute_tools 执行 │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

二、Tool Call 的来源

2.1 模型绑定工具 (`app/agent/graph.py`)

python

# 文件: app/agent/graph.py, 行 38-41
def build_agent(model: BaseChatModel, tools: list[BaseTool], 
                checkpointer=None, system_suffix: str = ""):
    tools_by_name = {t.name: t for t in tools}           # 行 40: 工具名映射
    model_with_tools = model.bind_tools(tools)           # 行 41: ⚠️ 关键！绑定工具到模型

原理：

python

# 当使用 model.bind_tools(tools) 后
# 模型会被"引导"去生成符合工具 schema 的 tool_calls

model_with_tools = ChatOpenAI(model="gpt-4o").bind_tools([
    ssh_run_tool,    # name="ssh_run", args_schema=SSHRunInput
    list_servers_tool, # name="list_servers"
    clarify_tool,    # name="clarify"
    # ...云 MCP 工具
])

# 模型现在"知道"有哪些工具可用，以及每个工具的参数格式

2.2 LLM 调用生成 tool_calls

python

# 文件: app/agent/graph.py, 行 45-112 (agent_node)
async def agent_node(state: AgentState) -> dict:
    messages = state["messages"]
    
    # 构建系统消息
    if not messages or not isinstance(messages[0], SystemMessage):
        messages = [SystemMessage(content=system_text), *messages]
    
    # 🔥 调用 LLM（关键步骤）
    response = await asyncio.wait_for(
        model_with_tools.ainvoke(messages),  # 行 56: 调用绑定工具的模型
        timeout=60.0
    )
    
    # response 是一个 AIMessage，其中包含 tool_calls 属性
    # ...

三、Tool Call 的结构

3.1 LangChain 的 AIMessage.tool_calls 结构

python

# 当 LLM 决定调用工具时，返回的 AIMessage 包含 tool_calls 属性
response = AIMessage(
    content="我将为您执行这个命令...",
    tool_calls=[
        {
            "name": "ssh_run",                    # 工具名称
            "args": {                             # 工具参数
                "server_name": "web-prod-1",
                "command": "df -h",
                "intent": "查看磁盘使用率",
                "timeout": 600
            },
            "id": "call_abc123xyz"                # 工具调用的唯一 ID
        },
        {
            "name": "ssh_run",
            "args": {
                "server_name": "web-prod-1", 
                "command": "du -sh /var/* | sort -rh | head -5",
                "intent": "查看最大目录",
                "timeout": 600
            },
            "id": "call_def456uvw"
        }
    ]
)

3.2 工具定义（生成 tool_calls schema 的依据）

python

# 文件: app/tools/ssh.py, 行 79-87
ssh_run_tool = StructuredTool.from_function(
    coroutine=_run_on_server,
    name="ssh_run",                              # ← tool_call.name 的来源
    description=(
        "在指定服务器上通过 SSH 执行 shell 命令并返回 stdout/stderr/exit code。"
        "用于巡检、诊断、变更等运维操作。"
    ),
    args_schema=SSHRunInput,                     # ← tool_call.args 的 schema 来源
)

# 文件: app/tools/ssh.py, 行 19-23
class SSHRunInput(BaseModel):
    server_name: str = Field(description="目标服务器名称")   # ← 参数定义
    command: str = Field(description="要在服务器上执行的 shell 命令")
    intent: str = Field(default="", description="命令目的说明")
    timeout: int = Field(default=600, description="超时秒数")

四、Tool Call 生成示例

4.1 对话场景

用户: "检查 web-prod-1 的磁盘使用率"

4.2 LLM 内部推理过程

┌─────────────────────────────────────────────────────────────────┐
│                   LLM 推理过程                                  │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  输入:                                                          │
│  ┌───────────────────────────────────────────────────────────┐  │
│  │ 系统提示词: "你是运维 Agent..."                            │  │
│  │ 用户消息: "检查 web-prod-1 的磁盘使用率"                   │  │
│  │ 可用工具: ssh_run, list_servers, clarify, aliyun-prod__* │  │
│  └───────────────────────────────────────────────────────────┘  │
│                                                                 │
│  LLM 思考:                                                      │
│  ┌───────────────────────────────────────────────────────────┐  │
│  │ 1. 用户要检查磁盘使用率                                     │  │
│  │ 2. 需要用 ssh_run 工具在 web-prod-1 上执行 df -h 命令      │  │
│  │ 3. 命令是只读的，可以自动执行                               │  │
│  │ 4. 需要返回给用户结果                                       │  │
│  └───────────────────────────────────────────────────────────┘  │
│                                                                 │
│  输出:                                                          │
│  ┌───────────────────────────────────────────────────────────┐  │
│  │ AIMessage {                                                │  │
│  │   content: "我来检查 web-prod-1 的磁盘使用率...",          │  │
│  │   tool_calls: [                                            │  │
│  │     {                                                      │  │
│  │       "name": "ssh_run",                                   │  │
│  │       "args": {                                            │  │
│  │         "server_name": "web-prod-1",                       │  │
│  │         "command": "df -h",                                │  │
│  │         "intent": "检查磁盘使用率"                         │  │
│  │       },                                                   │  │
│  │       "id": "call_abc123"                                  │  │
│  │     }                                                      │  │
│  │   ]                                                        │  │
│  │ }                                                          │  │
│  └───────────────────────────────────────────────────────────┘  │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

五、项目中 Tool Call 的处理流程

5.1 路由决策 (`app/agent/graph.py`)

python

# 文件: app/agent/graph.py, 行 210-216
def route_after_agent(state: AgentState) -> str:
    last = state["messages"][-1]
    
    # 检查是否有 tool_calls
    if not getattr(last, "tool_calls", None):
        return END  # 没有工具调用，结束
    
    # 检查是否包含 clarify 工具
    if any(tc["name"] == CLARIFY_TOOL_NAME for tc in last.tool_calls):
        return "clarify"  # 进入澄清节点
    
    # 其他工具调用，进入护栏节点
    return "guardrail"

5.2 护栏节点处理 (`app/agent/graph.py`)

python

# 文件: app/agent/graph.py, 行 115-162
async def guardrail_node(state: AgentState) -> dict:
    last: AIMessage = state["messages"][-1]
    
    # 遍历 LLM 生成的所有 tool_calls
    for tc in last.tool_calls:                    # 行 122
        if not needs_approval(tc["name"]):
            auto_ids.append(tc["id"])             # 不需要审批的工具
            continue
        
        if tc["name"] == "ssh_run":
            cmd = tc["args"].get("command", "")
            # LLM 辅助判断命令风险级别
            level = await classify_command_llm(cmd, model)  # 行 132

5.3 工具执行节点 (`app/agent/graph.py`)

python

# 文件: app/agent/graph.py, 行 165-186
async def execute_tools_node(state: AgentState) -> dict:
    last: AIMessage = state["messages"][-1]
    approved = set(state.get("approved_ids") or [])  # 已批准的 ID
    
    results = []
    for tc in last.tool_calls:                      # 行 169
        level, summary = classify_tool_call(tc["name"], tc["args"])
        
        # 未批准的跳过
        if tc["id"] not in approved:                # 行 171
            results.append(ToolMessage(
                content="[已跳过] 用户本轮未批准执行该命令。",
                tool_call_id=tc["id"], 
                name=tc["name"]
            ))
            continue
        
        # 执行工具
        tool = tools_by_name.get(tc["name"])        # 行 175
        if tool:
            output = str(await tool.ainvoke(tc["args"]))  # 行 180

六、Tool Call 完整生命周期

┌─────────────────────────────────────────────────────────────────┐
│                  Tool Call 完整生命周期                          │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  阶段 1: 工具绑定                                               │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │ build_agent()                                           │   │
│  │   model_with_tools = model.bind_tools(tools)            │   │
│  │   # 模型现在"知道"有哪些工具                             │   │
│  └─────────────────────────────────────────────────────────┘   │
│                          ↓                                      │
│  阶段 2: LLM 推理生成                                          │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │ agent_node()                                            │   │
│  │   response = await model_with_tools.ainvoke(messages)   │   │
│  │   # response.tool_calls = [...]                         │   │
│  └─────────────────────────────────────────────────────────┘   │
│                          ↓                                      │
│  阶段 3: 路由分发                                              │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │ route_after_agent()                                     │   │
│  │   if tool_calls exists: → "guardrail"                   │   │
│  └─────────────────────────────────────────────────────────┘   │
│                          ↓                                      │
│  阶段 4: 风险评估                                              │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │ guardrail_node()                                        │   │
│  │   classify_command_llm() → readonly/mutating/dangerous  │   │
│  │   # dangerous 命令触发 interrupt()                       │   │
│  └─────────────────────────────────────────────────────────┘   │
│                          ↓                                      │
│  阶段 5: 工具执行                                              │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │ execute_tools_node()                                    │   │
│  │   tool.ainvoke(tc["args"])                              │   │
│  │   # 返回 ToolMessage                                    │   │
│  └─────────────────────────────────────────────────────────┘   │
│                          ↓                                      │
│  阶段 6: 结果反馈                                              │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │ ToolMessage 追加到 messages 状态                        │   │
│  │ → 回到 agent_node 继续推理                              │   │
│  └─────────────────────────────────────────────────────────┘   │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

七、Tool Call 的数据结构

python

# 标准 tool_call 结构
tool_call = {
    "name": str,      # 工具名称（如 "ssh_run"）
    "args": dict,     # 工具参数（符合 args_schema）
    "id": str         # 唯一标识符（用于匹配 ToolMessage）
}

# 在项目中的具体示例
tool_call = {
    "name": "ssh_run",
    "args": {
        "server_name": "web-prod-1",
        "command": "df -h",
        "intent": "查看磁盘使用率",
        "timeout": 600
    },
    "id": "call_a1b2c3d4e5f6"
}

# 云 MCP 工具的 tool_call
tool_call = {
    "name": "aliyun-prod__DescribeInstances",  # 账号名__工具名
    "args": {
        "RegionId": "cn-hangzhou",
        "PageSize": 10
    },
    "id": "call_xyz789"
}

八、工具注册与 Tool Call Schema 生成

python

# 每个 StructuredTool 都会生成一个 JSON Schema
# 这个 Schema 会被发送给 LLM，指导它生成正确格式的 tool_calls

SSHRunInput.model_json_schema()
# 生成如下 Schema：
{
    "type": "object",
    "properties": {
        "server_name": {
            "type": "string",
            "description": "目标服务器名称(在后台已登记)"
        },
        "command": {
            "type": "string", 
            "description": "要在服务器上执行的 shell 命令"
        },
        "intent": {
            "type": "string",
            "description": "用一句话说明这条命令的目的/作用",
            "default": ""
        },
        "timeout": {
            "type": "integer",
            "description": "超时秒数",
            "default": 600
        }
    },
    "required": ["server_name", "command"]
}

九、总结

步骤	代码位置	说明
1	`graph.py:41`	`model.bind_tools(tools)` 绑定工具到模型
2	`graph.py:56`	`model_with_tools.ainvoke()` 调用 LLM
3	`graph.py:212`	`getattr(last, "tool_calls", None)` 获取 tool_calls
4	`graph.py:224`	`route_after_agent()` 路由到 guardrail
5	`graph.py:122-141`	遍历 tool_calls 进行风险评估
6	`graph.py:169-185`	`tool.ainvoke(tc["args"])` 执行工具

核心原理：LLM 根据系统提示词中描述的工具能力，结合用户输入，决定是否调用工具以及调用哪些工具。绑定的工具 schema 会引导 LLM 生成符合格式要求的 tool_calls。

一、整体流程概览 ​

二、Tool Call 的来源 ​

2.1 模型绑定工具 (app/agent/graph.py) ​

2.2 LLM 调用生成 tool_calls ​

三、Tool Call 的结构 ​

3.1 LangChain 的 AIMessage.tool_calls 结构 ​

3.2 工具定义（生成 tool_calls schema 的依据） ​

四、Tool Call 生成示例 ​

4.1 对话场景 ​

4.2 LLM 内部推理过程 ​

五、项目中 Tool Call 的处理流程 ​

5.1 路由决策 (app/agent/graph.py) ​

5.2 护栏节点处理 (app/agent/graph.py) ​

5.3 工具执行节点 (app/agent/graph.py) ​

六、Tool Call 完整生命周期 ​

七、Tool Call 的数据结构 ​

八、工具注册与 Tool Call Schema 生成 ​

九、总结 ​

一、整体流程概览

二、Tool Call 的来源

2.1 模型绑定工具 (`app/agent/graph.py`)

2.2 LLM 调用生成 tool_calls

三、Tool Call 的结构

3.1 LangChain 的 AIMessage.tool_calls 结构

3.2 工具定义（生成 tool_calls schema 的依据）

四、Tool Call 生成示例

4.1 对话场景

4.2 LLM 内部推理过程

五、项目中 Tool Call 的处理流程

5.1 路由决策 (`app/agent/graph.py`)

5.2 护栏节点处理 (`app/agent/graph.py`)

5.3 工具执行节点 (`app/agent/graph.py`)

六、Tool Call 完整生命周期

七、Tool Call 的数据结构

八、工具注册与 Tool Call Schema 生成

九、总结