第 10 章:Mini Emperor 综合项目——把 Agent 做成别人能使用的产品
开场:同一道菜,四种吃法
一家面馆的招牌牛肉面做得很好。老板发现,只靠堂食不够。于是他做了四件事:店里堂食照卖(Web 在线运行)、出了速食包装卖到超市(标准 Skill ZIP)、给别的餐厅供货半成品(远程 MCP)、把配方和质检标准公开给监管部门(GitLab CI)。
四条渠道看起来完全不同——堂食是端碗上桌,速食包是拆袋下锅,供货是冻货进冷库,监管是文件过审。但有一条底线老板从没含糊:不管从哪条渠道出去,用的都是同一锅汤、同一份配方、同一套卫生标准。 如果超市卖的速食包跟堂食不是一回事,配方各写各的,质量会立刻失控。
Agent 产品也一样。前九章分别解决了 Agent Loop、上下文、记忆、规划、Subagent、团队、工具、Hook、评估和 Goal——它们都是「做出一道菜」。这一章要回答更难的问题:
如何把这些能力组合成一个可以安装、调用、观察、评估、发布和回滚的产品?
Mini Emperor 把知识、Skill、工具和评估集中到后端,再用不同协议暴露给不同入口。本章结束时,你能在另一台干净环境里走完八步演示,并回答:为什么四个入口必须共享同一份内核,而不是各写一套?
同一份客服能力有四个入口,为什么不能各写一套业务逻辑?
- A:四个入口各写一套,反而更容易维护
- B:避免业务逻辑漂移——知识、Skill、工具、策略集中到后端,用不同协议暴露
- C:四个入口不需要共享任何东西
- D:因为前端代码写起来更方便
一个客服,四个入口,一份内核
「同一份客服能力」在 Mini Emperor 里有两个实体:可移植的 SKILL.md 和后端服务。四个入口共用这两样,只换暴露协议:
| 入口 | 面向谁 | 共享什么 |
|---|---|---|
| Web 在线运行 | 普通用户 | Skill 指令、RAG、权限策略 |
| 标准 Skill ZIP | Agent 用户 | SKILL.md、references、固定版本 |
| 远程 MCP | 开发者/其他 Agent | 检索和工单工具 |
| GitLab CI | 研发团队 | 审核 Skill、Verifier、门禁规则 |
Skill 本体分两层:标准层是 SKILL.md + scripts/ + references/ + assets/,跨 Agent 可移植;平台层是可选的 skill.web.json,只声明 Web 运行所需的 Schema、runtime、工具、网络和密钥可见性。没有 Web 扩展时,Skill 依然可以下载和安装——标准是主,平台扩展是可选的附加。
单一能力,多种交付面这句话,是全章的核心判断。能力集中,才不会被四个入口的重复实现慢慢拖散。
证据驱动的回答:RAG 不是「猜」
客服的核心是 rag.py,它的回答必须由证据驱动,而不是由模型自由发挥。一次回答走五步:
KnowledgeRetriever.search(question, top_k=3)检索知识库;- 丢弃低于
minimum_score的命中——分数不够,宁可不说; - 没有命中,拒答,而不是编一个答案;
- 有命中,返回
document_id / title / excerpt作为引用; - 投诉、赔偿、律师、报警、人工客服等高风险词,标记
needs_human。
使用模型时,证据和问题都作为不可信数据处理:模型只被要求基于证据回答。这样「引用可追溯」才有意义——用户能拿 document_id 去核对,而不是听一句无凭无据的话。
知识库检索没有任何命中时,客服应该怎么做?
- A:拒答——没有证据就不回答
- B:随便编一个答案
- C:让用户换一个问题再试
- D:引用一个空文档
两个「不重复」:SSE 续传与 run 幂等
客服运行会推给前端一串事件:run.started → retrieval.completed → review.completed → run.completed。断线重连时,GET /api/skill-runs/{id}/events 配合 Last-Event-ID 头,服务端跳过已交付的事件,只补发还没消费的部分。
这里有一个必须分清的点:SSE 续传和 run 创建幂等是两件事。
- SSE 投递恢复:同一个 event ID 不重复展示。浏览器
EventSource重连时自动携带Last-Event-ID,服务端跳过已交付事件。当前 API 已经实现了这一层——事件有递增 ID,续传不重复交付。 - run 创建幂等:同一个
Idempotency-Key不重复执行。当前 API 在create_run请求内同步完成工作,再保存事件,还没有实现「创建请求幂等键 + 后台任务队列」。
生产化时这两层都要有。说清楚当前做到了哪一层、还差哪一层,正是综合项目要求的诚实。
SSE 续传与 run 创建幂等是同一件事吗?
- A:是,都是防止重复
- B:完全无关
- C:不是——一个保证「事件不重复投递」,一个保证「run 不重复执行」;当前只实现了前者
- D:两个都是纯前端的事
ZIP 下载只是分发的一半
Skill 要能被别人安装,得先被安全地分发。packaging.py 把「下载」和「安全安装」拆成两半,每一半都有硬边界:
- 构建时拒绝符号链接;
- 安装时拒绝绝对路径、
..、多根目录和符号链接; - 校验 SHA-256;
- 目标已存在时默认拒绝覆盖;
- 更新时先装到暂存目录,再原子替换,失败恢复备份;
- 解压后重新执行
load_skill校验。
一个恶意 ZIP 如果带了 ../../etc/passwd 或符号链接指向系统目录,解压就会逃出目标目录。校验和、路径检查和原子替换,把这些洞一个个堵上。
远程 MCP:能力可以借出去,权限不能
第 7 章讲过,MCP 是「怎么接入」的协议。Mini Emperor 用 mcp_server.py 把检索和工单能力暴露成四个 Tool:
| 能力 | Scope | 风险 |
|---|---|---|
search_kb | kb:read | 只读,可能泄露内部知识 |
get_article | kb:read | 文档越权 |
create_support_ticket | ticket:write | 外部副作用 |
get_ticket_status | ticket:write | 客户信息泄露 |
Key 只保存 SHA-256 摘要;Token Verifier 返回 scope 和脱敏 client ID,不返回原始凭据。一个只有 kb:read 的 Key,能搜索知识库,但创建工单会被拒——能力借出去了,权限没有。
安装一个 Skill ZIP 时,正确的安全行为是?
- A:直接解压到目标目录,快就行
- B:拒绝绝对路径、
..、多根目录和符号链接,先解压到暂存目录再原子替换 - C:只检查文件名长短
- D:信任任何来源的 ZIP
Skill 自进化:只生成候选
第 3 章讲过:一次成功不等于一个 Skill。Mini Emperor 把进化压成一条状态机:
draft → evaluating → candidate → published
↘ quarantine
published → deprecated
deprecated → published (rollback)SkillOptimizer.propose 从脱敏轨迹生成候选,但候选只是草稿。晋级要求是一串不可互相抵消的约束:成功率至少提升 10 个百分点,或质量不下降且成本降低至少 15%;安全零回退;确定性测试全过;人工批准。发布是状态切换,不是覆盖文件——灰度发现异常时先隔离或回滚,稳定版始终可解析。
MR 审核:提议与证据分开
第 8 章讲过「Verifier 判断证据是否支持结论」。Mini Emperor 的 AsyncReviewPipeline 把它落地成三步:先跑确定性 Reviewer,再让模型提议发现,最后回查 unified diff 的文件、行号和原文。模型说「第 42 行有问题」,但如果 diff 里 42 行不是那行代码,这条发现直接丢弃。
gitlab_review.py 把报告落地到 GitLab:报告永远保存为 Artifact(稳定证据),有最小权限 Token 时再写 MR Note(协作入口);P0/P1 才让 CLI 退出码变为 1,P2/P3 只进入报告不阻断。.gitlab-ci.yml 的规则只允许 merge_request_event + target master + source feature/* 触发,保证门禁在合并前运行、不会在普通 push 时误触发。
Reviewer 提议「第 42 行有问题」,Verifier 应该做什么?
- A:直接相信并阻断
- B:跳过这条,不处理
- C:在 diff 中回查同一文件、行号和原文,找不到证据就丢弃该发现
- D:让模型重新提议一遍
完整流程:八个入口,一条证据链
毕业演示是一条状态链,每走一步都要有真实产物:
healthy
→ customer run completed with citations
→ SSE resumed from event 1
→ ZIP checksum verified and installed
→ MCP authenticated and tool discovered
→ candidate evaluated and approved
→ new version published
→ version quarantined or rolled back
→ MR report generated and gate enforced观察一个运行中的产品,至少要看四类状态:服务状态(/health/live、/health/ready、/api/skills)、Run 与 SSE 状态(事件 ID 与续传)、Skill 版本状态(status、evaluation.eligible、approved_by,不要只看版本号)、安全状态(日志里只能出现 scope、哈希、脱敏 client ID,绝不能出现 DEEPSEEK_API_KEY、GITLAB_TOKEN、MCP_API_KEY 的值,也不能有用户密码、支付信息或私人数据)。
动手:让两个失败发生,再用一个测试收住
理论和手感之间隔着一道墙,亲手撞一次才会留下印象。下面两个实验都是故意做错的。
失败一:SSE 事件重复投递。
# 错误:重连后把已经投递过的事件再发一遍
delivered = {1, 2}
new_client_events = [1, 2, 3, 4] # 客户端只缺 3、4
for event_id in new_client_events:
send_to_client(event_id) # ← 错:1、2 已经交付过
# 前端会重复渲染 1、2,用户看到同一事件出现两次修复是 Last-Event-ID 续传:只发送大于 resume_after 的事件。但注意——你修的是「投递重复」,不要误以为已经实现了「run 创建幂等」。
失败二:Reviewer 的发现没有回查证据就阻断。
# 错误:把模型的提议直接当成门禁证据
findings = [
{"file": "app.py", "line": 99, "text": "这里会崩溃"}, # 模型编的
]
if any(f["severity"] in ("P0", "P1") for f in findings):
exit(1) # ← 错:没有回查 diff,一条幻觉就能拦住合并模型说「第 99 行会崩溃」,但 diff 里根本没有这一行。正确做法是让 Verifier 回查同一文件、行号和原文,找不到证据的发现直接丢弃——只有通过证据核验的 P0/P1 才能阻断。
跑完后用现成测试验证「客服 run 以引用结束」这条完整链路真实有效——它一次跑通 RAG 检索、证据、引用和终态:
cd "$(git rev-parse --show-toplevel)"
uv run --python 3.12 --extra dev pytest \
mini-emperor/backend/tests/test_api.py::test_customer_support_skill_run_finishes_with_citations -v如果它不过,按这个顺序排查:
- 没有引用 →
rag.py的命中有没有低于minimum_score被丢弃; - 库外问题没有拒答 →
_evidence阈值和refused字段; - 测试根本没跑到 → 文件名与函数名是否一致、路径是否在
mini-emperor/backend/tests/下。
门禁·代码:跑通上面这条测试,把输出保存为证据。
本章小结
这一章把前九章的能力收进一个产品。五件事连起来看:
单一能力,多种交付面。 同一个客服,Web、ZIP、MCP、CI 四个入口共享同一份内核——知识、Skill、工具、策略集中到后端,用不同协议暴露。能力集中,才不会漂移。
回答必须有证据。 RAG 检索不到就拒答,命中带引用,高风险词转人工。证据和问题都当不可信数据,模型只能基于证据回答。
「不重复」要分清。 SSE 续传保证事件不重复投递(已实现);run 创建幂等保证请求不重复执行(待生产化)。两件事,两个承诺。
分发与发布都要守边界。 ZIP 拒绝路径穿越、符号链接和错误覆盖;MCP 让能力借出去但权限收住(kb:read 与 ticket:write 分开);Skill 候选必须评估、批准才能发布;MR 的 Reviewer 提议必须经 Verifier 回查才能阻断。
一句话记住这一章:
不是「页面打开了」,而是另一台干净环境能完成八步演示——跑通、续传、校验、鉴权、进化、发布、回滚、门禁,每一步都有真实产物可查。
如果上面这些话你自己也能说得出来(不看资料),就说明这一章的知识已经连成一片。
复述与证据
合上全部教材,画出端到端流程:Vue / CLI / MCP Client / CI → FastAPI + SSE + /mcp → SkillRegistry / CapabilityPolicy / Agent Runtime → RAG / Reviewer → ZIP / 事件 / 候选 → 人工批准 / 合并门禁。标出四个入口分别共享了什么、各守了什么权限边界。
再追问五个问题:
- 一个 Web 客服请求如何从 Vue 到 RAG,再以 SSE 和引用返回?
- SSE 续传和 run 幂等为什么不是同一件事?
- Skill ZIP 安装器如何防止路径穿越、符号链接逃逸和错误覆盖?
- Skill 候选为什么不能自动发布?完整状态链是什么?
- MR Reviewer 的发现为什么必须再由 Verifier 回查?
门禁·证据:提交一条客服 run 的真实运行轨迹,标出事件序列、引用文档 ID、以及 run.completed 终态。
24 小时后:不看资料,在白板上画出完整状态链 healthy → run completed → SSE resumed → ZIP verified → MCP scoped → candidate approved → published → quarantined/rolled back → MR gated,并说出每一步对应的真实文件或 API。
有兴趣可以继续看
以下资料按优先级排。至少完成一项 P0,并保存完成证据。
工程P090 分钟 · mini-emperor/backend/src/mini_emperor/api.py 的 create_run / stream_run / proposal / evaluate / approve / publish / rollback
- 预计
- 90 分钟
- 位置
mini-emperor/backend/src/mini_emperor/api.py 的 create_run / stream_run / proposal / evaluate / approve / publish / rollback- 阅读任务
- 从 create_run 追到 stream_run,再追踪 proposal、evaluate、approve、publish 和 rollback。
- 完成证据
- 运行 test_api.py,并为每个端点标注状态输入与输出。
书与实验P060 分钟 · chapter3/retrieval-pipeline/evaluate.py
- 预计
- 60 分钟
- 位置
chapter3/retrieval-pipeline/evaluate.py- 阅读任务
- 运行评估并解释 Recall@5、拒答率和引用追溯为什么必须分开。
- 完成证据
- 保存指标,并手工审查一个召回成功但答案不合格的案例。
书与实验P060 分钟 · chapter8/self-modifying-agent/evolution.py
- 预计
- 60 分钟
- 位置
chapter8/self-modifying-agent/evolution.py- 阅读任务
- 比较运行经验、候选指令和稳定版本,确认优化器无发布权限。
- 完成证据
- 提交一次候选被拒绝后稳定版仍有效的测试。
官方文档P050 分钟 · ../source-notes/official/official-docs.md 的 MCP Authorization、Security 与 Registry
- 预计
- 50 分钟
- 位置
../source-notes/official/official-docs.md 的 MCP Authorization、Security 与 Registry- 阅读任务
- 核对 Bearer、scope、Streamable HTTP、server.json 和远程分发边界。
- 完成证据
- 写出本地演示鉴权与生产 OAuth 的差异。
教学仓库P140 分钟 · claude-agent-examples@54a18980334541773f13940aa0c0475728d30ee0:build-agent-example/code/step12_hooks.py
- 预计
- 40 分钟
- 位置
claude-agent-examples@54a18980334541773f13940aa0c0475728d30ee0:build-agent-example/code/step12_hooks.py- 阅读任务
- 把单文件 Hook 的职责逐项映射到 Mini Emperor 的权限、事件、评估和审核模块。
- 完成证据
- 提交一张从教学单文件到产品模块的映射表。
论文P160 分钟 · ../source-notes/papers/classic-papers.md 的 SWE-bench、AgentBench 与 LLM-as-a-Judge
- 预计
- 60 分钟
- 位置
../source-notes/papers/classic-papers.md 的 SWE-bench、AgentBench 与 LLM-as-a-Judge- 阅读任务
- 比较可执行测试、环境交互和模型裁判的适用边界。
- 完成证据
- 为客服、Skill 发布和 MR 审核各选择一种主要评估方式并说明理由。
本文提到的代码与出处(有兴趣逐条核对时用):
claude-agent-examples@54a18980334541773f13940aa0c0475728d30ee0:build-agent-example/code/step01_single_call.pyclaude-agent-examples@54a18980334541773f13940aa0c0475728d30ee0:build-agent-example/code/step12_hooks.pyclaude-agent-examples@54a18980334541773f13940aa0c0475728d30ee0:tests/test_step12_hooks.pyai-agent-book@1c18370279f8f0457bf2c44dfa585d08d4d5f281:chapter3/retrieval-pipeline/evaluate.pyai-agent-book@1c18370279f8f0457bf2c44dfa585d08d4d5f281:chapter5/coding-agent/agent.pyai-agent-book@1c18370279f8f0457bf2c44dfa585d08d4d5f281:chapter6/agent-cost-analysis/tracer.pyai-agent-book@1c18370279f8f0457bf2c44dfa585d08d4d5f281:chapter8/trajectory-verifier/verifier.pyai-agent-book@1c18370279f8f0457bf2c44dfa585d08d4d5f281:chapter8/self-modifying-agent/evolution.pymini-emperor/backend/src/mini_emperor/rag.pymini-emperor/backend/src/mini_emperor/packaging.pymini-emperor/backend/src/mini_emperor/mcp_server.pymini-emperor/backend/src/mini_emperor/review.pymini-emperor/backend/src/mini_emperor/gitlab_review.pymini-emperor/backend/tests/test_api.pymini-emperor/backend/tests/test_mcp_service.py