一、核心原理：Function Calling / Tool Use
┌─────────────────────────────────────────────────────────────────┐
│ LLM Function Calling 工作原理 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 1. 工具定义（Schema） │
│ 把工具的 name、description、parameters 发送给 LLM │
│ │
│ 2. LLM "理解"工具能力 │
│ 根据系统提示词 + 工具描述 + 用户输入，理解每个工具的用途 │
│ │
│ 3. 决策：是否调用工具 + 调用哪个工具 │
│ 分析用户意图，决定是否需要调用工具，以及调用哪些 │
│ │
│ 4. 填充参数 │
│ 根据工具的 parameters schema，填充实际参数值 │
│ │
└─────────────────────────────────────────────────────────────────┘二、项目中的具体实现
2.1 工具 Schema 生成
python
# 文件: app/tools/ssh.py, 行 79-87
ssh_run_tool = StructuredTool.from_function(
coroutine=_run_on_server,
name="ssh_run", # 工具名
description=(
"在指定服务器上通过 SSH 执行 shell 命令并返回 stdout/stderr/exit code。"
"用于巡检、诊断、变更等运维操作。先用 list_servers 查看可用服务器。"
),
args_schema=SSHRunInput, # 参数 schema
)python
# SSHRunInput 定义了参数结构
class SSHRunInput(BaseModel):
server_name: str = Field(description="目标服务器名称(在后台已登记)")
command: str = Field(description="要在服务器上执行的 shell 命令")
intent: str = Field(default="", description="用一句话说明这条命令的目的/作用")
timeout: int = Field(default=600, description="超时秒数")LLM 收到的工具 Schema（简化版）：
json
{
"name": "ssh_run",
"description": "在指定服务器上通过 SSH 执行 shell 命令并返回 stdout/stderr/exit code。用于巡检、诊断、变更等运维操作。先用 list_servers 查看可用服务器。",
"parameters": {
"type": "object",
"properties": {
"server_name": {
"type": "string",
"description": "目标服务器名称(在后台已登记)"
},
"command": {
"type": "string",
"description": "要在服务器上执行的 shell 命令"
},
"intent": {
"type": "string",
"description": "用一句话说明这条命令的目的/作用"
},
"timeout": {
"type": "integer",
"description": "超时秒数",
"default": 600
}
},
"required": ["server_name", "command"]
}
}2.2 系统提示词引导
python
# 文件: app/agent/prompts.py, 行 1-27
SYSTEM_PROMPT = """你是一个资深 SRE / 运维工程师 Agent,能通过 SSH 操作服务器、通过云厂商 MCP 操作云资源,完成复杂运维任务。
## 工作原则
1. **先观察,再动手**:动变更前,先用只读命令(ls/df/ps/systemctl status、云的 Describe/List)摸清现状。
2. **plan-and-execute**:复杂任务先在脑中分解成清晰步骤,逐步执行,每步根据真实结果调整。
3. **最小变更**:只做完成目标必需的操作,不顺手乱改。
4. **危险操作会被拦截**:rm -rf、重启生产、删云资源等高危操作会暂停等待人工审批——这是正常机制,被拒绝就换更安全的方案。
5. **失败要反思**:命令失败时先读 stderr 找根因,不要盲目重试同一条命令。
6. **可解释**:每一步说清你"为什么这么做"和"看到了什么",最后给出结论与后续建议。
## 工具
- `list_servers` / `ssh_run`:查看与操作服务器。调用 `ssh_run` 时,务必在 `intent` 参数里用一句中文说明"这条命令是做什么的"(给人看)。
- `clarify`:当任务不明确、有歧义、或存在多种可行方案、需要用户在范围/风险上拍板时,调用它向用户提问并给出候选项,等用户选择后再继续;信息已足够时不要滥用。单独调用,不要和其它工具混在同一轮。
- 云工具命名形如 `<账号名>__<动作>`(如 `aliyun-prod__DescribeInstances`),Describe/List/Get 为只读,其余为变更。
## 输出
用中文回答。涉及命令/输出用代码块。任务完成时给出简明总结。
"""2.3 模型调用
python
# 文件: app/agent/graph.py, 行 41-56
def build_agent(model: BaseChatModel, tools: list[BaseTool], ...):
tools_by_name = {t.name: t for t in tools}
model_with_tools = model.bind_tools(tools) # ← 关键！绑定工具
system_text = SYSTEM_PROMPT + ...
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]
# 调用模型（模型会"看到"工具定义）
response = await model_with_tools.ainvoke(messages)
# response.tool_calls 包含模型决定调用的工具三、LLM 决策过程图解
┌─────────────────────────────────────────────────────────────────┐
│ LLM 工具调用决策流程 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 输入: │
│ ┌───────────────────────────────────────────────────────────┐ │
│ │ 系统提示词: "你是运维 Agent，可使用 ssh_run, list_servers" │ │
│ │ │ │
│ │ 用户输入: "检查 web-prod-1 的磁盘使用率" │ │
│ │ │ │
│ │ 工具定义: │ │
│ │ - list_servers: 列出服务器 │ │
│ │ - ssh_run: 执行 SSH 命令 │ │
│ │ - clarify: 澄清任务歧义 │ │
│ └───────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌───────────────────────────────────────────────────────────┐ │
│ │ LLM 内部思考 │ │
│ │ │ │
│ │ 1. 用户要我检查磁盘使用率 │ │
│ │ 2. 需要用到 ssh_run 工具 │ │
│ │ 3. 但是我需要知道有哪些服务器 │ │
│ │ 4. 先调用 list_servers 获取服务器列表 │ │
│ │ 5. 然后调用 ssh_run 执行 df -h │ │
│ │ │ │
│ └───────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ 输出: │
│ ┌───────────────────────────────────────────────────────────┐ │
│ │ AIMessage { │ │
│ │ content: "我来帮您检查磁盘使用率...", │ │
│ │ tool_calls: [ │ │
│ │ { │ │
│ │ "name": "list_servers", │ │
│ │ "args": {} │ │
│ │ }, │ │
│ │ { │ │
│ │ "name": "ssh_run", │ │
│ │ "args": { │ │
│ │ "server_name": "web-prod-1", │ │
│ │ "command": "df -h", │ │
│ │ "intent": "检查磁盘使用率" │ │
│ │ } │ │
│ │ } │ │
│ │ ] │ │
│ │ } │ │
│ └───────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘四、影响 LLM 决策的关键因素
4.1 工具描述 (description)
python
# 好的描述帮助 LLM 正确决策
description=(
"在指定服务器上通过 SSH 执行 shell 命令并返回 stdout/stderr/exit code。"
"用于巡检、诊断、变更等运维操作。先用 list_servers 查看可用服务器。"
)| 描述要素 | 作用 |
|---|---|
| 功能说明 | 让 LLM 知道这个工具能做什么 |
| 使用场景 | 帮助 LLM 判断何时使用 |
| 前置条件 | 如"先用 list_servers 查看可用服务器" |
| 注意事项 | 如"单独调用，不要和其它工具混在同一轮" |
4.2 参数描述 (Field description)
python
# 参数描述帮助 LLM 正确填充参数
server_name: str = Field(description="目标服务器名称(在后台已登记)")
command: str = Field(description="要在服务器上执行的 shell 命令")
intent: str = Field(default="", description="用一句话说明这条命令的目的/作用(给人看)")4.3 系统提示词
系统提示词告诉 LLM：
1. 角色定位："你是运维 Agent"
2. 工作原则："先观察，再动手"、"plan-and-execute"
3. 工具使用规范："调用 ssh_run 时，务必在 intent 参数里用一句中文说明"
4. 输出要求："用中文回答"、"任务完成时给出简明总结"五、LLM 判断使用工具的逻辑
┌─────────────────────────────────────────────────────────────────┐
│ LLM 工具调用决策逻辑 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 问题: "检查 web-prod-1 的磁盘使用率" │
│ │
│ LLM 思考过程: │
│ │
│ Step 1: 分析用户意图 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 用户想检查磁盘使用率 │ │
│ │ → 需要在服务器上执行命令 │ │
│ │ → 需要使用 ssh_run 工具 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ ↓ │
│ Step 2: 检查工具描述 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ ssh_run: "在指定服务器上通过 SSH 执行 shell 命令" │ │
│ │ → 匹配用户需求 ✓ │ │
│ └─────────────────────────────────────────────────────────┘ │
│ ↓ │
│ Step 3: 确定参数 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ server_name: "web-prod-1"（从用户输入获取） │ │
│ │ command: "df -h"（根据用户需求构造） │ │
│ │ intent: "检查磁盘使用率"（描述命令目的） │ │
│ └─────────────────────────────────────────────────────────┘ │
│ ↓ │
│ Step 4: 决定调用 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 调用 ssh_run(server_name="web-prod-1", │ │
│ │ command="df -h", │ │
│ │ intent="检查磁盘使用率") │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘六、复杂场景：多工具调用
用户: "查看所有服务器状态，如果有宕机的就重启它"
LLM 决策过程:
┌─────────────────────────────────────────────────────────────────┐
│ Step 1: 列出所有服务器 │
│ → 调用 list_servers │
│ → 返回: [web-prod-1, web-prod-2, db-master] │
├─────────────────────────────────────────────────────────────────┤
│ Step 2: 检查服务器状态（循环调用） │
│ → 调用 ssh_run(server="web-prod-1", command="uptime") │
│ → 调用 ssh_run(server="web-prod-2", command="uptime") │
│ → 调用 ssh_run(server="db-master", command="uptime") │
├─────────────────────────────────────────────────────────────────┤
│ Step 3: 发现宕机服务器（假设 db-master） │
│ → 判断需要重启（mutating 操作） │
│ → ⚠️ 进入 guardrail_node，触发审批 │
├─────────────────────────────────────────────────────────────────┤
│ Step 4: 用户审批通过后 │
│ → 调用 ssh_run(server="db-master", command="sudo systemctl restart nginx") │
└─────────────────────────────────────────────────────────────────┘七、不调用工具的场景
用户: "你好"
LLM 决策:
┌─────────────────────────────────────────────────────────────────┐
│ 问题: "你好" │
│ │
│ 分析: │
│ - 用户只是打招呼 │
│ - 不需要执行任何命令 │
│ - 不需要任何工具 │
│ │
│ 输出: │
│ ┌───────────────────────────────────────────────────────────┐ │
│ │ AIMessage { │ │
│ │ content: "你好！我是运维 Agent，可以帮你：\n1. 检查服务 │ │
│ │ 器状态\n2. 执行运维命令\n3. 操作云资源..." │ │
│ │ tool_calls: [] ← 不调用任何工具 │ │
│ │ } │ │
│ └───────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘八、总结
| 阶段 | 内容 | 代码位置 |
|---|---|---|
| 1. 工具定义 | StructuredTool 包含 name/description/args_schema | ssh.py:79-87 |
| 2. Schema 发送 | model.bind_tools(tools) 传递给 LLM | graph.py:41 |
| 3. 系统提示词 | 引导 LLM 正确使用工具 | prompts.py:1-27 |
| 4. LLM 决策 | 根据用户输入 + 工具描述决定是否/如何调用 | 模型内部 |
| 5. 参数填充 | 根据 args_schema 填充实际参数 | 模型内部 |
| 6. 返回 tool_calls | response.tool_calls 包含调用信息 | graph.py:99 |
核心：LLM 通过"理解"工具的描述和用户的意图，智能决定是否调用工具、调用哪个工具、以及如何填充参数。