手搓 Agent 学习路线（完整版）
这是学习大纲 + 全部代码合集。按章节查阅即可；代码已内嵌，无需再找独立
.py。 主线逻辑：让模型说话 → 让它记住 → 让它会做事 → 让它做得对 → 让它们协作 → 让它能上线。
目录
阶段 0 · 认知地基（概念先行）
| # | 章节 | 学什么 | 笔记 |
|---|---|---|---|
| 0.1 | Agent 本质 | LLM + 工具 + 记忆 + 循环；7 层架构 | Agent.md §一、二 |
| 0.2 | 推理模式 | ReAct / CoT / Plan&Execute / Self-Reflection | Agent.md §三 |
| 0.3 | 完整流程 | 用户输入到输出的整条链路 | Agent.md §十一 |
概念课，不写码。重读笔记即可。
阶段 1 · 调通模型（让模型说话）
1.1 环境准备
学什么：虚拟环境、密钥管理、gitignore、依赖清单。所有后续 step 都从这里拿 client 和 MODEL。 产出：llm.py
python
"""公共配置模块 —— 所有 step 都从这里拿 client 和 MODEL。"""
import os
from pathlib import Path
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv(Path(__file__).parent.parent / ".env")
MODEL = "deepseek-v4-flash"
client = OpenAI(
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com",
timeout=60, # 永远设超时，避免网络异常时无限挂起
)1.2 单次调用
学什么：messages 数组、role 角色、无状态认知 产出：step1_hello.py
python
"""
Step 1: 调通一次模型调用
目标：理解大模型 API 的本质 —— 输入 messages 数组，输出一段文本。
"""
from llm import MODEL, client
# 这就是一次模型调用的全部内容
response = client.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": "你是一个简洁的助手，回答不超过两句话。"},
{"role": "user", "content": "用一句话解释什么是 Agent。"},
],
)
# 模型的回答藏在这个位置
print(response.choices[0].message.content)
# 顺便看看这次花了多少 token（成本意识要从第一天建立）
print("\n--- 本次消耗 ---")
print(f"输入 token: {response.usage.prompt_tokens}")
print(f"输出 token: {response.usage.completion_tokens}")1.3 聊天循环
学什么：while 循环，故意暴露失忆问题产出：step2_loop.py
python
"""
Part 1.3: 聊天循环（故意不带记忆）
注意：这个程序有 bug（模型会「失忆」）。这是故意的 ——
跑起来看到问题，下一步才能明白「记忆」为什么需要。
"""
from llm import MODEL, client
print("聊天开始，输入 exit 退出。")
while True:
user_input = input("\n[我] ")
if user_input.strip() == "exit":
break
# 每次调用都是全新请求 —— 只发这一句，不带历史
response = client.chat.completions.create(
model=MODEL,
messages=[{"role": "user", "content": user_input}],
)
reply = response.choices[0].message.content
print(f"[AI] {reply}")阶段 2 · 单 Agent 骨架（让它会做事）
2.1 记忆
学什么：messages 累积修复失忆；assistant 回复也要存 产出：step3_memory.py
python
"""
Part 2.1: 记忆 —— 修复失忆
上一步 step2_loop 里，模型每轮只看到当前这句话，历史全丢。
这里用一个 messages 列表把全部对话累积起来，每次调用都带上。
"""
from llm import MODEL, client
# ============================================================
# 1. messages 里的三个角色 —— 想象成「发给演员的剧本」
# ============================================================
# 每次调用大模型，我们都发一份「完整剧本」过去，让它续写。
# 剧本里每句话都要标注「谁说的」，模型靠这个理解来龙去脉：
#
# role = "system" 导演 → 开场设定，全场只有一条，不参与对话但约束一切
# role = "user" 观众 → 用户说的话，每轮一条
# role = "assistant" 演员 → 模型自己说的话，每轮一条
#
# 还有隐藏的第 4 个角色 tool（工具结果），到 2.3 工具调用才登场。
#
# 为什么必须区分谁说的？
# 同样一句"你今年25岁"：
# 来自 user = 用户在陈述事实
# 来自 assistant = 模型在回答
# 混在一起模型就分不清哪些是自己说的、哪些是用户说的，逻辑会乱。
#
# 协议要求 user / assistant 必须交替出现，永远 user 在前、assistant 在后。
# ============================================================
# ============================================================
# 2. 消息内部的字段 —— 哪些是死的，哪些是活的
# ============================================================
# role → 定死的枚举，只能选 system / user / assistant / tool，不能自创
# content → 正文，必填。不一定是字符串，也可以是数组（多模态发图片）：
# {"role":"user","content":[{"type":"text","text":"..."},
# {"type":"image_url","image_url":{"url":"..."}}]}
# name → 可选，给角色起别名（如区分多个 assistant）
# tool_calls / tool_call_id → 可选，2.3 工具调用时才会出现
#
# 一句话：role 是死的，content 是活的，还有几个可选字段。
# ============================================================
# ============================================================
# 3. create() 调用 —— 只有 2 个必填，其他全是旋钮
# ============================================================
# model → 必填。用哪个模型
# messages → 必填。完整对话历史
# temperature → 随机性：0=每次几乎一样，1=天马行空（默认 0.7 左右）
# max_tokens → 回答最大长度
# top_p → 采样的另一个旋钮
# stream → 流式开关，2.2 的主角
# tools / tool_choice → 工具列表，2.3 的主角
# stop → 遇到这些词就停止生成
#
# 学会「哪些必填、哪些是旋钮」，读任何模型文档都不会懵。
# ============================================================
# 这个列表就是 Agent 的「短期记忆」，它活在内存里
messages = [
# 第一条永远是 system —— 整个对话的「剧本设定」
{"role": "system", "content": "你是一个简洁的助手，回答不超过两句话。"},
]
def chat(user_input):
# ① 把用户这句话追加进剧本（观众说话）
messages.append({"role": "user", "content": user_input})
# ② 把【完整剧本】发给演员，让它续写
response = client.chat.completions.create(
model=MODEL,
messages=messages,
)
reply = response.choices[0].message.content
# ③ 关键！把演员的话也追加进剧本。
# 否则下一轮演员看不到自己说过什么，会重复回答或前后矛盾。
messages.append({"role": "assistant", "content": reply})
return reply
print("聊天开始，输入 exit 退出。")
while True:
user_input = input("\n[我] ")
if user_input.strip() == "exit":
break
print(f"[AI] {chat(user_input)}")2.2 流式输出
学什么：stream=True，逐字蹦出，timeout 防卡死 产出：step4_stream.py
python
"""
Part 2.2: 流式输出
问题：非流式调用要等模型把话全部生成完才返回，几十秒里屏幕一片空白，
看起来像卡死了。
解法：stream=True。模型每生成一小块就推送给你，边收边打印。
总耗时不变，但第一个字几乎立刻出现。
"""
from llm import MODEL, client
messages = [
{"role": "system", "content": "你是一个简洁的助手，回答控制在两句话内。"},
]
def chat_stream(user_input):
messages.append({"role": "user", "content": user_input})
# ① 唯一的关键改动：加 stream=True
stream = client.chat.completions.create(
model=MODEL,
messages=messages,
stream=True, # ← 打开流式
)
print("[AI] ", end="", flush=True)
chunks = [] # 收集所有碎片，最后拼成完整回复存回历史
# ② 流式返回的是一个「迭代器」—— 每循环一次，就收到模型吐的一小块
for chunk in stream:
# ③ 关键差异：非流式用 message.content，流式用 delta.content
piece = chunk.choices[0].delta.content
if piece:
print(piece, end="", flush=True) # 立刻打印，不换行不缓冲
chunks.append(piece)
# ④ 把碎片拼成完整句子，存回记忆（和 step3 一样）
reply = "".join(chunks)
messages.append({"role": "assistant", "content": reply})
print()
print("聊天开始，输入 exit 退出。")
while True:
user_input = input("\n[我] ")
if user_input.strip() == "exit":
break
chat_stream(user_input)2.3 工具调用
学什么：工具说明书 JSON + 模型返回 tool_calls + 执行 + 结果喂回 产出：step5_tool.py
python
"""
TOOLS 格式固定的
你的messages 中带入工具类, 大模型回复 content 和 tool_calls 二选一
三要素（模型告诉你的）：
① 调哪个方法 → tc.function.name → "get_weather"
② 参数是什么 → tc.function.arguments → '{"city": "北京"}'（字符串！要 json.loads）
③ 点单编号 → tc.id → 用来把「点单」和「结果」配对
代码中执行 方法 拿到结果后，传给messages 再进行下一轮对话
"""
import json
from llm import MODEL, client
# ============================================================
# 1. 菜单：告诉模型你能提供什么服务
# ============================================================
# 模型看不到你的 Python 函数，只看到这份 JSON「菜单」。
# 格式字段是死的（name/description/parameters 不能自创），
# 但 description 要用心写 —— 写得好不好，决定模型会不会用对工具。
TOOLS = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "查询指定城市的当前天气",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名，如 北京"}
},
"required": ["city"],
},
},
},
{
"type": "function",
"function": {
"name": "get_stock",
"description": "查询指定股票代码的实时价格",
"parameters": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "股票代码，如 600519"}
},
"required": ["code"],
},
},
},
]
# ============================================================
# 2. 真实函数：模型看不到它，只能看菜单
# ============================================================
def get_weather(city: str) -> str:
fake_db = {"北京": "晴，25°C", "上海": "多云，28°C"}
return fake_db.get(city, f"没有{city}的天气数据")
def get_stock(code: str) -> str:
"""股票查询工具：模型点单后，代码查到模拟数据"""
fake_db = {"600519": "茅台 1680元 +2.3%", "000001": "平安银行 12.5元 -0.8%"}
return fake_db.get(code, f"没有{code}的股票数据")
# ============================================================
# 2.5 查表映射：模型点哪个名，就调哪个函数
# ============================================================
# 必须放在函数定义之后 —— Python 从上往下执行，此刻函数才存在
# （放早了会报 "未定义 get_weather"）
TOOL_FUNCTIONS = {
"get_weather": get_weather,
"get_stock": get_stock,
}
# ============================================================
# 主流程
# ============================================================
messages = [{"role": "user", "content": input("用户说: ")}] # 用户输入一句话，放进 messages
# 第一轮：带上菜单问模型
response = client.chat.completions.create(
model=MODEL, messages=messages, tools=TOOLS,
)
msg = response.choices[0].message
# 判断走哪条路：模型是「点单」还是「直接回答」？
if msg.tool_calls: # 情况一：点单（content 此时是空的）
tc = msg.tool_calls[0] # 拿第一个请求
messages.append(msg) # ① 把模型的「点单」存回历史（必须，否则协议报错）
args = json.loads(tc.function.arguments) # 参数是字符串，转成字典
# ② 按模型点的名，查表找到函数执行 —— 模型真正「选择」了方法
result = TOOL_FUNCTIONS[tc.function.name](**args)
messages.append({ # ③ 把结果喂回给模型
"role": "tool",
"tool_call_id": tc.id, # 用编号配对「点单」和「结果」
"content": result,
})
# 第二轮：带着结果，模型最终作答
final = client.chat.completions.create(
model=MODEL, messages=messages, tools=TOOLS,
)
print(final.choices[0].message.content)
else: # 情况二：没点单，直接回答（content 有内容）
print(msg.content)2.4 ReAct 主循环
学什么：思考→行动→观察→再思考，自己决定何时停。第一个完整 Agent产出：step6_ReAct.py
python
"""
Part 2.4: ReAct —— 第一个完整 Agent
一次对话的完整 ReAct 流程示例，模型可以调用工具，工具结果喂回模型继续决策，直到模型不再调用工具直接回答。
流程：问模型 → 工具? → 执行结果喂回 → 再问 → 直到不工具直接回答
本质与核心要素：
① 最少执行 1 次对话（for 保证第一轮必跑），最多 MAX_STEPS 次
② 每轮循环 = 一次 create() 对话，带的历史越来越厚（累积工具结果）
③ 循环 = 喂信息：模型全靠 messages 里累积的信息决策
④ 出口 = 模型不工具（它觉得信息够了）→ break
⑤ 兜底 = MAX_STEPS 耗尽 → else 报警（防死循环）
"""
import json
from llm import MODEL, client
# ========== 菜单：模型能点什么 ==========
TOOLS = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "查询指定城市的当前天气",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名，如 北京"}
},
"required": ["city"],
},
},
},
{
"type": "function",
"function": {
"name": "get_stock",
"description": "查询指定股票代码的实时价格",
"parameters": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "股票代码，如 600519"}
},
"required": ["code"],
},
},
},
]
# ========== 后厨：工具后真正干活的 ==========
def get_weather(city: str) -> str:
fake_db = {"北京": "晴，25°C", "上海": "多云，28°C", "深圳": "雷阵雨，30°C"}
return fake_db.get(city, f"没有{city}的天气数据")
def get_stock(code: str) -> str:
fake_db = {"600519": "茅台 1680元 +2.3%", "000001": "平安银行 12.5元 -0.8%"}
return fake_db.get(code, f"没有{code}的股票数据")
# ========== 查表：模型点的名 → 真函数 ==========
TOOL_FUNCTIONS = {
"get_weather": get_weather,
"get_stock": get_stock,
}
# ========== 开场：身份 + 需求 ==========
messages = [
{"role": "system", "content": "你是一个助手。能用工具就用工具，最后给出完整回答。"},
{"role": "user", "content": "北京、上海、深圳三地的天气对比一下，另外茅台股票多少钱？"},
]
MAX_STEPS = 10 # 安全阀：防模型无限工具
for step in range(MAX_STEPS):
print(f"\n=== 第 {step+1} 轮 ===")
# 带【完整历史 + 菜单】问模型
response = client.chat.completions.create(
model=MODEL, messages=messages, tools=TOOLS,
)
msg = response.choices[0].message
if msg.tool_calls: # 情况一：工具
messages.append(msg) # ① 工具存历史
for tc in msg.tool_calls: # ② 逐个执行（可一次点多个）
args = json.loads(tc.function.arguments)
result = TOOL_FUNCTIONS[tc.function.name](**args)
print(f" 工具: {tc.function.name}({args}) → {result}")
messages.append({ # ③ 结果喂回历史
"role": "tool",
"tool_call_id": tc.id,
"content": result,
})
continue # 回顶部再问（模型看到结果继续决策）
print(f" [AI] {msg.content}") # 情况二：不工具 → 直接回答
# 结构化打印 messages 列表
for m in messages:
print('\n---')
print(m)
break # 结束
else:
print(f"达到最大步数 {MAX_STEPS}，强制结束") # 循环耗尽才到这（防死循环）完成 2.4，你已经能手搓一个真正自主的 Agent。
阶段 3 · 让它做得对（质量关键）
3.1 系统提示词设计
学什么：角色 / 能力边界 / 行为规范 / 工具规则 / 输出格式 —— "Agent 的灵魂" 产出：step7_prompt.py
python
"""
Part 3.1: 系统提示词
是什么：messages 第一行 role="system"，告诉模型「你是谁、怎么回答」。
怎么写好（5 段，照着填）：
① 角色 你是X（例：你是营养师）
② 边界 只能做X，其他拒绝
③ 规范 必须/禁止（例：禁止编造）
④ 工具 什么时候用哪个工具
⑤ 格式 怎么回答（例：用表格、不超过50字）
system 和 user 冲突时：
user 能改模型「人设」（你说你是老师，它就变老师）
但 system 定的「行为规范」还在暗处生效
安全规则必须写 system，否则 user 一句"你现在是黑客"就能顶掉
对比实验：同一问题，烂 system vs 好 system
"""
from llm import MODEL, client
def ask(system, question):
r = client.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": question},
],
)
print(f"[系统] {system}")
print(f"[回答] {r.choices[0].message.content}\n")
print("═══ 对比 1：水果营养 ═══")
print("【烂 system】只有一句角色")
ask("你是营养师。", "苹果、香蕉、橙子三种水果的好处")
print("【好 system】5 段齐全")
GOOD1 = """你是资深营养师。 # ① 角色
只能回答营养健康相关问题，其他问题拒绝。 # ② 边界
数据要准确，不确定就说不确定，禁止编造。 # ③ 规范
必须用表格对比三种水果。 # ⑤ 格式
每种水果至少列 2 条好处，附一句总结。 # ⑤ 格式
"""
ask(GOOD1, "苹果、香蕉、橙子三种水果的好处")
print("═══ 对比 2：旅游推荐 ═══")
print("【烂 system】只有一句角色")
ask("你是旅游顾问。", "我想去旅游，推荐个地方")
print("【好 system】5 段齐全")
GOOD2 = """你是资深旅游顾问。 # ① 角色
你只推荐国内目的地。 # ② 边界
基于常识推荐，不编造不存在的地点。 # ③ 规范
必须给出：目的地、最佳季节、推荐理由、预算。 # ⑤ 格式
用列表输出，每项一行。 # ⑤ 格式
"""
ask(GOOD2, "我想去旅游，推荐个地方")3.2 任务拆解与规划
学什么：CoT 拆步、Plan&Execute 先计划后执行、Self-Reflection 自检 产出：step8_plan.py
python
"""
Part 3.2: 任务拆解与规划
先分清三个词（层次递进）：
拆解 = 把任务切成几块 例：做晚饭 → 买菜/洗切/炒菜/摆盘
计划 = 拆解 + 排顺序 + 定目标 例：先买菜 → 再炒 → 6点开饭
规划 ≈ 计划（口语常混用，不必纠结）
对应三种模式：
CoT（思维链） → 拆解：先列步骤再算
Plan&Execute → 计划：先列计划再执行
Self-Reflection → 复核：答完自检改错
拆解指令放哪？（优缺点 + 适用）
① 写 system → 每次生效/简单题啰嗦/Agent循环里最省事
② 写 user → 这次最认真/只生效一次/单次对话
③ 帮拆好 → 零发挥不错/失自主性/固定流程不能错
一句话选法：
想让模型自己拆（长期）→ system
想让模型这次拆清楚（单次）→ user
想让模型按我的步骤（控制）→ 帮拆好
注意：模型本来就会的简单题，不用拆，直接问更快。
实验：同一个数学题，三种放法对比。
"""
from llm import MODEL, client
Q = "小明有 3 个苹果，又买了 2 打（每打 12 个），吃了一半，还剩几个？"
def ask(system, user):
r = client.chat.completions.create(
model=MODEL,
messages=[{"role": "system", "content": system},
{"role": "user", "content": user}],
)
print(f"[回答] {r.choices[0].message.content}\n")
print("═══ ① 不拆解：直接问 ═══")
ask("你是一个助手。", Q)
print("═══ ② 写 system：让模型养成先拆习惯 ═══")
ask("你是一个助手。遇到计算题，先列出思考步骤，再计算。", Q)
print("═══ ③ 写 user：只引导这一次 ═══")
ask("你是一个助手。", "请先列出思考步骤，再算：" + Q)
print("═══ ④ 帮它拆好：你给具体步骤，它执行 ═══")
ask("你是一个助手。", "按这三步计算并回答：第一步 2打=12×2=24个；第二步 总数=3+24=27个；第三步 27的一半=？")3.3 多 Agent 协作
学什么：主控拆任务 → 研究 / 分析 / 报告 Agent 分工 产出：step9_multiagent.py
python
"""
Part 3.3: 多 Agent 协作
本质：多 Agent = 多次 create() 调用 + 各自不同的 system（角色）+ 结果传递。
没有复杂机制，就是把单 Agent 调用组织成"流水线"。
流程（接力式）：
研究 Agent → 查资料，产出素材
分析 Agent → 拿素材，加工成要点
报告 Agent → 拿要点，写成最终报告
对比上节：
3.2 拆解 = 提示词让一个模型自己拆步骤
3.3 多 Agent = 代码把任务拆成几块，每块发给不同角色的模型
"""
from llm import MODEL, client
def agent_call(system, user):
"""一个 Agent = 一次独立调用，带自己的角色(system)"""
r = client.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": user},
],
)
return r.choices[0].message.content
# ========== 接力流水线 ==========
# ① 研究 Agent：收集信息
research = agent_call(
"你是资深市场研究员。只收集事实，不评价，用要点列出。",
"收集智能手表这个产品的 3 个核心卖点和目标用户群体。",
)
print("【研究 Agent 产出】\n" + research + "\n")
# ② 分析 Agent：加工研究结果
analysis = agent_call(
"你是数据分析师。把输入内容整理成有逻辑的分析，指出优劣势。",
"以下是调研素材，请分析这款产品的优劣势：\n" + research,
)
print("【分析 Agent 产出】\n" + analysis + "\n")
# ③ 报告 Agent：写成最终报告
report = agent_call(
"你是报告撰写人。根据分析写成一份简洁的书面报告，分三个部分。",
"以下是产品分析，请写成最终报告：\n" + analysis,
)
print("【报告 Agent 产出】\n" + report)3.4 结构化输出
学什么：强制返回合法 JSON，解析不炸 产出：step10_json.py
python
"""
Part 3.4: 结构化输出 —— 让模型返回固定格式的 JSON
要解决的问题：模型默认输出自由文本，代码很难解析。
解法：① system 里规定输出 JSON 格式 → ② 解析失败就让它重来
核心代码：
json.loads(text) → 文本转成字典，取字段用 d["xxx"]
except json.JSONDecodeError → 模型没按格式输出，捕获后重试
对比实验：不规定格式 vs 规定格式 + 兜底重试。
"""
import json
from llm import MODEL, client
def ask_json(system, user, retries=3):
"""带格式要求 + 兜底重试的结构化请求"""
messages = [{"role": "system", "content": system},
{"role": "user", "content": user}]
for attempt in range(retries):
r = client.chat.completions.create(model=MODEL, messages=messages)
text = r.choices[0].message.content
try:
data = json.loads(text) # 解析成字典
print(f"✅ 第 {attempt+1} 次解析成功: {data}")
return data
except json.JSONDecodeError:
# 模型没按格式输出 → 把报错喂回去，让它重来
print(f"❌ 第 {attempt+1} 次解析失败，让模型重来")
messages.append({"role": "assistant", "content": text})
messages.append({"role": "user",
"content": "你刚才的输出不是合法 JSON，请只输出合法 JSON。"})
print("重试次数用尽，返回原始文本")
return text
print("═══ 对比 1：不规定格式 ═══")
print("【不规定】只让它提取")
r1 = client.chat.completions.create(
model=MODEL,
messages=[{"role": "user", "content": "从'北京今天晴，气温25度，风力3级'提取天气信息"}],
)
text1 = r1.choices[0].message.content
print(f"[模型输出] {text1}")
try:
json.loads(text1)
print("✅ 恰好是合法 JSON")
except json.JSONDecodeError:
print("❌ 不是合法 JSON，无法用 json.loads 解析")
print("\n═══ 对比 2：规定格式 + 兜底重试 ═══")
print("【规定】system 要求输出固定 JSON 结构")
GOOD = """你是一个信息提取器。必须只输出合法 JSON，不要任何多余文字。
格式：
{"weather": "晴/多云/雨", "temp": 数字, "wind": "风力描述"}"""
ask_json(GOOD, "从'北京今天晴，气温25度，风力3级'提取天气信息")3.5 上下文管理
学什么：滚动截断、摘要压缩、成本控制 产出：step11_context.py
python
"""
Part 3.5: 上下文管理
问题：messages 一直累积 → 越聊越贵，还可能超窗口上限。
解法：三种策略
① 滚动截断 只留最近 N 条（简单，老信息丢）
② 摘要压缩 老对话压成一段摘要（保留要点）
③ 滚动+摘要 保留 system + 摘要 + 最近 N 条（兼顾）
本节演示 ① 滚动截断 + ③ 滚动+摘要。
核心：system 永远保留（身份不能丢），砍掉中间最老的。
主流生产方案（大致的公认做法，各家细节不同）：
触发时机：不是按"轮数"，而是按【token 用量】——
当历史接近窗口上限的 50%-70% 时触发压缩（各家阈值不同）
压缩对象：最老的一批对话 → 用模型压成一段摘要
保留 system + 最近几轮（最近对话上下文最重要）
进阶变体：
· 摘要滚动更新（每次压缩在前次摘要上追加，不重压全部）
· 多级压缩（摘要之上再建"长期记忆"库）
· 触发成本控制（只在必要时压缩，因为压缩本身要花钱）
一句话：先进 Agent 用"token 用量触发 + 老对话摘要化"，不是固定轮数。
关键机制：压缩是【触发式】，不是每轮都压。
每次对话前检查 len(messages)：
没超阈值 → 不压，正常对话（零额外开销）
超了 → 压一次（最老一批进摘要），继续对话
流程是阶梯式：累积 → 超限 → 压一批 → 累积 → 再压 → ...
为什么不能每轮压：压缩=多一次模型调用（花钱），每轮压成本翻倍。
对比"全压成 1 条"（你的想法）：
进行中的对话 → 压老留新（主流），保留当下语境
会话收尾/存档 → 全压成 1 条，开新会话时当开场记忆
"""
import json
from llm import MODEL, client
# ========== ③ 滚动 + 摘要 的核心函数 ==========
def trim_messages(messages, max_len=6):
"""超长时裁剪：system 保留 + 老对话压成摘要 + 留最近几条"""
if len(messages) <= max_len:
return messages
system = messages[0] # system 永远留着
recent = messages[-4:] # 最近 4 条保留
old = messages[1:-4] # 中间最老的要压缩
# 用模型把老对话压成一句话摘要
r = client.chat.completions.create(
model=MODEL,
messages=[{"role": "system", "content": "把对话压缩成一句话摘要，保留关键信息。"},
{"role": "user", "content": json.dumps(old, ensure_ascii=False)}],
)
summary = r.choices[0].message.content
return [system, {"role": "user", "content": "【对话摘要】" + summary}] + recent
# ========== 演示：模拟 20 轮对话累积 ==========
messages = [{"role": "system", "content": "你是一个简洁的助手。"}]
# 模拟 20 轮用户提问（带具体信息，好让摘要有内容可总结）
questions = [
"我叫小明，今年25岁。", "我是程序员。",
"我住在上海。", "我喜欢吃苹果。",
"我家有一只猫。", "我的猫叫咪咪。",
"我最近在学Python。", "我会用Flask。",
"我下周去北京出差。", "打算住王府井附近。",
"我女朋友叫小红。", "她是一名老师。",
"我妈妈是医生。", "我爸爸是司机。",
"我喜欢打篮球。", "我打中锋位置。",
"我周末喜欢爬山。", "最近去过泰山。",
"我想学AI开发。", "正在学Agent。",
]
for i, q in enumerate(questions):
messages.append({"role": "user", "content": q})
messages.append({"role": "assistant", "content": f"好的，我记住了：{q.rstrip('。')}。"})
print("=" * 60)
print(f"压缩前：{len(messages)} 条（20 轮对话，每轮 2 条）")
print("=" * 60)
for i, m in enumerate(messages):
if i == 0:
fate = "保留"
elif i < len(messages) - 4:
fate = "→ 压成摘要"
else:
fate = "保留"
print(f" [{m['role']:9}] {m['content']:30} ← {fate}")
trimmed = trim_messages(messages, max_len=6)
print("\n" + "=" * 60)
print(f"压缩后：{len(trimmed)} 条（对比上面）")
print("=" * 60)
for m in trimmed:
is_summary = "摘要" in str(m.get("content", ""))
note = "← 摘要，替代了 37 条老对话" if is_summary else "← 原样保留"
print(f" [{m['role']:9}] {str(m['content'])[:60]:62} {note}")
# ========== 对比：不压缩 vs 压缩（看 token 差异） ==========
print("\n=== 不压缩发全部 vs 压缩后发 ===")
# 用 token 估算：简单按字符数看差异
raw_chars = sum(len(m["content"]) for m in messages)
trim_chars = sum(len(m["content"]) for m in trimmed)
print(f"不压缩：发 {len(messages)} 条，约 {raw_chars} 字符")
print(f"压缩后：发 {len(trimmed)} 条，约 {trim_chars} 字符")
print(f"节省：{len(messages) - len(trimmed)} 条消息，{raw_chars - trim_chars} 字符")
print(f"→ 每次调用都省这么多。20 轮对话后，每轮都为压缩前的全量付费！")阶段 4 · 工程化（让它能上线）
4.1 工程护栏
学什么：错误处理、步数上限、死循环检测、危险操作人工确认 产出：step12_guardrails.py
python
"""
Part 4.1: 工程护栏
Agent 会翻车（工具报错 / 无限循环 / 危险操作）。
护栏 = 保险丝，出问题时不崩、不烧钱、不做危险事。
四个护栏：
① try/except → 工具失败不崩溃
② MAX_STEPS → 防无限点单
③ 危险操作确认 → 删库这类，先停下问人
④ 工具失败喂回 → 报错给模型看，让它换办法
演示：模拟模型点单，看四个护栏怎么兜底。
"""
import json
# ========== 护栏 ③：危险工具名单 ==========
DANGEROUS_TOOLS = {"delete_database"}
# ========== 工具表 ==========
TOOL_FUNCTIONS = {
"get_weather": lambda city: "晴，25°C",
"delete_database": lambda db_name: "已删除！",
"broken_tool": lambda: (_ for _ in ()).throw(ValueError("模拟的工具报错")),
}
# ========== 模拟：模型一次点了 3 个单（一个正常、一个危险、一个会报错） ==========
fake_tool_calls = [
{"id": "c1", "name": "get_weather", "args": '{"city": "北京"}'},
{"id": "c2", "name": "delete_database", "args": '{"db_name": "prod"}'},
{"id": "c3", "name": "broken_tool", "args": "{}"},
]
print("模拟模型点单：查天气 / 删库 / 一个坏工具\n")
# ========== 逐个执行，每个都有护栏 ==========
for tc in fake_tool_calls:
name, args_str, id_ = tc["name"], tc["args"], tc["id"]
# 护栏 ③：危险操作，先停下问人
if name in DANGEROUS_TOOLS:
print(f" 护栏③ 拦截！{name} 是危险操作，需人工确认")
confirm = input(" 确认执行？[y/N] ").strip().lower()
if confirm != "y":
print(f" 用户拒绝 → 跳过 {name}（数据安全）\n")
continue
print(" 用户确认 → 继续")
# 护栏 ①④：工具执行包 try/except，失败不崩溃、报错喂回模型
try:
args = json.loads(args_str)
result = TOOL_FUNCTIONS[name](**args)
print(f" {name} → {result}")
except Exception as e:
result = f"工具失败: {e}"
print(f" 护栏① 捕获异常，不崩溃；{result}（护栏④：这条会喂回模型让它重试）")
print()
# 护栏 ②：MAX_STEPS（2.4 学过，这里只点一句）
print("护栏② MAX_STEPS = 循环设上限，防无限点单（见 step6_ReAct.py）")4.2 测试评估
学什么：单元 / 集成 / Evals / 回归测试怎么搭 产出：step13_eval.py
python
"""
Part 4.2: 测试评估 —— 专业完整版
═══════════════════════════════════════════════════
一、要解决的问题
═══════════════════════════════════════════════════
大模型是概率性系统：同样的输入，每次输出可能不同。
所以无法用普通代码测试（== 比较）验证它。测试评估要回答：
① 功能对不对 给定输入，输出是否满足预期
② 改了没改坏 修改 system/工具后，原有功能是否退化（回归测试）
③ 质量好不好 输出是否准确、无幻觉（Eval）
═══════════════════════════════════════════════════
二、三个核心要素
═══════════════════════════════════════════════════
测试用例集 → 固定的 (输入, 期望) 对 → CASES 列表
断言 → 判断输出是否满足期望 → 关键词/语义匹配
通过率 → 断言通过数 / 总数 → sum()/len()
═══════════════════════════════════════════════════
三、两种断言方式
═══════════════════════════════════════════════════
关键词匹配（本节）→ 期望词是否出现在回答里
适合：有明确输出词的任务（"回答必须含'晴'"）
实现：all(w in answer for w in expect)
语义匹配（进阶）→ 用另一个 LLM 判断回答质量
适合：开放任务、无法用关键词判定（"回答是否专业"）
实现：调 LLM 让模型打分/判断，返回 True/False
═══════════════════════════════════════════════════
四、回归测试的含义（本节的核心用途）
═══════════════════════════════════════════════════
基线：改动前跑一遍用例集 → 记录通过率
改动后：再跑一遍 → 通过率下降 = 引入了回归（行为退化）
为什么重要：提示词、模型版本、参数都会影响输出，
必须有可重复的验证手段，防止"改一处坏全局"。
═══════════════════════════════════════════════════
五、通过率低怎么处理 —— 三步走
═══════════════════════════════════════════════════
定位：哪个用例挂了？（看失败用例的实际输出）
归因：四类原因之一
① 测试写错了 期望值不符合真实数据 → 改测试，不是改代码
② 提示词问题 回答跑偏/缺信息 → 改 system
③ 工具/数据问题 工具返回错/没接真实数据 → 查工具和数据源
④ 模型波动 同一用例时过时挂 → 多次跑取均值/换模型
修复：针对原因改，改完重跑验证
关键：先怀疑测试，别急着改 Agent。
先确认"哪个是真相"，再决定改谁。
═══════════════════════════════════════════════════
六、"重跑直到通过"可以吗？
═══════════════════════════════════════════════════
偶发失败（时过时挂）→ 概率抖动，重跑合理
稳定失败（次次都挂）→ 确定性 bug，重跑无意义，必须修
正确姿势：
设达标阈值（如 ≥90%）+ 重跑上限（如 3 次）
达标 → 通过；超上限仍未达标 → 停下修，不硬跑
关键：重跑是为了过滤概率抖动，不是掩盖确定性 bug。
"""
import json
from llm import MODEL, client
# ============================================================
# 一、被测对象：Agent（这里用一个简单的问答函数代表）
# ============================================================
# 真实场景中，这里可以是完整的 ReAct 循环（step6 那种）。
def run_agent(system, user):
"""执行一次 Agent 调用，返回回答"""
r = client.chat.completions.create(
model=MODEL,
messages=[{"role": "system", "content": system},
{"role": "user", "content": user}],
)
return r.choices[0].message.content
# ============================================================
# 二、断言：判断"回答是否满足期望"
# ============================================================
def assert_keywords(answer, expected):
"""关键词断言：expected 里每个词都必须出现在 answer 中"""
return all(w in answer for w in expected)
def assert_json_has(answer, required_fields):
"""JSON 断言：answer 必须是合法 JSON，且包含要求的字段"""
try:
data = json.loads(answer)
except json.JSONDecodeError:
return False
return all(f in data for f in required_fields)
# ============================================================
# 三、测试框架：跑用例、统计通过率、输出报告
# ============================================================
def run_tests(system, cases, name="测试集"):
"""跑一组测试，返回通过率，打印报告"""
print(f"\n═══ {name} ═══")
passed = 0
for user, check, expected in cases:
answer = run_agent(system, user)
ok = check(answer, expected) # 用哪种断言由用例决定
passed += ok
mark = "✅" if ok else "❌"
print(f" {mark} {user}")
if not ok:
print(f" 期望 {expected}，实际: {answer[:60]}")
rate = passed / len(cases)
print(f" 通过 {passed}/{len(cases)} = {rate:.0%}")
return rate
# ============================================================
# 四、定义测试用例
# ============================================================
SYSTEM = """你是天气助手。
只能回答天气相关问题，其他问题礼貌拒绝。
回答必须包含城市名和天气描述。"""
CASES = [
# (输入, 断言函数, 期望值)
("北京今天天气如何？", assert_keywords, ["晴"]),
("上海今天天气如何？", assert_keywords, ["多云"]),
("帮我写首诗", assert_keywords, ["天气"]), # 验证能力边界：拒绝也会提及天气
]
# ============================================================
# 五、执行 + 回归测试流程
# ============================================================
print("=== 基线测试：记录当前通过率 ===")
baseline = run_tests(SYSTEM, CASES, "基线")
# —— 模拟"改动了 system"，看回归测试怎么发现问题 ——
print("\n=== 模拟：把 system 改坏（去掉边界约束）===")
BAD_SYSTEM = "你是天气助手，回答只报城市和天气。"
regression = run_tests(BAD_SYSTEM, CASES, "改动后")
print("\n=== 回归判断 ===")
print(f" 基线通过率: {baseline:.0%}")
print(f" 改动后通过率: {regression:.0%}")
if regression < baseline:
print(" ⚠️ 通过率下降 → 这次改动引入了回归，需要检查！")
else:
print(" ✅ 通过率未降 → 改动安全")毕业项目
主干学完（1.1 → 4.2）后合体为 my_agent.py：
带记忆 + 流式 + 多工具 + 系统提示词 + 任务拆解 + 多 Agent 协作
+ 结构化输出 + 上下文管理 + 工程护栏python
"""
毕业项目：my_agent —— 完整可用的旅游助手
一个文件合体 14 节能力：
工具（2.3） 查天气 / 算汇率
提示词（3.1） 角色 + 边界 + 格式
循环（2.4） ReAct 主循环
护栏（4.1） 危险操作确认 + MAX_STEPS
记忆（2.1） messages 累积多轮
流式（2.2） 逐字输出（含工具调用的分片拼接）
压缩（3.5） 超长自动摘要
结构化（3.4） 结果可存成 JSON
"""
import json
from llm import MODEL, client
# ========== 工具：菜单 + 真实函数 + 查表（2.3） ==========
TOOLS = [
{"type": "function", "function": {"name": "get_weather",
"description": "查询城市天气", "parameters": {"type": "object",
"properties": {"city": {"type": "string", "description": "城市名"}},
"required": ["city"]}}},
{"type": "function", "function": {"name": "get_exchange",
"description": "人民币换外币", "parameters": {"type": "object",
"properties": {"amount": {"type": "number"}, "currency": {"type": "string",
"description": "如 USD"}}, "required": ["amount", "currency"]}}},
]
def get_weather(city):
return {"北京": "晴25°C", "上海": "多云28°C", "三亚": "晴30°C", "成都": "小雨22°C"}.get(city, f"无{city}数据")
def get_exchange(amount, currency):
rate = {"USD": 7.1, "JPY": 0.048, "EUR": 7.7}.get(currency.upper())
return f"{amount}元 ≈ {amount*rate:.2f} {currency}" if rate else f"无{currency}汇率"
TOOL_FUNCTIONS = {"get_weather": get_weather, "get_exchange": get_exchange}
DANGEROUS = set() # 危险工具名单（4.1），当前没有，留作扩展
# ========== 系统提示词：角色 + 边界 + 格式（3.1） ==========
SYSTEM = """你是资深旅游助手。 # ① 角色
只回答旅行相关问题，其他问题礼貌拒绝。 # ② 边界
查天气/汇率必须用工具，禁止编造。 # ③ 规范
回答简洁，天气只报城市+天气，汇率只报结果。 # ⑤ 格式"""
# ========== 上下文压缩（3.5）：超长时老对话压成摘要 ==========
def msg_text(m):
"""只取消息的文本内容（兼容 dict 和模型对象，忽略 tool_calls）"""
if isinstance(m, dict):
return str(m.get("content", ""))
return str(m.content or "")
def compress(messages):
system, recent = messages[0], messages[-4:]
old = messages[1:-4]
if not old:
return messages
# 只把老对话的【文本】交给模型总结（tool_calls 对象不参与序列化）
r = client.chat.completions.create(model=MODEL, messages=[
{"role": "system", "content": "把对话压缩成一句话摘要，保留关键信息。"},
{"role": "user", "content": json.dumps([msg_text(m) for m in old], ensure_ascii=False)}])
return [system, {"role": "user", "content": "【摘要】" + r.choices[0].message.content}] + recent
# ========== 流式调用（2.2）：逐字打印 + 收集完整内容 ==========
# 难点：流式下工具调用是分片到达的，要按 index 把 id/name/arguments 拼起来
def stream_call(messages):
stream = client.chat.completions.create(
model=MODEL, messages=messages, tools=TOOLS, stream=True)
content_chunks, tool_calls = [], {} # tool_calls: index -> 拼接中的对象
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content: # 文本分片 → 打印 + 收集
print(delta.content, end="", flush=True)
content_chunks.append(delta.content)
if delta.tool_calls: # 工具分片 → 按 index 拼接
for tc in delta.tool_calls:
i = tc.index
tool_calls.setdefault(i, {"id": "", "name": "", "arguments": ""})
if tc.id:
tool_calls[i]["id"] = tc.id
if tc.function:
if tc.function.name:
tool_calls[i]["name"] += tc.function.name
if tc.function.arguments:
tool_calls[i]["arguments"] += tc.function.arguments
print() # 换行结束流式输出
content = "".join(content_chunks)
calls = [{"id": t["id"], "type": "function",
"function": {"name": t["name"], "arguments": t["arguments"]}}
for _, t in sorted(tool_calls.items())]
return content, calls or None
# ========== 主循环：ReAct + 护栏 + 记忆（2.4 + 4.1 + 2.1） ==========
def agent_loop(messages, user_input):
messages.append({"role": "user", "content": user_input})
if len(messages) > 8:
messages[:] = compress(messages) # 超长自动压缩（3.5）
for _ in range(5): # MAX_STEPS 护栏（4.1）
try:
print("[AI] ", end="")
content, calls = stream_call(messages) # 流式调用（2.2）
except Exception as e:
print(f"[护栏] 调用失败: {e}"); return
if calls:
messages.append({"role": "assistant", "content": content, "tool_calls": calls}) # ① 点单存历史
for c in calls: # ② 逐个执行
name, args_str = c["function"]["name"], c["function"]["arguments"]
args = json.loads(args_str)
if name in DANGEROUS: # 护栏：危险操作确认
if input(f"确认执行 {name}？[y/N] ").lower() != "y":
result, print(f"[护栏] 已拒绝 {name}")
else:
result = TOOL_FUNCTIONS[name](**args)
else:
try:
result = TOOL_FUNCTIONS[name](**args)
except Exception as e:
result = f"失败: {e}" # 护栏：工具报错喂回
messages.append({"role": "tool", "tool_call_id": c["id"], "content": str(result)}) # ③ 喂回
print(f" [工具] {name}({args}) → {result}")
continue
messages.append({"role": "assistant", "content": content}) # 最终回答存回记忆
return
# ========== 主程序：多轮对话（2.1 记忆） ==========
if __name__ == "__main__":
messages = [{"role": "system", "content": SYSTEM}]
print("旅游助手已就绪，输入 exit 退出。\n")
while True:
user_input = input("你: ").strip()
if user_input.lower() == "exit":
break
agent_loop(messages, user_input)阶段 5 · 进阶能力（学完主干再回来）
5.1 RAG
学什么：文档→切块→向量→检索→回答，私有知识库 产出：step14_rag.py（笔记：Agent.md §六）
python
"""
Part 5.1: RAG（检索增强生成）—— 给 Agent 接私有知识库
一句话理解：RAG = 自己数据的【语义搜索】 + 把搜到的喂给模型生成回答。
搜索部分（R 检索）：
普通搜索 → 按字面匹配（"北京好玩"必须含这些字）
RAG 搜索 → 按语义匹配（"故宫长城"也能搜到）
生成部分（G 生成）：
搜到片段 → 片段 + 问题喂给模型 → 模型组织成自然语言回答
流程：
离线：文档 → 切块 → 向量化 → 存"库"
在线：提问 → 向量化 → 检索相关片段 → 塞进上下文 → 回答
向量化 = 文字 → 一组数字（坐标），语义近则坐标近。
检索 = 找坐标最近的片段。
教学用简化向量：用字符的哈希生成一个"伪向量"，零依赖演示原理。
真实项目用 Embedding API 或 sentence-transformers。
接进 Agent：search 变成一个工具，Agent 自主决定"要不要查我的库"。
"""
from llm import MODEL, client
# ========== ① 向量化：文字 → 一组数字 ==========
def simple_embed(text, dim=50):
"""简化向量：按字符哈希生成 dim 维向量（教学用，语义近似）"""
vec = [0.0] * dim
for ch in text:
# 每个字符的哈希值，落到一个维度上累加
vec[ord(ch) % dim] += 1.0
return vec
# ========== ② 相似度：两个向量的余弦相似度 ==========
import math
def cosine(a, b):
dot = sum(x * y for x, y in zip(a, b))
na = math.sqrt(sum(x * x for x in a))
nb = math.sqrt(sum(x * x for x in b))
return dot / (na * nb) if na and nb else 0
# ========== ③ 知识库：我们的"私有文档" ==========
DOCS = [
"北京是中国的首都，有故宫、长城、天安门等著名景点。",
"上海是中国的经济中心，外滩和东方明珠是地标。",
"三亚位于海南岛，以热带海滩和度假著称。",
"成都以川菜和大熊猫基地闻名。",
"旅行建议：旺季提前订票，注意防晒和补水。",
"汇率：1美元约等于7.1元人民币。",
"签证：去日本需要办理签证，东南亚多数国家免签或落地签。",
]
# ========== ④ 切块 + 向量化 + 建库 ==========
# 真实场景文档很长，需要切成小块（300-500字）。这里文档已短，整条入库。
KB = [(doc, simple_embed(doc)) for doc in DOCS]
# ========== ⑤ 检索：找与问题最相关的片段 ==========
def search(query, top_k=2):
"""返回最相关的 top_k 个片段"""
q_vec = simple_embed(query)
ranked = sorted(KB, key=lambda x: cosine(q_vec, x[1]), reverse=True)
return [doc for doc, _ in ranked[:top_k]]
# ========== ⑥ 回答：检索片段 + 问题 → 模型 ==========
def rag_answer(question):
hits = search(question)
print(f" 检索到: {hits}")
# 把相关片段 + 问题一起发给模型（检索增强的"增强"在这）
r = client.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": "根据提供的资料回答，资料里没有就说不知道。"},
{"role": "user", "content": f"相关资料：\n{'、'.join(hits)}\n\n问题：{question}"},
],
)
return r.choices[0].message.content
# ========== 演示 ==========
questions = [
"北京有什么好玩的？",
"去日本需要签证吗？",
"美元汇率是多少？",
]
for q in questions:
print(f"\n[问] {q}")
print(f"[答] {rag_answer(q)}")5.2 记忆持久化
学什么：跨会话记住（文件 / 数据库） 产出：step15_persist.py（笔记：Agent.md §五）
python
"""
Part 5.2: 记忆持久化 —— 跨会话记住
问题：内存里的 messages 程序一关就没了。
解法：把记忆写进 JSON 文件，下次启动读回来。
三个时刻：
启动时 → 从文件读回历史（如果有）
对话中 → messages 照常累积（2.1）
退出时 → 把 messages 写回文件
对比内存记忆（2.1）：
内存 → 程序关了就丢
文件 → 关了再开还记得（本节）
数据库 → 量大/并发时才需要（进阶）
"""
import json
from pathlib import Path
from llm import MODEL, client
# 记忆文件路径（agent/ 子目录下，注意 .env 是上一层的写法）
MEMORY_FILE = Path(__file__).parent / "memory.json"
SYSTEM = "你是一个简洁的助手，记住用户告诉你的个人信息。"
def load_memory():
"""启动时：从文件读回历史"""
if MEMORY_FILE.exists():
return json.loads(MEMORY_FILE.read_text(encoding="utf-8"))
return [{"role": "system", "content": SYSTEM}]
def save_memory(messages):
"""退出时：把 messages 写回文件"""
MEMORY_FILE.write_text(json.dumps(messages, ensure_ascii=False, indent=2),
encoding="utf-8")
def chat(messages, user_input):
"""一次对话：累积历史 + 调模型（同 2.1）"""
messages.append({"role": "user", "content": user_input})
r = client.chat.completions.create(model=MODEL, messages=messages)
reply = r.choices[0].message.content
messages.append({"role": "assistant", "content": reply})
print(f"[AI] {reply}")
return reply
# ========== 主程序 ==========
if __name__ == "__main__":
# 启动时：读回历史（第二次运行能看到第一次的记忆）
messages = load_memory()
print(f"已从文件加载 {len(messages)} 条历史记忆\n")
while True:
user_input = input("你: ").strip()
if user_input.lower() == "exit":
save_memory(messages) # 退出时：写回文件
print(f"\n记忆已保存到 {MEMORY_FILE.name}，下次启动还记得。")
break
chat(messages, user_input)5.3 追踪观测
学什么：LangSmith / Langfuse，每步行为回放 产出：step16_trace.py（笔记：Agent.md §八）
python
"""
Part 5.3: 追踪观测 —— 记录 Agent 每一步行为
问题：Agent 是黑盒循环，点了几次单、为什么停、花了多少 token 都看不见。
解法：给每一步打日志（追踪 trace），出问题能回放。
专业方案（LangSmith / Langfuse）做的是同一件事，只是更完善
（可视化面板、云端存储）。本节手搓最朴素的版本：
每次模型调用 → 记 输入/输出/token
每个工具执行 → 记 工具名/耗时/结果
追踪的价值（你笔记 §八）：
"Agent 不能单步调试，追踪是定位问题的唯一方式。"
"""
import json
import time
from llm import MODEL, client
# ========== 追踪日志（收集每一步的记录） ==========
TRACE = []
def get_text(m):
"""取消息文本，兼容 dict 和模型对象"""
return m.get("content", "") if isinstance(m, dict) else (m.content or "")
def log_model_call(messages, response, duration):
"""记录一次模型调用"""
TRACE.append({
"类型": "模型调用",
"耗时": f"{duration:.1f}s",
"输入": [get_text(m)[:20] for m in messages[-2:]],
"输出": (response.choices[0].message.content or "")[:50],
"token": response.usage.total_tokens,
})
def log_tool_call(name, args, result, duration):
"""记录一次工具执行"""
TRACE.append({
"类型": "工具执行",
"工具": name,
"参数": args,
"结果": str(result)[:50],
"耗时": f"{duration:.2f}s",
})
# ========== 工具（简化） ==========
TOOLS = [{"type": "function", "function": {"name": "get_weather",
"description": "查询城市天气", "parameters": {"type": "object",
"properties": {"city": {"type": "string"}}, "required": ["city"]}}}]
TOOL_FUNCTIONS = {"get_weather": lambda city: "晴25°C"}
# ========== 带追踪的 ReAct 循环 ==========
def agent_loop(user_input):
messages = [{"role": "system", "content": "你是天气助手，用工具查天气。"},
{"role": "user", "content": user_input}]
for _ in range(5):
t0 = time.time()
r = client.chat.completions.create(model=MODEL, messages=messages, tools=TOOLS)
log_model_call(messages, r, time.time() - t0) # ← 追踪点 1
msg = r.choices[0].message
if msg.tool_calls:
messages.append(msg)
for tc in msg.tool_calls:
name, args = tc.function.name, json.loads(tc.function.arguments)
t1 = time.time()
result = TOOL_FUNCTIONS[name](**args)
log_tool_call(name, args, result, time.time() - t1) # ← 追踪点 2
messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})
continue
return msg.content
# ========== 演示 ==========
print("=== 运行 Agent ===")
answer = agent_loop("北京天气怎么样？")
print(f"回答: {answer}\n")
print("=== 追踪回放：看 Agent 到底做了什么 ===")
for i, entry in enumerate(TRACE, 1):
print(f" 步骤{i}: {entry}")当前进度
阶段0 ✅ 1.1-1.3 ✅ 2.1-2.4 ✅ 3.1-3.5 ✅ 4.1-4.2 ✅
毕业项目完成 ✅（my_agent.py）
阶段5 进阶全部完成 ✅（step14_rag / step15_persist / step16_trace）
全套学习路线完成 🎉学习方式
- 每个 Part：先读学什么，再读本节完整代码，对照注释理解关键点
- 需要动手跑时：把对应代码块复制成
.py即可（依赖llm.py的章节一并复制公共配置）
概念 → 章节速查
| 概念 | 章节 | 代码 |
|---|---|---|
| 工具调用（Function Calling） | 2.3 | step5_tool.py |
| ReAct / CoT / Plan&Execute / Self-Reflection | 0.2 + 3.2 | step6_ReAct.py / step8_plan.py |
| 记忆（短期 / 长期） | 2.1 + 5.2 | step3_memory.py / step15_persist.py |
| RAG | 5.1 | step14_rag.py |
| 多 Agent | 3.3 | step9_multiagent.py |
| 系统提示词 | 3.1 | step7_prompt.py |
| 上下文超长 / 无限循环 / 危险操作 | 3.5 + 4.1 | step11_context.py / step12_guardrails.py |
| 测试评估 / 追踪 | 4.2 + 5.3 | step13_eval.py / step16_trace.py |
| 毕业合体 | — | my_agent.py |