学习目标
学完本章你应该能够：
- 说清 Agent 和普通 LLM 对话的本质区别——「闭环自主循环（感知→决策→行动→反馈）」vs「一问一答」。
- 讲清一个能用的 Agent 必备的 5 种能力（规划 / 推理 / 工具 / 记忆 / 反思）以及各自最常见的坑。
- 用 ReAct 范式串起一次真实运维排障，画出
Thought → Action → Observation的循环。 - 说清 Function Calling 本质是「结构化输出协议」，并能在代码里正确处理
tool_call_id、arguments 解析、max_turns。 - 讲清多 Agent 协同的角色划分与权限隔离，以及为什么高风险操作必须走 approval。
前置知识：
- 会用至少一种 LLM API（OpenAI 兼容格式）。
- Python 基础（函数、字典、
for循环、异常处理）。 - 了解 K8s 基本概念（Pod / Deployment）会更顺，但不是必须。
本章你会动手做的事：
- 跑通文末的
run_agent最小循环，故意观察「不设max_turns」会怎样失控、设了如何停。 - 给
restart_pod工具加一个dry_run守护：真正删除前先打印将要删除的 Pod。 - 把多 Agent 的 Coordinator / Observer / Executor 用消息协议串起来，模拟一次「OOM → 诊断 → 审批 → 扩容」链路。
类比：普通 LLM 对话像「问路」——你问一句，它答一句，答完拉倒；Agent 像「派出去办事的同事」——他得自己看现场（感知）、想怎么办（决策）、动手干（行动）、看干得对不对（反馈），干不对再调整，直到事办成才回来汇报。核心差异就一句：对话是问答，Agent 是闭环。
flowchart LR
P[感知 Perceive] --> D[决策 Decide]
D --> A[行动 Act]
A -->|结果回灌| F[反馈 Feedback]
F --> P什么是 Agent
说实话我第一次听到 “Agent” 这词儿的时候，心里是抵触的。这玩意儿听着玄学，跟当年 “中台” 一样属于被吹爆的名词。但真上手写过两个之后我发现，它其实没那么神，本质就是个 能自己决定下一步干啥的程序。
跟传统脚本最大的区别是：脚本是 if x then y，写死了；Agent 是 感知 -> 决策 -> 行动 -> 反馈 的循环，决策那一步交给 LLM 来做，所以它能处理写脚本时没想到的情况。
我自己总结的 Agent 自主性循环长这样：
┌─────────────────────────────────────┐
│ │
▼ │
┌──────────┐ ┌──────────┐ ┌──────────┘
│ 感知 │───▶│ 决策 │───▶│ 行动 │
│ Perceive │ │ Decide │ │ Act │
└──────────┘ └──────────┘ └─────┬────┘
▲ │
│ ┌──────────┐ │
└────────────│ 反馈 │◀──────┘
│ Feedback │
└──────────┘
下面用更规范的图把这条自主循环画出来（和上面 ASCII 图是同一个意思，方便你记）：
flowchart LR
P[感知 Perceive
拉状态/查日志/看指标] --> D[决策 Decide
LLM 选工具或回答]
D --> A[行动 Act
删Pod/扩容/改配置]
A -->|结果回灌| F[反馈 Feedback
塞回感知通道]
F --> P挨个解释一下，运维场景里它们分别是啥：
- 感知（Perceive）：拉集群状态、查日志、看指标。Agent 不靠人喂信息，它自己长眼睛。
- 决策（Decide）：LLM 拿着感知到的数据，结合工具列表和记忆，决定接下来调哪个工具或者直接回答。
- 行动（Act）：真正去执行——删 Pod、扩容、改配置、发告警。这一步必须用代码落地，不能让 LLM 直接动 K8s API（别笑，真有人这么干）。
- 反馈（Feedback）：把行动的结果塞回感知通道，下一轮循环用。这步最容易被新手忽略，没有反馈 Agent 就是个开环炮弹，打出去就回不来了。
我个人觉得，Agent 和普通 LLM 对话的核心差异就一句话：对话是问答，Agent 是闭环。ChatGPT 你问一句它答一句，答完拉倒；Agent 必须自己跑循环、自己看结果、自己决定停还是继续。所以写 Agent 的时候，“什么时候停” 是个关键设计点——不设计好它要么无限循环烧 token，要么第一轮就怂了不动手。
运维 Agent 的目标，说白了就是把老 SRE 脑子里那点 “看一眼告警就知道是哪台机器 OOM 了” 的肌肉记忆，变成可自动执行的能力。但请记住，是 “老 SRE” 的肌肉记忆，不是实习生的好奇心——Agent 该不该动手、动多大的手，必须你来定边界。
Agent 的核心能力
把上面那个循环拆开看，一个能用的 LLM-based Agent 通常得有这几把刷子。我按重要性从高到低排：
规划（Planning）
把 “排查线上 P99 飙高” 这种模糊任务，拆成 “查 P99 曲线 → 查 Pod CPU → 查依赖 → 定位根因 → 决定扩容还是重启” 这种可执行子任务。
为什么这么重要？因为 LLM 单次推理能力是有限的。你直接问它 “线上慢了怎么办”，它会给你背一遍排查手册；但你让它先拆任务再逐步执行，它就能真的定位到问题。规划能力差，Agent 就是个嘴炮。
实现上有两种主流做法：
- Task Decomposition：一次性把任务全拆完，类似
Plan -> Execute。简单可控，但遇到意外不会调整。 - ReAct / 逐步规划：每步边想边做，根据上一步结果调整。灵活但 token 烧得多，容易跑偏。
我个人偏好后者，因为运维场景太脏了，预先规划基本都会被打脸。
推理（Reasoning）
基于观察到的信息做逻辑推断。比如 “Pod 重启 12 次 + 最后一次 OOMKilled + 内存 limit 256Mi” → “大概率内存给小了”。
这能力靠的是 LLM 本身，但你能做的是 给够上下文。LLM 推理翻车，十次有八次是因为上下文给少了，不是模型笨。
工具使用（Tool Use）
调外部 API、脚本、数据库。这是 Agent 的手脚，没工具的 Agent 就是个会打字的瘫痪病人。
工具设计有个反直觉的点：工具描述越短越准，反而越好用。我之前给一个 restart_pod 工具写了一堆参数说明、返回值、异常情况，结果 LLM 经常不敢调；后来砍到一句话 “重启指定 Pod，参数 namespace 和 pod_name”，调用率立马上来了。LLM 跟人一样，选项越多越纠结。
记忆（Memory）
分短期和长期。短期就是当前任务的对话历史，长期是跨会话的知识。
短期记忆有个大坑：context window 有限，对话一长就爆。常见解法是滑动窗口 + 摘要，把老消息压缩成一段总结。但摘要本身也会丢信息，所以重要的事实（比如 “用户已经确认可以重启”）最好单独存一份，别让它进摘要。
长期记忆通常配 RAG 用，下面会讲。
反思（Reflection）
执行完一步看结果不对，能自己说一句 “嗯这条路走错了，换一个”。这能力高级 Agent 才有，简单的 ReAct 不带。
实现上就是让 LLM 在每轮循环最后多输出一段 “self-critique”，自己问自己 “刚才那步有效吗？要不要换方向”。听着美好，但实测 LLM 经常会过度反思——明明走对了，它非觉得自己错了要换路。所以反思我建议只在 “工具调用失败” 或 “连续 N 步没进展” 时触发，别每步都反思。
ReAct 推理模式
ReAct 的核心就是「想一步、动一步、看结果、再想」的循环。下面这张图把
Thought → Action → Observation的闭环画出来，理解它你就理解了 Agent 为什么比「直接 Action」稳：
flowchart TD
Q[用户问题] --> T[Thought 思考]
T --> Ac[Action 调工具]
Ac --> O[Observation 观察结果]
O -->|未解决 继续| T
O -->|已解决| Ans[给出最终答案]ReAct（Reasoning + Acting）是 Agent 里最经典的范式。核心就一句话：让模型先想再动，动了看结果再想。
为啥要这样？因为 LLM 直接 Action 经常瞎调——你问它 Pod 为啥重启，它上来就 restart_pod，根本没看日志。让它先输出 Thought，强迫它把推理过程写出来，错误率会明显下降。
我用一个真实的运维场景演示：线上一个 Pod api-gateway-7d8-x2 一直重启，SRE 接到告警，丢给 Agent 去查。
问题：prod 命名空间的 Pod api-gateway-7d8-x2 持续重启，重启计数 14 次，
最近一次状态 CrashLoopBackOff，需要定位根因并决定是否自动修复。
Thought 1: 我得先看这个 Pod 的当前状态和容器状态，确认是不是真的 CrashLoopBackOff，
以及最后退出原因是什么。
Action 1: get_pod(namespace="prod", pod_name="api-gateway-7d8-x2")
Observation 1: {
"phase": "Running",
"containerStatuses": [{
"name": "api-gateway",
"restartCount": 14,
"lastState": { "terminated": { "reason": "OOMKilled", "exitCode": 137 } },
"state": { "running": { "startedAt": "2025-11-04T06:12:33Z" } }
}]
}
Thought 2: 上一次是 OOMKilled 退出的（exitCode 137），说明容器内存超了被内核 kill。
但现在又 Running 了，说明 K8s 把它拉起来了。我得看日志，确认死之前在干啥。
Action 2: get_pod_logs(namespace="prod", pod_name="api-gateway-7d8-x2", tail=200)
Observation 2: 日志最后 50 行全是 "Allocating batch of 50000 records, mem=240MB"，
然后突然断掉，没有正常退出日志。
Thought 3: 240MB 已经接近 limit 256Mi 了，加上 JVM 元数据肯定爆。这是典型的
内存给小了，单纯重启 Pod 治标不治本，下次还会 OOM。正确做法是扩内存 limit。
但改 limit 涉及改 Deployment，风险偏高，应该先确认有没有最近发版。
Action 3: get_deployment(namespace="prod", name="api-gateway")
Observation 3: Deployment 昨晚 23:00 刚发版，image 从 v1.4.2 升到 v1.5.0，
v1.5.0 的 changelog 提到 "支持批量处理，单批默认 50000 条"。
Thought 4: 找到根因了——v1.5.0 默认批量从 10000 改成 50000，导致内存翻倍。
两个选项：A) 改回批量大小环境变量 BATCH_SIZE=10000；B) 扩 memory limit。
A 风险更小且可逆，先做 A。但我需要人确认是否降级，因为可能影响吞吐。
Action 4: ask_human(question="建议把 BATCH_SIZE 改回 10000，吞吐会下降到原来的 1/5，
是否确认？或选择扩 memory limit 到 512Mi。")
Observation 4: 人工确认：扩 memory limit 到 512Mi。
Thought 5: 收到。执行 patch deployment 修改内存 limit。改完之后等 Pod 滚动更新，
再观察 5 分钟重启计数是否还涨。
Action 5: patch_deployment_resources(namespace="prod", name="api-gateway",
container="api-gateway", memory_limit="512Mi", memory_request="256Mi")
你看，整个推理链是清晰的、可审计的。每一步 LLM 都说了 “我看到啥 -> 我在想啥 -> 我打算干啥”，出问题随时能定位是哪一步推理错了。
踩坑提示：ReAct 最常见的两个坑——一是 Thought 写太长，LLM 一思考就停不下来，token 烧得哗哗的，建议在 prompt 里限定 “Thought 不超过 3 句话”；二是 Observation 塞太多数据，比如把 Pod 的完整 JSON 扔进去，context 立刻爆。Observation 一定要预处理，只给关键字段。
Function Calling 与工具定义
Function Calling 是 LLM 跟外部世界交互的桥梁。说实话我觉得这玩意儿被神化了，本质就是个 结构化输出协议——LLM 不再吐自由文本，而是吐一个 JSON，告诉你 “调哪个函数、传什么参数”。
流程通常是：
- 你预先定义好一批工具（函数名、参数 schema、描述）。
- 用户提问 + 工具定义一起喂给 LLM。
- LLM 决定要不要调工具，要调就输出一个
tool_callsJSON。 - 你的代码执行这个函数，把结果作为
tool角色消息塞回去。 - LLM 看到结果继续推理，或者再调下一个工具，或者直接答用户。
OpenAI 兼容格式的工具定义长这样（运维场景，定义三个常用工具）：
{
"tools": [
{
"type": "function",
"function": {
"name": "get_pod",
"description": "获取指定 Pod 的当前状态，包括 phase、容器状态、重启次数、最后退出原因",
"parameters": {
"type": "object",
"properties": {
"namespace": {
"type": "string",
"description": "Pod 所在的命名空间"
},
"pod_name": {
"type": "string",
"description": "Pod 名称"
}
},
"required": ["namespace", "pod_name"]
}
}
},
{
"type": "function",
"function": {
"name": "restart_pod",
"description": "删除指定 Pod，由控制器重新拉起。仅用于 CrashLoopBackOff 或无状态服务重启场景，禁止用于 StatefulSet",
"parameters": {
"type": "object",
"properties": {
"namespace": { "type": "string" },
"pod_name": { "type": "string" },
"dry_run": {
"type": "boolean",
"description": "是否只模拟不真正执行，默认 false",
"default": false
}
},
"required": ["namespace", "pod_name"]
}
}
},
{
"type": "function",
"function": {
"name": "patch_deployment_resources",
"description": "修改 Deployment 中指定容器的 CPU/内存 request 与 limit",
"parameters": {
"type": "object",
"properties": {
"namespace": { "type": "string" },
"name": { "type": "string", "description": "Deployment 名称" },
"container": { "type": "string", "description": "容器名称" },
"cpu_request": { "type": "string", "description": "如 100m、500m、1" },
"cpu_limit": { "type": "string" },
"memory_request": { "type": "string", "description": "如 128Mi、256Mi" },
"memory_limit": { "type": "string" }
},
"required": ["namespace", "name", "container"]
}
}
}
]
}
注意几个细节：
description是 LLM 选工具的唯一依据，必须说清楚 “干啥用、什么时候该用、什么时候不该用”。我上面restart_pod的描述里特意写了 “禁止用于 StatefulSet”，因为之前真发生过 LLM 把 StatefulSet 的 Pod 删了导致数据错乱的事故。dry_run这种参数建议每个写操作都带，调试和审计用得着。- 枚举值用
enum字段，别让 LLM 自由发挥字符串。
下面是一段完整的 Python 调用代码，OpenAI 兼容 API，带错误处理和工具执行循环：
import json
import requests
from typing import Callable
# 工具实现注册表：函数名 -> 实际可调用对象
# 这里用假的实现，真实场景里换成 client-go 调 K8s API 的封装
TOOL_REGISTRY: dict[str, Callable] = {
"get_pod": lambda ns, name: {
"phase": "Running",
"restartCount": 14,
"lastState": {"terminated": {"reason": "OOMKilled", "exitCode": 137}},
},
"restart_pod": lambda ns, name, dry_run=False: (
{"action": "deleted", "namespace": ns, "pod_name": name}
if not dry_run
else {"action": "dry_run_ok", "namespace": ns, "pod_name": name}
),
"patch_deployment_resources": lambda **kw: {"patched": True, **kw},
}
# 工具的 JSON Schema 定义，喂给 LLM 用
TOOLS_SCHEMA = [
{
"type": "function",
"function": {
"name": "get_pod",
"description": "获取指定 Pod 的当前状态",
"parameters": {
"type": "object",
"properties": {
"namespace": {"type": "string"},
"name": {"type": "string"},
},
"required": ["namespace", "name"],
},
},
},
{
"type": "function",
"function": {
"name": "restart_pod",
"description": "删除指定 Pod 触发重启，禁止用于 StatefulSet",
"parameters": {
"type": "object",
"properties": {
"namespace": {"type": "string"},
"name": {"type": "string"},
"dry_run": {"type": "boolean", "default": False},
},
"required": ["namespace", "name"],
},
},
},
]
def call_llm(messages: list[dict], tools: list[dict]) -> dict:
"""调用 OpenAI 兼容 API，返回原始响应 JSON。"""
# 这里用 requests 直连，生产环境建议用官方 SDK，能省一堆重试逻辑
resp = requests.post(
"https://api.example.com/v1/chat/completions",
headers={"Authorization": "Bearer " + MY_API_KEY},
json={
"model": "gpt-4o",
"messages": messages,
"tools": tools,
"tool_choice": "auto", # auto 让模型自己决定调不调
},
timeout=60,
)
resp.raise_for_status()
return resp.json()
def run_agent(user_query: str, max_turns: int = 10) -> str:
"""跑一个最简 Agent 循环：LLM 出 tool_call -> 执行 -> 把结果塞回去 -> 直到 LLM 不再调工具。"""
messages = [
{"role": "system", "content": "你是一名 Kubernetes SRE Agent，按需调用工具排查问题。"},
{"role": "user", "content": user_query},
]
for turn in range(max_turns):
# 1. 调 LLM
resp = call_llm(messages, TOOLS_SCHEMA)
msg = resp["choices"][0]["message"]
messages.append(msg)
# 2. 没工具调用 = Agent 觉得搞定了，直接返回
if not msg.get("tool_calls"):
return msg.get("content", "")
# 3. 有工具调用，逐个执行
for tool_call in msg["tool_calls"]:
fn_name = tool_call["function"]["name"]
# 有坑：LLM 输出的 arguments 是字符串，必须 json.loads
try:
args = json.loads(tool_call["function"]["arguments"])
except json.JSONDecodeError as e:
args = {}
print(f"[warn] LLM 输出的 arguments 不是合法 JSON: {e}")
if fn_name not in TOOL_REGISTRY:
# 工具不存在时也要回一个 tool 消息，否则 LLM 会一直等结果
result = {"error": f"unknown tool: {fn_name}"}
else:
try:
result = TOOL_REGISTRY[fn_name](**args)
except Exception as e:
# 关键：工具异常不能让 Agent 整个挂，把错误回给 LLM 让它自己处理
result = {"error": f"tool execution failed: {str(e)}"}
print(f"[turn {turn}] {fn_name}({args}) -> {result}")
# 4. 把工具结果作为 tool 角色消息塞回对话
messages.append({
"role": "tool",
"tool_call_id": tool_call["id"], # 有坑：这个 id 必须对上，不然 API 报错
"content": json.dumps(result, ensure_ascii=False),
})
return "达到最大轮次，Agent 未能给出最终答案。"
if __name__ == "__main__":
answer = run_agent("prod 命名空间的 api-gateway-7d8-x2 一直重启，帮我看看。")
print("Agent 最终回答：", answer)
踩坑提示：
tool_call_id必须严格对上。OpenAI API 要求每个 tool 角色消息的tool_call_id必须对应assistant消息里某个tool_call.id，错一个就报 400。- arguments 一定要 try-except 解析。LLM 偶尔会吐不合法的 JSON，特别是参数里有引号或换行的时候。
max_turns必须设。我见过 Agent 卡死循环烧了 200 美金 token 的真实案例。- 工具异常别 raise，回 JSON 给 LLM。LLM 看到
{"error": "..."}会自己调整策略，你 raise 整个 Agent 就挂了。
Embedding 与 RAG
Agent 光会调工具还不够，遇到 “我们公司线上 Prometheus 怎么配的” 这种问题，工具调不出来——这种知识在公司 wiki 里。Embedding + RAG 就是用来把私有知识塞给 Agent 的。
RAG 流程不复杂，但每一步都有坑。先看整体：
建库阶段（一次性）：
文档 -> 切分 -> embedding -> 存入向量库
查询阶段（每次提问）：
问题 -> embedding -> 向量检索 top-k -> 拼到 prompt -> LLM 生成
我先把建库和查询代码都写出来，用 Python + 一个最简的内存向量库（生产环境换 Milvus/Qdrant）：
import json
import numpy as np
from dataclasses import dataclass, field
from typing import Iterable
# 假设你有一个 embedding 模型 API，输入文本返回 1536 维向量
# 生产环境常用 text-embedding-3-small / bge-large-zh / m3e-base
EMBED_DIM = 1536
def embed(texts: list[str]) -> list[list[float]]:
"""调用 embedding API，批量返回向量。"""
resp = requests.post(
"https://api.example.com/v1/embeddings",
headers={"Authorization": "Bearer " + MY_API_KEY},
json={"model": "text-embedding-3-small", "input": texts},
timeout=30,
)
resp.raise_for_status()
return [d["embedding"] for d in resp.json()["data"]]
@dataclass
class VectorStore:
"""最简内存向量库，用 numpy 算余弦相似度。
生产环境请换 Milvus / Qdrant / Weaviate，本实现不持久化、不支持过滤。"""
chunks: list[str] = field(default_factory=list)
vectors: list[np.ndarray] = field(default_factory=list)
def insert(self, texts: Iterable[str]) -> None:
texts = list(texts)
vecs = embed(texts)
self.chunks.extend(texts)
self.vectors.extend(np.array(v) for v in vecs)
def search(self, query: str, top_k: int = 5) -> list[tuple[str, float]]:
if not self.vectors:
return []
q = np.array(embed([query])[0])
mat = np.vstack(self.vectors) # (N, dim)
# 余弦相似度：归一化后点积
q_norm = q / (np.linalg.norm(q) + 1e-9)
mat_norm = mat / (np.linalg.norm(mat, axis=1, keepdims=True) + 1e-9)
sims = mat_norm @ q_norm # (N,)
# 取 top_k 的下标
idx = np.argsort(-sims)[:top_k]
return [(self.chunks[i], float(sims[i])) for i in idx]
def split_documents(docs: list[str], chunk_size: int = 500, overlap: int = 50) -> list[str]:
"""按字符长度切分，带 overlap 避免切断上下文。
生产环境建议按 token 切，且优先按段落/标题切，别傻按字符。"""
chunks = []
for doc in docs:
start = 0
while start < len(doc):
end = start + chunk_size
chunks.append(doc[start:end])
start = end - overlap # 关键：overlap 防止边界信息丢
return chunks
# ---- 建库阶段 ----
docs = [
"我们的 Prometheus 部署在 monitor 命名空间，地址是 prometheus.monitor:9090。",
"Pod OOMKilled 的排查步骤：1) 看 lastState.terminated.reason；2) 确认 memory limit；"
"3) 查日志是否内存泄漏；4) 临时扩 limit 到 2 倍观察。",
"api-gateway 服务依赖 Redis 和 MySQL，Redis 在 cache 命名空间，MySQL 在 db 命名空间。",
# ... 更多文档
]
chunks = split_documents(docs, chunk_size=200, overlap=30)
store = VectorStore()
store.insert(chunks)
# ---- 查询阶段 ----
def rag_answer(question: str, llm_caller, top_k: int = 5) -> str:
# 1. 检索相关文档
retrieved = store.search(question, top_k=top_k)
context = "\n---\n".join(text for text, _ in retrieved)
# 顺手把相似度也打印出来，调试用
for text, score in retrieved:
print(f"[sim={score:.3f}] {text[:60]}...")
# 2. 拼到 prompt 里
prompt = (
f"你是 Kubernetes SRE 助手。请基于下面的知识库内容回答问题。"
f"如果知识库里没有相关信息，请直接说不知道，不要编造。\n\n"
f"【知识库】\n{context}\n\n"
f"【问题】{question}\n"
)
# 3. 调 LLM 生成
return llm_caller(prompt)
# 测试一下
if __name__ == "__main__":
answer = rag_answer("Prometheus 部署在哪？", lambda p: call_some_llm(p))
print(answer)
踩坑提示（RAG 是个深坑，挨个说）：
- chunk_size 别太大也别太小。太大召回粗糙，太小上下文断裂。运维文档我一般用 300-500 字符 + 50 overlap。代码类文档按函数切，别按字符切。
- 相似度阈值要调。
top_k=5不是绝对的，如果第 5 个相似度只有 0.3，那基本就是噪声，不如不塞。我一般会过滤掉相似度低于 0.5 的结果。 - embedding 模型和 LLM 不要混。
text-embedding-3-small配 GPT-4 效果好，配国产 LLM 未必；中文场景bge-large-zh通常更稳。 - 不要把整个 RAG 上下文当万能解。context 太长 LLM 会 “中间遗忘”，重要信息放前面，次要的放后面。
- 召回失败要兜底。Agent 看到检索结果跟问题对不上时，应该明确说 “知识库无相关信息”，而不是硬编。
多 Agent 协同
多 Agent 不是「多雇几个人」，而是「明确分工 + 权限隔离」。下面这张图把五个角色的消息流向画出来：只有 Executor 有写权限，Observer 只读，Diagnostician 连 K8s 都不碰，这样即使某个 Agent 被注入攻击，影响范围也可控。
flowchart LR
C[Coordinator 协调者] -->|派发任务| O[Observer 观测]
C -->|派发任务| D[Diagnostician 诊断]
C -->|派发任务| E[Executor 执行]
O -->|数据| D
D -->|审批请求| C
D -->|结论| C
E -->|执行结果| C
C -->|记录| A[Auditor 审计]单个 Agent 啥都干，听着美好，实际有两个问题：一是 prompt 太长导致推理质量下降；二是工具一多 LLM 就乱调。所以复杂任务通常拆成多个 Agent，每个负责一摊。
我之前做的一个故障自愈系统，角色这么分：
| 角色 | 职责 | 持有工具 |
|---|---|---|
| Coordinator（协调者） | 接收任务、拆分、分派、汇总 | 无工具，只调度 |
| Observer（观测 Agent） | 采集 Pod/Node/Event/指标 | get_pod, get_pod_logs, query_prometheus, list_events |
| Diagnostician（诊断 Agent） | 基于数据推理根因 | search_kb（RAG）, ask_human |
| Executor（执行 Agent） | 调 K8s API 执行修复 | restart_pod, scale_deployment, patch_deployment_resources |
| Auditor（审计 Agent） | 记录每次操作到 Event 和审计日志 | create_event, write_audit_log |
这种分工有个隐性好处：权限隔离。Executor 才有写权限，Observer 只读，Diagnostician 连 K8s 都不碰。万一某个 Agent 被注入攻击（prompt injection），影响范围可控。
多 Agent 之间通信通常用消息协议。我贴一个我们用的简化版：
# 消息协议：所有 Agent 之间通信都走这个格式
from dataclasses import dataclass, asdict
from datetime import datetime
from enum import Enum
from typing import Any
import uuid
class MessageType(str, Enum):
TASK = "task" # 任务派发
RESULT = "result" # 结果返回
QUERY = "query" # 询问（向其他 Agent 要信息）
APPROVAL = "approval" # 审批请求（高风险操作前）
@dataclass
class AgentMessage:
msg_id: str # 唯一 ID，便于追踪
from_agent: str # 发送方，如 "coordinator"
to_agent: str # 接收方，如 "executor"
msg_type: str # MessageType 之一
task_id: str # 关联的任务 ID，便于串链路
payload: dict[str, Any] # 实际内容，结构由 msg_type 决定
timestamp: str = "" # ISO8601 时间戳
def __post_init__(self):
if not self.timestamp:
self.timestamp = datetime.utcnow().isoformat() + "Z"
if not self.msg_id:
self.msg_id = str(uuid.uuid4())
# 协调者派任务给观测者
msg1 = AgentMessage(
msg_id="",
from_agent="coordinator",
to_agent="observer",
msg_type=MessageType.TASK.value,
task_id="INC-2025-11-04-001",
payload={
"action": "inspect_pod",
"namespace": "prod",
"pod_name": "api-gateway-7d8-x2",
},
)
# 观测者返回结果
msg2 = AgentMessage(
msg_id="",
from_agent="observer",
to_agent="diagnostician",
msg_type=MessageType.RESULT.value,
task_id="INC-2025-11-04-001",
payload={
"pod_status": {
"phase": "Running",
"restartCount": 14,
"lastState": {"terminated": {"reason": "OOMKilled"}},
},
"recent_logs": "...",
},
)
# 诊断者请求审批（因为要改 Deployment 资源）
msg3 = AgentMessage(
msg_id="",
from_agent="diagnostician",
to_agent="coordinator",
msg_type=MessageType.APPROVAL.value,
task_id="INC-2025-11-04-001",
payload={
"proposed_action": "patch_deployment_resources",
"args": {"namespace": "prod", "name": "api-gateway", "memory_limit": "512Mi"},
"reason": "OOMKilled caused by v1.5.0 batch size increase, need to bump memory limit",
"risk_level": "medium",
},
)
# 审计 Agent 把所有消息序列化存档
print(json.dumps(asdict(msg3), ensure_ascii=False, indent=2))
消息传递可以用内存 channel（单机）、Redis Stream / Kafka（分布式）。我个人建议一开始就用 Redis Stream，单机也用——后期从单机切分布式不用改代码。
踩坑提示：
- 必须设全局 task_id，否则多 Agent 链路一长就乱套，排查问题想哭。
- 协调者不要做业务。它的活就是拆任务 + 转发，让它去推理根因就本末倒置了。
- 高风险操作必须走 approval。我们线上规则是：所有写操作 risk_level=medium 以上必须人工确认，Agent 不能自己拍板。被坑过你就懂——LLM 一兴奋把你生产数据库的 Deployment 给 patch 了，没处哭。
- 死锁检测必须有。多 Agent 互相等结果会死锁，协调者要带超时，N 分钟没响应直接 fail 整个任务。
常用 Agent 框架
不写裸代码的话，框架能省不少事。但框架选错了改起来更痛苦。我用过的几个对比一下：
| 框架 | 出品方 | 核心卖点 | 适合场景 | 主要坑 |
|---|---|---|---|---|
| LangChain / LangGraph | LangChain | 链式抽象 + 工具集成 + LangGraph 支持状态图 | 通用 Agent、复杂流程编排 | 抽象层多、调试难、API 频繁 breaking change |
| AutoGen | Microsoft | 多 Agent 对话、GroupChat、代码执行 | 多 Agent 协作、研究原型 | 偏研究向，生产化要做不少封装 |
| CrewAI | CrewAI | 角色化（Crew + Agent + Task）、上手快 | 业务流程自动化、SOP 落地 | 复杂流程表达能力弱，灵活度不如 LangGraph |
| LlamaIndex | LlamaIndex | RAG 优先、数据索引能力强 | 知识库问答、文档检索 | Agent 能力相对弱，主打 RAG |
我的建议：
- 第一次写 Agent，别用框架。手撸一遍 ReAct 循环，你才知道框架在帮你做什么。直接上 LangChain 的人，多半到后面调不动。
- 生产环境推荐 LangGraph。它的状态图模型适合运维场景的 “多步骤 + 条件分支 + 人工审批” 流程，比 LangChain 的链式更可控。
- 快速原型用 CrewAI。三天能搭出个像样的多角色流程，适合给老板演示。
- RAG 重度场景配 LlamaIndex。它的索引、检索、reranking 模块比 LangChain 那套成熟。
- AutoGen 我个人不太推荐生产用，它的 GroupChat 模式好看但难控，token 烧得也凶。
补一句，运维场景有个特殊性——Agent 动的是生产环境。所以选框架时一定要看它对 “人工审批、dry-run、回滚、审计日志” 的支持。这几个 LangGraph 都能做（自己实现），CrewAI 需要绕一绕，AutoGen 基本得自己造。
总结
写到这里你应该看出来了，Agent 这东西不是银弹，它就是个 把 LLM 的推理能力接到外部工具上的工程框架。Planning、Function Calling、RAG、多 Agent，每一块都是工程活，没有哪块是 “接个 LLM 就能跑” 的。
我个人的踩坑心得就三条：一是工具描述和工具实现一样重要，值得花时间打磨；二是 Agent 的边界（什么时候停、什么不能做）比能力更重要；三是上线前先在测试环境跑一周，别信 demo——demo 永远是最好看的。
下一篇我们就开始用 client-go 把这些概念落到代码里，真正写一个会查集群、会调 LLM、会动手修 Pod 的运维 Agent。
自测题与动手练习
自测题（合上书能答出来，才算懂）：
- Agent 和「问一句答一句的 LLM 对话」本质区别是什么？为什么说「什么时候停」是关键设计点？
- 一个能用的 Agent 通常要具备哪 5 种能力？其中哪一个能力「高级 Agent 才有、简单 ReAct 不带」？
- Function Calling 本质是什么？为什么说它「被神化了」？
tool_call_id对不上会发生什么？ - ReAct 最常见的两个坑是什么？生产上一般怎么规避？
- 多 Agent 协同里，为什么「高风险操作必须走 approval」？权限隔离具体隔离了什么？
动手练习（建议真做一遍）：
- 跑通文末
run_agent最小循环，把max_turns调成一个很小的值（如 3）和一个很大的值，对比 Agent 在「查不清根因」时的行为差异。 - 给
restart_pod工具加dry_run守护：真正删除 Pod 前先打印「将要删除/ 」，确认危险操作先可模拟。 - 用文末
AgentMessage协议，把 Coordinator → Observer → Diagnostician(请求 approval) → Coordinator → Executor 的链路串起来，模拟一次完整故障自愈。
本章小结
- Agent 本质是「把 LLM 推理接到外部工具上的工程框架」：感知→决策→行动→反馈的闭环，而非一问一答。
- 五大能力里，规划决定 Agent 能不能干活，反思高级但易过度；工具描述要短而准，观测数据要预处理避免爆 context。
- Function Calling 是结构化输出协议，落地时务必处理好
tool_call_id对齐、argumentsjson.loads、设max_turns、工具异常回 JSON 不 raise。 - 多 Agent 靠角色分工 + 权限隔离 + 全局 task_id + 高风险 approval 来控风险；框架先手撸 ReAct，生产再上 LangGraph。
- 下一篇我们用 client-go 把这套概念落成代码，真正写一个会查集群、会调 LLM、会动手修 Pod 的运维 Agent。