10. 综合练习：搭一个能干活的 Agent
这是什么：本章要交付的是一个命令行 Agent——输入一句话任务，它自己查资料、记笔记、存偏好，最后给出答案。
为什么用：前面几章的零件都是散的；把它们拼成一个能跑的工具，才能看清一个完整 Agent 该有哪些层。
如何用：按「工具层 + 记忆层 + 主循环」三份文件照着写，跑一遍验收清单。
10.1 功能设计
输入：一句话任务
输出：最终回答 + 一路执行过程
能力：
· 查「知识库」——从本地 JSON 文件里检索
· 记「笔记」——把要点写进文件
· 记住「偏好」——跨会话存下来
· 危险操作需人工确认10.2 项目结构
agent-lab/
├── tools.py # 工具函数 + 工具说明书
├── memory.py # 跨会话记忆
├── agent.py # Agent 循环
└── kb.json # 知识库（自己准备几条）10.3 工具集 tools.py
"""工具层：真正干活的函数 + 给模型看的说明书"""
import json
import re
from pathlib import Path
KB_FILE = Path("kb.json")
NOTES_FILE = Path("notes.md")
PREFS_FILE = Path("prefs.json")
# ---------- 真正干活的函数 ----------
def kb_search(keyword: str) -> str:
"""在知识库里找含关键词的条目"""
if not KB_FILE.exists():
return "知识库文件不存在，请先创建 kb.json"
data = json.loads(KB_FILE.read_text("utf-8"))
hits = [f"【{d['title']}】{d['content']}" for d in data if keyword in d["content"] or keyword in d["title"]]
return f"共 {len(hits)} 条：\n" + "\n".join(hits) if hits else f"没找到含「{keyword}」的条目"
def write_note(content: str) -> str:
"""把内容追加到笔记文件"""
with NOTES_FILE.open("a", encoding="utf-8") as f:
f.write(f"- {content}\n")
return f"已记下：{content}"
def read_notes() -> str:
"""读出全部笔记"""
if not NOTES_FILE.exists():
return "还没有任何笔记"
return NOTES_FILE.read_text("utf-8")
# ---------- 给模型看的说明书 ----------
TOOLS = [
{
"type": "function",
"function": {
"name": "kb_search",
"description": "在本地知识库 kb.json 中检索条目。传入关键词（如 '报销'），"
"返回所有命中的条目。找不到时会明确说没有。",
"parameters": {
"type": "object",
"properties": {
"keyword": {"type": "string", "description": "检索关键词，中文"}
},
"required": ["keyword"]
}
}
},
{
"type": "function",
"function": {
"name": "write_note",
"description": "把一条笔记追加写入 notes.md。用于记录用户明确要求保留的信息。",
"parameters": {
"type": "object",
"properties": {
"content": {"type": "string", "description": "笔记正文，一句话"}
},
"required": ["content"]
}
}
},
{
"type": "function",
"function": {
"name": "read_notes",
"description": "读出 notes.md 里已有的全部笔记。不需要参数。",
"parameters": {"type": "object", "properties": {}}
}
},
]
# ---------- 分发器 ----------
FUNCTIONS = {
"kb_search": kb_search,
"write_note": write_note,
"read_notes": read_notes,
}
def run_tool(name: str, args: dict) -> str:
if name not in FUNCTIONS:
return f"没有名为 {name} 的工具"
try:
return FUNCTIONS[name](**args)
except Exception as e:
return f"工具执行出错：{e}"
10.4 记忆 memory.py
"""跨会话记忆：把用户的偏好存下来，下次还在"""
import json
from pathlib import Path
PREFS_FILE = Path("prefs.json")
def load_prefs() -> list[str]:
if not PREFS_FILE.exists():
return []
try:
return json.loads(PREFS_FILE.read_text("utf-8"))
except json.JSONDecodeError:
return []
def save_pref(text: str):
prefs = load_prefs()
if text not in prefs:
prefs.append(text)
PREFS_FILE.write_text(json.dumps(prefs, ensure_ascii=False, indent=2), "utf-8")
return f"已记住：{text}"
10.5 Agent 循环 agent.py
"""Agent 主循环：思考 → 行动 → 观察，转到完成或超限"""
import json
import os
from openai import OpenAI
from memory import load_prefs, save_pref
from tools import TOOLS, run_tool
client = OpenAI(
api_key=os.environ["LLM_API_KEY"],
base_url=os.environ.get("LLM_BASE_URL", "https://api.openai.com/v1"),
)
CHAT_MODEL = os.environ.get("CHAT_MODEL", "gpt-4o-mini")
MAX_STEPS = 8
def compress(messages, keep_recent=6):
"""老消息压成摘要，最近几条保留原文"""
if len(messages) <= keep_recent:
return messages
old, recent = messages[:-keep_recent], messages[-keep_recent:]
summary = client.chat.completions.create(
model=CHAT_MODEL,
messages=[{"role": "user", "content":
"把下面的过程总结成一段话，保留关键结论与已执行动作，200 字以内：\n"
+ "\n".join(str(m.get("content", "")) for m in old)}],
).choices[0].message.content
return [{"role": "system", "content": f"【此前过程摘要】{summary}"}] + recent
def run_agent(question: str) -> str:
prefs = load_prefs()
system = "你是一个严谨的助手。优先使用工具查证，不要凭空编造。"
if prefs:
system += "\n已知用户偏好：" + "；".join(prefs)
messages = [
{"role": "system", "content": system},
{"role": "user", "content": question},
]
for step in range(MAX_STEPS):
resp = client.chat.completions.create(
model=CHAT_MODEL,
messages=compress(messages),
tools=TOOLS,
)
msg = resp.choices[0].message
if not msg.tool_calls:
return msg.content
messages.append(msg)
for call in msg.tool_calls:
args = json.loads(call.function.arguments)
name = call.function.name
# 记忆类操作直接执行
if name == "remember":
result = save_pref(args.get("pref", ""))
else:
result = run_tool(name, args)
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": result,
})
print(f"[第 {step + 1} 步] {name}({args}) → {result[:60]}")
return f"超过 {MAX_STEPS} 步仍未完成，请把问题拆小一点再试。"
if __name__ == "__main__":
q = " ".join(os.sys.argv[1:]) or input("任务：")
print(run_agent(q))
10.6 运行与验收
# 准备知识库
echo '[{"title":"报销标准","content":"差旅住宿每晚上限 500 元，需凭发票报销。"}]' > kb.json
# 提问
python agent.py "差旅住宿能报多少？帮我记下来"
你会看到：
[第 1 步] kb_search({'keyword': '差旅住宿'}) → 共 1 条：【报销标准】差旅住宿每晚上限 500 元…
[第 2 步] write_note({'content': '差旅住宿上限 500 元'}) → 已记下：差旅住宿上限 500 元
差旅住宿每晚可以报销到 500 元，需要凭发票。这条我已经记进笔记了。验收清单：
- 问知识库里有的问题，答案来自检索结果而不是编的
- 问知识库里没有的问题，它会说「没找到」
- 跑一次后
notes.md里真的多了一行 - 换个终端再问，它还记得你之前的偏好
10.7 常见翻车现场
| 报错 / 现象 | 原因 | 修法 |
|---|---|---|
FileNotFoundError: kb.json |
知识库没建 | 按 6.6 建一个 |
工具参数 keyword 传空 |
Schema 里没标必填 | 保留 required |
| Agent 反复查同一个词 | 结果没告诉它「已经查过」 | 在返回里加提示 |
| 笔记写成了乱码 | 没指定编码 | encoding="utf-8" |
| 上下文超长报错 | 工具返回太肥 | 裁剪返回 + 用 compress() |
| 偏好没记住 | prefs.json 写错目录 |
确认工作目录一致 |
10.8 动手改
- 给 Agent 加一个「删除笔记」工具，并给它加人工确认——对照第 8 章的刹车设计
- 把知识库换成 20 条以上的真实文档，测测它还能不能找准
- 加上「多关键词检索」：让
kb_search支持一次传多个词 - 引入 RAG 教程 的向量检索替换
kb_search的关键词匹配，体会两者的差别
10.9 小结
- 一个完整 Agent = 工具层 + 记忆层 + 主循环，三层分开写才清楚
- 主循环的核心还是那三步，压缩历史只是加了个
compress() - 工具说明书（
TOOLS）是 Agent 能力的上限，值得反复打磨 - 危险工具在执行前要人工确认
- 跨会话记忆用文件就够，量大再上向量库
Agent 教程完成！接下来可以看看现成的 Agent 框架长什么样：Hermes、opencode、OpenClaw 🐸