高级 Agent 工程师
口袋手记

拆解三原则：

常见反模式：

为什么用状态机？线性链无法处理重试、回退、条件跳转。状态机天然支持这些。

class AgentState(Enum):
    IDLE = "idle"
    INTENT_RECOGNITION = "intent_recognition"
    INFO_EXTRACTION = "info_extraction"
    ORDER_QUERY = "order_query"
    HUMAN_REVIEW = "human_review"    # 低置信度 → 转人工
    COMPLETED = "completed"
    FAILED = "failed"

# 状态转移规则示例
INTENT_RECOGNITION → INFO_EXTRACTION  # confidence > 0.8
INTENT_RECOGNITION → HUMAN_REVIEW      # confidence ≤ 0.8
INFO_EXTRACTION    → INFO_EXTRACTION   # 自循环=追问，最多3次
INFO_EXTRACTION    → HUMAN_REVIEW      # 追问3次仍无结果

验收条件量化示例：

criteria = [
    {"name": "confidence_threshold", "rule": "gt", "value": 0.8},
    {"name": "intent_in_whitelist", "rule": "in", "values": ["退货","换货","投诉","咨询"]},
    {"name": "has_reasoning", "rule": "not_null", "severity": "warning"}  # warning 不阻断
]

LangChain 的问题：过度抽象、黑盒调试、版本地狱、性能开销。

正确姿势：

• 直接调用 LLM API（openai / anthropic 原生 SDK）
• 自己写轻量编排器（< 200 行代码）
• 只在确实需要时引入框架（如 LlamaIndex 做复杂 RAG）
• 框架是库，不是架构——不要让框架定义你的系统边界

Chunking 策略选型：

检索策略——混合检索（推荐）：

RAG 常见失败模式：

上下文预算分配（128K Token 示例）：

System Prompt    ████  5K  (4%)     工具定义        ████  5K  (4%)
长期记忆(检索)   ████████  10K (8%)     RAG检索结果    ████████████  15K (12%)
对话历史         ████████████████  20K (16%)  当前用户输入    ██  2K  (2%)
预留(工具调用等) ████████████████████████████████████████  71K (55%)

• Step1 Schema：description 是否清晰？parameters 有约束？有 examples？
• Step2 返回：格式统一(JSON)？错误含 hint？内容是否过长？
• Step3 上下文：工具定义未被截断？多个工具名称相似混淆？
• Step4 LLM：模型支持 Function Calling？复杂调用需更强模型？
• Step5 业务：工具本身有 bug？超时正确处理？副作用正确标注？

Tracing 核心数据结构：

Trace (一次完整执行)
├── trace_id, session_id, user_input
├── total_tokens, total_cost, duration_ms
└── Spans[] (树形结构)
    ├── Span: "intent_recognition" (type: llm, tokens: 500, duration: 320ms)
    ├── Span: "tool:search_orders" (type: tool, duration: 150ms)
    │   └── Span: "db_query" (type: internal, duration: 45ms)
    └── Span: "generate_response" (type: llm, tokens: 800, duration: 680ms)

每个 Span 记录：span_id、parent_id、name、type(llm/tool/retriever/chain)、start/end time、input/output、tokens、status(success/error)、error信息

核心原理：用一个更强的LLM（Judge）对Agent的输出进行多维度打分。这是目前最实用的自动化评测方法，但需要精心设计Prompt和评分标准。

Judge Prompt 模板（可直接使用）：

你是一个严格的评测专家。请根据以下标准对【AI回复】进行评分。

【用户问题】
{user_question}

【参考上下文】（来自知识库检索结果）
{retrieved_context}

【AI回复】
{ai_response}

评分维度（每项1-5分）：
1. 准确性：回复内容是否与参考上下文一致？有无编造事实？
2. 完整性：是否完整回答了用户问题？有无遗漏关键信息？
3. 相关性：回复是否紧扣用户问题？有无跑题？
4. 简洁性：回复是否简洁明了？有无冗余废话？
5. 安全性：回复是否安全合规？有无不当内容？

评分规则：
- 5分：完美，无可挑剔
- 4分：良好，有小瑕疵但不影响使用
- 3分：及格，有明显问题需要改进
- 2分：较差，存在严重错误
- 1分：完全不可用

请输出JSON格式：
{
  "accuracy": {"score": 5, "reason": "..."},
  "completeness": {"score": 4, "reason": "..."},
  "relevance": {"score": 5, "reason": "..."},
  "conciseness": {"score": 4, "reason": "..."},
  "safety": {"score": 5, "reason": "..."},
  "overall": 4.6,
  "hallucination_detected": false,
  "hallucination_detail": "..."
}

Judge 模型选择：

• 高精度场景：GPT-4o / Claude 3.5 Sonnet（成本高但评分最准）
• 性价比场景：Qwen2.5-72B / DeepSeek-V3（开源模型，评分一致性约85%）
• 批量评测：用GPT-4o评100条建立"黄金标准"，再用小模型批量评分

数据集分层设计（建议总量 200-500 条）：

每条评测数据包含：

• user_input：用户原始输入
• expected_tools：期望调用的工具列表（可选）
• expected_output_pattern：期望输出中应包含的关键信息（正则或关键词）
• forbidden_patterns：输出中不应出现的内容（如"我无法"、"作为AI"等推脱话术）
• context：对话上下文（多轮对话场景）
• difficulty：难度标签（easy/medium/hard）

设计目标：把评测集成到CI/CD流程中，每次代码/Prompt/模型变更自动触发评测，不通过则阻止部署。

流水线架构：

Git Push / PR Merge
    │
    ▼
┌─────────────────────────────────────────────┐
│  Stage 1: 快速检查（< 2min）                 │
│  ├── 格式校验：输出JSON格式是否正确？          │
│  ├── 安全扫描：输出是否包含敏感词？            │
│  └── 工具调用：必调工具是否被调用？            │
│      → 不通过则直接失败，不进入下一阶段        │
└─────────────────────────────────────────────┘
    │ 通过
    ▼
┌─────────────────────────────────────────────┐
│  Stage 2: 核心评测（< 10min）                │
│  ├── 跑50条核心评测集（基础功能+边界情况）     │
│  ├── LLM-as-Judge 自动评分                   │
│  └── 对比上一版本分数（回归检测）              │
│      → 总分下降>5% 或任一维度下降>10% → 失败  │
└─────────────────────────────────────────────┘
    │ 通过
    ▼
┌─────────────────────────────────────────────┐
│  Stage 3: 完整评测（< 30min，可选）          │
│  ├── 跑全部200-500条评测集                    │
│  ├── 生成详细评测报告                         │
│  └── 自动标注bad case供人工review            │
└─────────────────────────────────────────────┘
    │ 通过
    ▼
  部署到生产环境

关键设计决策：

• 分层执行：快速检查→核心评测→完整评测，越往后越慢但越全面。PR时跑Stage 1+2（<12min），发版前跑Stage 3
• 回归检测：每次评测结果与上一版本对比，分数下降超过阈值自动告警。这是防止"修了一个bug引入三个新bug"的关键
• Bad Case 自动归档：评测失败的case自动打标签存入数据集，下次评测必须通过
• 工具链推荐：GitHub Actions/Jenkins 触发 → Python 脚本跑评测 → LangFuse/LangSmith 存储结果 → 飞书/钉钉/Slack 通知

幂等执行流程：

1. 每个请求带唯一 idempotency_key
2. 先查缓存：已完成 → 直接返回缓存结果
3. 再查运行锁：进行中 → 返回"请稍后"
4. 获取分布式锁（Redis SETNX, TTL 5min）
5. 执行业务逻辑
6. 缓存结果（TTL 1h）+ 释放锁

审核节点关键设计：

• 超时自动升级：审核超时 → 自动升级到更高级别
• 审批链可配置：支持多级审批、多人会签
• 决策可追溯：所有审批记录持久化，含审批人、时间、意见

必读论文：

• ★★★★★ Attention Is All You Need (2017) — Transformer原理解析
• ★★★★★ ReAct: Synergizing Reasoning and Acting (2023) — Agent推理+行动范式
• ★★★★★ DSPy: Compiling Declarative LM Calls (2024) — 编程式Prompt优化
• ★★★★ RAG原始论文 (2020) + Toolformer (2023) + GPT-4 Tech Report (2023)

必读源码：

• ★★★★★ LangGraph — 状态图驱动的Agent编排
• ★★★★★ OpenHands (OpenDevin) — 生产级Agent架构参考
• ★★★★★ MCP (Anthropic) — 工具调用协议标准

pdb++ 启用方式：export PYTHONBREAKPOINT=pdbpp.set_trace 后，所有 breakpoint() 自动使用 pdb++。
pudb 启用：breakpoint() 处改为 import pudb; pudb.set_trace() 或设置环境变量。

from loguru import logger

logger.remove()
logger.add("app_{time}.log", rotation="500 MB", retention="10 days")
logger.add(sys.stderr, format="{time} | {level} | {name}:{line} | {message}")

@logger.catch
def risky_function():
    1 / 0

logger.info("用户 {} 登录成功", user_id)
logger.bind(user_id=123).info("订单创建")  # 结构化上下文

• 优势：开箱即用的彩色输出、支持装饰器 @logger.catch 自动捕获异常、日志按大小/时间自动轮转、结构化日志原生支持
• @logger.catch 装饰器：自动捕获被装饰函数的异常并记录完整 traceback，适合快速排查未预期异常
• .bind() 方法：绑定上下文信息到日志记录器，后续所有日志自动附带该上下文

import logging
import json
import time

class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_obj = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "logger": record.name,
            "file": record.filename,
            "line": record.lineno,
            "message": record.getMessage(),
            "module": record.module,
            "request_id": getattr(record, "request_id", ""),
        }
        if record.exc_info:
            log_obj["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_obj, ensure_ascii=False)

关键字段建议：timestamp（时间戳）、level（级别）、service（服务名）、trace_id（链路追踪ID）、request_id（请求ID）、user_id（用户ID）、error_code（错误码）、duration_ms（耗时）。这些字段让日志从"文本记录"升级为"可查询数据源"。

• Evaluate Expression（Alt+F8）：在断点处执行任意 Python 表达式，修改变量值后继续运行
• 条件断点（Ctrl+Shift+F8）：右键断点→设置条件，如 x > 100 and status == "error"
• 异常断点：Run → View Breakpoints → Python Exception Breakpoint，在任何异常抛出时暂停
• Watch 表达式：监控复杂表达式：[item.price for item in cart if item.in_stock]
• Drop Frame：在调用栈中右键→"Drop Frame"，回退到当前方法的调用点重新执行
• Attach to Process：Run → Attach to Process，附加到正在运行的 Python 进程
• Data View 复制：在 Variables 面板右键变量→"Copy Value"，快速导出数据到测试代码
• 多线程调试：Frames 面板切换线程，查看各线程调用栈和变量状态

pip install line_profiler

# 使用装饰器
from line_profiler import LineProfiler
lp = LineProfiler()

@lp
def slow_function():
    # ... 代码 ...

slow_function()
lp.print_stats()

# 命令行方式
kernprof -l -v script.py  # -l: 逐行分析, -v: 立即显示结果

line_profiler 输出每一行的执行次数、总耗时和每次平均耗时，适合精确定位函数内部的瓶颈行。一般先用 cProfile 找到慢函数，再用 line_profiler 聚焦优化。

pip install py-spy

py-spy top --pid <PID>                # 实时 top，类似 htop
py-spy record -o flame.svg --pid <PID>  # 生成火焰图，可在浏览器查看
py-spy dump --pid <PID>                # 打印当前所有线程的调用栈

• 核心优势：不需要修改代码、不需要重启进程、对目标进程性能影响极小（采样而非侵入）
• 排查场景：生产环境 CPU 突然飙高、服务响应变慢、死循环排查
• 火焰图：SVG 交互式火焰图，鼠标悬停查看函数名和耗时占比，从下往上看调用链

pip install objgraph

import objgraph

objgraph.show_most_common_types(limit=20)
objgraph.show_growth(limit=10)  # 哪些类型对象在增长
objgraph.show_backrefs(leaked_object, max_depth=3, filename="leak.png")

经典内存泄漏原因：①循环引用导致引用计数无法清零 ②全局列表/cache 无限增长 ③未关闭的文件句柄/数据库连接 ④C 扩展未正确管理内存 ⑤闭包引用了大对象。排查流程：tracemalloc 找增长 → objgraph 看引用链 → memory_profiler 定位代码行。

pip install debugpy
python -m debugpy --listen 0.0.0.0:5678 --wait-for-client manage.py runserver

容器端口 5678 暴露出去，VS Code attach 即可调试容器内代码。--wait-for-client 让程序等待调试器连接后再启动。

kubectl debug <pod> -it --image=python:3.12 --target=<container> -- bash
# 进入容器后可以 py-spy dump，调试进程

# 启用 asyncio 调试模式（开发环境）
import asyncio
asyncio.run(main(), debug=True)  # 检测慢回调、未 await 的协程

# 环境变量方式
PYTHONASYNCIODEBUG=1 python script.py

# 使用 aiomonitor 实时监控
pip install aiomonitor
import aiomonitor
with aiomonitor.start_monitor(loop):
    loop.run_forever()  # 可 telnet 连接查看所有 Task 状态

# asyncio 调试检测的问题：
# - 超过 100ms 的协程回调（可能阻塞事件循环）
# - 协程创建了但从未被 await（资源泄漏）

异步调试痛点：pdb 的 s(tep) 无法进入 async 函数内部，需要在 await 处打断点。使用 asyncio.get_event_loop().slow_callback_duration = 0.1 可检测阻塞事件循环的耗时回调。

• pytest + pdb：pytest --pdb 测试失败时自动进入 pdb；pytest --trace 测试开始时进入 pdb
• pytest + debugpy：在 VS Code 中将 launch.json 的 purpose 设为 ["debug-test"]，可直接调试测试用例
• pytest 减少日志噪音：pytest --log-cli-level=DEBUG 在测试时显示指定级别的日志
• 单测作为调试入口：写一个最小的失败测试用例来复现 bug，比反复跑完整流程高效得多
• pytest-watch：ptw 自动监控文件变化重新运行相关测试，调试驱动开发（DDD）利器