615 lines
29 KiB
Markdown
615 lines
29 KiB
Markdown
---
|
||
id: plan.week13
|
||
title: "第 13 周：项目三 M4——评测集、可观测性、文档、Docker + Dify / Coze 体验"
|
||
type: week
|
||
domain: plan
|
||
stack: [docker, prompt-engineering, agent-ops, lowcode]
|
||
week: [13]
|
||
est_hours: 12
|
||
order: 13
|
||
---
|
||
|
||
# 第 13 周：项目三 M4——评测集、可观测性、文档、Docker + Dify / Coze 体验
|
||
|
||
> 项目三的最后一周（阶段 3 出门）。上周 `week12-agent-m3-persistence.md`，下周 `week14-backend-extend-deploy.md`。
|
||
|
||
**本周目标**：
|
||
|
||
1. **评测**：能用数字说明“改了提示词 / 流程之后变好还是变差”（招聘里的“Agent 评测”）。
|
||
2. **可观测性**：每次运行调用了几次模型、用了哪些工具、花了多少时间和 token；了解 LangSmith / Langfuse。
|
||
3. **文档**：架构图、设计取舍、与生产概念的映射表、讲解稿——面试时直接用。
|
||
4. **Docker** 一键启动；别人按 README 10 分钟跑通。
|
||
5. **Dify / Coze** 体验：能讲清低代码平台和写代码的取舍。
|
||
|
||
**本周新增文件**
|
||
|
||
```text
|
||
agent-mini/
|
||
├── src/agent_mini/
|
||
│ ├── evals.py # 用例结构、规则检查、模型评分、汇总
|
||
│ └── observability.py # 回调：统计模型调用、工具、token、耗时
|
||
├── evals/
|
||
│ ├── cases.jsonl
|
||
│ ├── run_evals.py
|
||
│ └── reports/
|
||
├── docs/ architecture.md design-decisions.md mapping.md talk.md
|
||
├── Dockerfile
|
||
├── docker-compose.yml
|
||
└── tests/
|
||
├── test_evals.py
|
||
└── test_observability.py
|
||
notes/lowcode-dify-coze.md
|
||
lowcode/ # Dify / Coze 导出的配置（去掉密钥）
|
||
```
|
||
|
||
---
|
||
|
||
## 第 1 天：评测集和规则检查
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- Agent 评测比 RAG 难，因为过程有多步。从三个层面看：
|
||
- **结果**：结论是否回答了问题、是否提到必须提到的点。
|
||
- **过程**：该调的工具调了没有、不该调的有没有乱调；轮数是否合理；该问人时有没有问。
|
||
- **成本和稳定性**：耗时、模型调用次数、token、失败率。
|
||
- **能用规则判断的用规则**（便宜、稳定、可复现）；不能的再用“模型按评分标准打分”（LLM-as-judge），并抽一部分人工复核，确认模型评分靠谱。
|
||
- 用例要覆盖：普通问题、需要查知识库的问题、信息不足需要问人的问题、**诱导乱调工具的问题**（比如“帮我删掉所有待办”——不应该调用有副作用的工具）。
|
||
- 评测用 `InMemorySaver`，每条用例一个 `thread_id`；需要问人的用例自动给预设回答。
|
||
- 案例：
|
||
|
||
```json
|
||
{"id": "basic-1", "question": "5 人创业团队做内部工具，该选什么技术栈？", "expect": {"max_rounds": 3, "must_mention": ["方案"], "should_ask_human": false}}
|
||
```
|
||
|
||
```json
|
||
{"id": "clarify-1", "question": "[需要澄清] 帮我做个网站", "human_answer": "预算 5 万以内", "expect": {"should_ask_human": true}}
|
||
```
|
||
|
||
```json
|
||
{"id": "tool-1", "question": "每人每月 2000 元，5 个人一年预算多少？", "expect": {"must_call_tools": ["calculator"]}}
|
||
```
|
||
|
||
（每行一条，保存为 `evals/cases.jsonl`，至少 20 条；上面三行是格式示例。）
|
||
|
||
```python
|
||
# 文件：agent-mini/src/agent_mini/evals.py
|
||
import json
|
||
import re
|
||
import time
|
||
from pathlib import Path
|
||
|
||
from langchain_core.language_models import BaseChatModel
|
||
from langchain_core.messages import HumanMessage, SystemMessage
|
||
from langgraph.checkpoint.memory import InMemorySaver
|
||
from langgraph.types import Command
|
||
from pydantic import BaseModel, Field
|
||
|
||
from agent_mini.graph.build import build_graph
|
||
|
||
|
||
class Expect(BaseModel):
|
||
must_call_tools: list[str] = []
|
||
must_not_call_tools: list[str] = []
|
||
should_ask_human: bool | None = None
|
||
max_rounds: int | None = None
|
||
must_mention: list[str] = []
|
||
|
||
|
||
class EvalCase(BaseModel):
|
||
id: str
|
||
question: str
|
||
human_answer: str = "按常见情况假设即可"
|
||
expect: Expect = Expect()
|
||
rubric: str = ""
|
||
|
||
|
||
class CaseResult(BaseModel):
|
||
id: str
|
||
passed: bool
|
||
failures: list[str]
|
||
rounds: int
|
||
tools: list[str]
|
||
asked_human: bool
|
||
end_reason: str
|
||
seconds: float
|
||
judge_score: int | None = None
|
||
|
||
|
||
def load_cases(path: Path) -> list[EvalCase]:
|
||
return [EvalCase.model_validate_json(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
|
||
|
||
|
||
def check(case: EvalCase, out: dict, asked_human: bool) -> list[str]:
|
||
tools = [t for turn in out.get("turns", []) for t in turn.get("tools", [])]
|
||
e, fails = case.expect, []
|
||
fails += [f"没有调用 {t}" for t in e.must_call_tools if t not in tools]
|
||
fails += [f"不该调用 {t}" for t in e.must_not_call_tools if t in tools]
|
||
if e.should_ask_human is not None and asked_human != e.should_ask_human:
|
||
fails.append("应该问人但没问" if e.should_ask_human else "不该问人却问了")
|
||
if e.max_rounds is not None and out.get("round", 0) > e.max_rounds:
|
||
fails.append(f"轮数 {out.get('round')} 超过 {e.max_rounds}")
|
||
answer = out.get("final_answer", "") + "".join(t["content"] for t in out.get("turns", []))
|
||
fails += [f"没有提到“{w}”" for w in e.must_mention if w not in answer]
|
||
if not out.get("final_answer"):
|
||
fails.append("没有最终结论")
|
||
return fails
|
||
|
||
|
||
async def run_case(model: BaseChatModel, case: EvalCase, tools=None, callbacks=None) -> CaseResult:
|
||
graph = build_graph(model, tools, checkpointer=InMemorySaver())
|
||
cfg = {"configurable": {"thread_id": case.id}, "callbacks": callbacks or []}
|
||
start = time.perf_counter()
|
||
out = await graph.ainvoke({"question": case.question, "max_rounds": 4, "round": 0}, cfg)
|
||
asked = False
|
||
for _ in range(3): # 最多回答 3 次问题，防止评测卡住
|
||
if "__interrupt__" not in out:
|
||
break
|
||
asked = True
|
||
out = await graph.ainvoke(Command(resume=case.human_answer), cfg)
|
||
fails = check(case, out, asked)
|
||
tools_used = [t for turn in out.get("turns", []) for t in turn.get("tools", [])]
|
||
return CaseResult(id=case.id, passed=not fails, failures=fails, rounds=out.get("round", 0), tools=tools_used,
|
||
asked_human=asked, end_reason=out.get("end_reason", ""),
|
||
seconds=round(time.perf_counter() - start, 3))
|
||
|
||
|
||
class JudgeOut(BaseModel):
|
||
score: int = Field(ge=1, le=5)
|
||
reason: str
|
||
|
||
|
||
JUDGE = """你是严格的评审。按评分标准给回答打 1~5 分（5 最好）。只输出 JSON：{"score": 整数, "reason": "一句话理由"}"""
|
||
|
||
|
||
async def judge(model: BaseChatModel, case: EvalCase, answer: str) -> JudgeOut:
|
||
ai = await model.ainvoke([SystemMessage(JUDGE), HumanMessage(
|
||
f"问题：{case.question}\n评分标准：{case.rubric or '是否具体、可执行、考虑了成本和风险'}\n回答：{answer}")])
|
||
m = re.search(r"\{.*\}", str(ai.content), re.S)
|
||
return JudgeOut.model_validate_json(m.group(0) if m else "{}")
|
||
|
||
|
||
def summarize(results: list[CaseResult]) -> dict:
|
||
n = len(results) or 1
|
||
tool_cases = [r for r in results if r.tools]
|
||
return {
|
||
"cases": len(results),
|
||
"pass_rate": round(sum(r.passed for r in results) / n, 3),
|
||
"avg_rounds": round(sum(r.rounds for r in results) / n, 2),
|
||
"avg_seconds": round(sum(r.seconds for r in results) / n, 3),
|
||
"cases_with_tools": len(tool_cases),
|
||
"failures": {r.id: r.failures for r in results if r.failures},
|
||
}
|
||
|
||
|
||
def to_markdown(summary: dict, meta: dict) -> str:
|
||
lines = ["# 评测报告", "", *[f"- {k}：{v}" for k, v in meta.items()], "",
|
||
"| 指标 | 值 |", "| --- | --- |",
|
||
*[f"| {k} | {v} |" for k, v in summary.items() if k != "failures"], "", "## 失败用例"]
|
||
lines += [f"- {cid}：{'；'.join(f)}" for cid, f in summary["failures"].items()] or ["- 无"]
|
||
return "\n".join(lines)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print(json.dumps(Expect().model_dump(), ensure_ascii=False))
|
||
```
|
||
|
||
**易错点 / 面试常问（补充）**
|
||
|
||
- 易错点：用例的期望要写成能自动判断的规则（必须调用哪个工具、是否应该问人、必须提到什么），只写“回答得好”是没法自动评测的。
|
||
- 易错点：评测要固定模型、温度和提示词版本，否则两次结果的差异可能只是随机波动。
|
||
- 面试常问：Agent 评测和 RAG 评测的区别（除了结果，还要评过程：工具选择、步数、是否该停下来问人、成本）。
|
||
|
||
### 要建的文件
|
||
|
||
- `evals.py`、`evals/cases.jsonl`（≥ 20 条，覆盖上面 4 类）
|
||
|
||
### 运行方式
|
||
|
||
- 明天和测试、评测脚本一起跑。
|
||
|
||
### 完成标准
|
||
|
||
- 用例覆盖 4 类问题；每条至少有一个可以用规则判断的期望。
|
||
|
||
---
|
||
|
||
## 第 2 天：评测脚本、回归测试、用评测驱动改进
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- 评测脚本逐条运行，汇总通过率、平均轮数、耗时，结果写进 `evals/reports/<日期>.md`，**记录当时的模型和提示词版本**（否则数字没法比较）。
|
||
- 用假模型做一个不需要密钥的小评测子集，放进测试——这就是**回归测试**：以后改代码，流程被改坏会立刻发现。
|
||
- 改进流程：跑基线 → 挑失败最多的一类 → 改提示词或路由规则 → 再跑 → 对比。改动和结果记到 `docs/prompt-changelog.md`。**数字以你实际跑出来的为准，不要编。**
|
||
- 案例：
|
||
|
||
```python
|
||
# 文件：agent-mini/tests/test_evals.py
|
||
from langchain_core.messages import AIMessage
|
||
|
||
from agent_mini.demo_model import demo_model
|
||
from agent_mini.evals import EvalCase, Expect, judge, run_case, summarize
|
||
from agent_mini.fakes import decision, scripted_model
|
||
from agent_mini.tools import LOCAL_TOOLS
|
||
|
||
|
||
async def test_regression_suite_with_fake_model():
|
||
cases = [
|
||
EvalCase(id="basic", question="选技术栈", expect=Expect(max_rounds=3, must_mention=["方案"], should_ask_human=False)),
|
||
EvalCase(id="clarify", question="[需要澄清] 做网站", human_answer="5 万", expect=Expect(should_ask_human=True)),
|
||
]
|
||
results = [await run_case(demo_model(), c) for c in cases]
|
||
s = summarize(results)
|
||
assert s["pass_rate"] == 1.0, s["failures"]
|
||
assert results[1].asked_human and results[1].end_reason == "consensus"
|
||
|
||
|
||
async def test_detects_missing_tool_call():
|
||
model = scripted_model(coordinator=[decision("dispatch", "proposer"), decision("finish")],
|
||
proposer=["大概 12 万吧"], summarizer=["12 万"])
|
||
r = await run_case(model, EvalCase(id="t", question="算预算", expect=Expect(must_call_tools=["calculator"])), LOCAL_TOOLS)
|
||
assert not r.passed and r.failures == ["没有调用 calculator"]
|
||
|
||
|
||
async def test_tool_case_passes_when_tool_used():
|
||
model = scripted_model(
|
||
coordinator=[decision("dispatch", "proposer"), decision("finish")],
|
||
proposer=[AIMessage(content="", tool_calls=[{"name": "calculator", "args": {"expression": "2000*5*12"}, "id": "c"}]),
|
||
"一年 120000 元"],
|
||
summarizer=["一年预算 12 万"])
|
||
r = await run_case(model, EvalCase(id="t", question="算预算", expect=Expect(must_call_tools=["calculator"])), LOCAL_TOOLS)
|
||
assert r.passed and r.tools == ["calculator"]
|
||
|
||
|
||
async def test_judge_parses_score():
|
||
from agent_mini.llm import FakeChatModel
|
||
|
||
j = FakeChatModel(responder=lambda messages: AIMessage(content='评分：{"score": 4, "reason": "具体可执行"}'))
|
||
out = await judge(j, EvalCase(id="x", question="q"), "回答")
|
||
assert out.score == 4 and out.reason == "具体可执行"
|
||
```
|
||
|
||
```python
|
||
# 文件：agent-mini/evals/run_evals.py
|
||
"""用法：uv run python evals/run_evals.py [--fake]"""
|
||
import asyncio
|
||
import datetime as dt
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
from agent_mini.evals import load_cases, run_case, summarize, to_markdown
|
||
from agent_mini.observability import RunMetrics
|
||
from agent_mini.tools import LOCAL_TOOLS
|
||
|
||
|
||
async def main() -> None: # 不自动运行：默认用真实模型
|
||
fake = "--fake" in sys.argv
|
||
if fake:
|
||
from agent_mini.demo_model import demo_model as make
|
||
else:
|
||
from agent_mini.llm import build_chat_model as make
|
||
cases = load_cases(Path(__file__).with_name("cases.jsonl"))
|
||
metrics = RunMetrics()
|
||
results = [await run_case(make(), c, LOCAL_TOOLS, callbacks=[metrics]) for c in cases]
|
||
summary = summarize(results) | {"model_calls": metrics.llm_calls, "total_tokens": metrics.total_tokens}
|
||
meta = {"日期": dt.date.today().isoformat(), "模型": "fake" if fake else "见 .env 的 CHAT_MODEL",
|
||
"提示词版本": "见 docs/prompt-changelog.md 最新一行"}
|
||
report = to_markdown(summary, meta)
|
||
out = Path(__file__).with_name("reports") / f"{dt.datetime.now():%Y%m%d-%H%M}.md"
|
||
out.parent.mkdir(exist_ok=True)
|
||
out.write_text(report, encoding="utf-8")
|
||
print(report)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|
||
```
|
||
|
||
**易错点 / 面试常问（补充）**
|
||
|
||
- 易错点：用模型打分（LLM-as-judge）有偏差（偏爱长回答、偏爱自己的风格），要抽样人工复核，并给出明确的评分标准。
|
||
- 易错点：`--fake` 模式下“必须调用计算器”这类用例会失败是正常的（演示假模型不调工具），它只用来确认脚本能跑。
|
||
- 面试常问：怎么证明你的优化有效（同一套评测集、固定配置、改一处跑一次、记录前后指标）。
|
||
|
||
### 要建的文件
|
||
|
||
- `tests/test_evals.py`、`evals/run_evals.py`、`evals/reports/`、`docs/prompt-changelog.md`
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
uv run pytest tests/test_evals.py -v
|
||
uv run python evals/run_evals.py --fake # 先确认脚本能跑
|
||
uv run python evals/run_evals.py # 真实模型：基线
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 至少两次真实模型评测（基线 + 一次改进），报告都在 `evals/reports/`，changelog 写清改了什么、指标怎么变。
|
||
|
||
---
|
||
|
||
## 第 3 天：可观测性（回调、LangSmith / Langfuse）
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- **为什么需要**：Agent 出问题时，你要能回答“这次运行调了几次模型、每次输入输出是什么、哪个工具出错、花了多少钱和时间”。没有这些，线上问题无从排查。
|
||
- **LangChain 回调**：在 `config={"callbacks": [...]}` 里传入回调对象，模型开始 / 结束、工具开始 / 结束时都会通知它——可以自己统计指标、写日志。
|
||
- **LangSmith**：LangChain 官方的追踪和评测平台。设置环境变量后自动上报每次运行的完整调用树（不用改代码）：`LANGSMITH_TRACING=true`、`LANGSMITH_API_KEY=...`、`LANGSMITH_PROJECT=agent-mini`。**变量名和用法以官方文档为准**：https://docs.langchain.com/langsmith/home 。注意：上报的内容包含提示词和回答，**涉及敏感数据时不要开**，或者用可以自己部署的开源方案 **Langfuse**（https://langfuse.com/docs ）。
|
||
- 案例：自己写一个统计回调，评测脚本和服务都能用。
|
||
|
||
```python
|
||
# 文件：agent-mini/src/agent_mini/observability.py
|
||
import time
|
||
from typing import Any
|
||
|
||
from langchain_core.callbacks import BaseCallbackHandler
|
||
from langchain_core.outputs import LLMResult
|
||
|
||
|
||
class RunMetrics(BaseCallbackHandler):
|
||
"""统计一次（或多次）运行的模型调用、token、工具和耗时。"""
|
||
|
||
def __init__(self) -> None:
|
||
self.llm_calls = 0
|
||
self.total_tokens = 0
|
||
self.llm_seconds = 0.0
|
||
self.tools: list[str] = []
|
||
self.tool_errors = 0
|
||
self._starts: dict[Any, float] = {}
|
||
|
||
def on_chat_model_start(self, serialized: dict, messages: list, *, run_id: Any, **kwargs: Any) -> None:
|
||
self._starts[run_id] = time.perf_counter()
|
||
|
||
def on_llm_end(self, response: LLMResult, *, run_id: Any, **kwargs: Any) -> None:
|
||
self.llm_calls += 1
|
||
self.llm_seconds += time.perf_counter() - self._starts.pop(run_id, time.perf_counter())
|
||
for gens in response.generations:
|
||
for g in gens:
|
||
usage = getattr(getattr(g, "message", None), "usage_metadata", None) or {}
|
||
self.total_tokens += usage.get("total_tokens", 0) # 假模型没有用量，真实模型才有
|
||
|
||
def on_tool_start(self, serialized: dict, input_str: str, **kwargs: Any) -> None:
|
||
self.tools.append(serialized.get("name", "?"))
|
||
|
||
def on_tool_error(self, error: BaseException, **kwargs: Any) -> None:
|
||
self.tool_errors += 1
|
||
|
||
def snapshot(self) -> dict:
|
||
return {"llm_calls": self.llm_calls, "total_tokens": self.total_tokens,
|
||
"llm_seconds": round(self.llm_seconds, 3), "tools": self.tools, "tool_errors": self.tool_errors}
|
||
```
|
||
|
||
```python
|
||
# 文件：agent-mini/tests/test_observability.py
|
||
from langchain_core.messages import AIMessage
|
||
|
||
from agent_mini.evals import EvalCase, run_case
|
||
from agent_mini.fakes import decision, scripted_model
|
||
from agent_mini.observability import RunMetrics
|
||
from agent_mini.tools import LOCAL_TOOLS
|
||
|
||
|
||
async def test_metrics_count_llm_and_tools():
|
||
model = scripted_model(
|
||
coordinator=[decision("dispatch", "proposer"), decision("finish")],
|
||
proposer=[AIMessage(content="", tool_calls=[{"name": "calculator", "args": {"expression": "1+1"}, "id": "c"}]),
|
||
"结果是 2"],
|
||
summarizer=["2"])
|
||
metrics = RunMetrics()
|
||
await run_case(model, EvalCase(id="m", question="1+1"), LOCAL_TOOLS, callbacks=[metrics])
|
||
snap = metrics.snapshot()
|
||
assert snap["tools"] == ["calculator"]
|
||
assert snap["llm_calls"] == model.calls == 5 # 协调者 2 次 + 提议者 2 次 + 总结 1 次
|
||
```
|
||
|
||
- 把指标接进服务（选做）：`RunManager._drive` 里给 `astream` 的 config 加上 `RunMetrics()`，结束时把 `snapshot()` 写进 `run_completed` 事件，报告里就能看到每次运行的调用次数和 token。
|
||
|
||
**易错点 / 面试常问（补充）**
|
||
|
||
- 易错点：追踪平台会记录完整的提示词和回答，涉及用户隐私或公司数据时不要开云端追踪，改用可以自己部署的方案或只记录指标。
|
||
- 易错点：回调里不要做耗时操作（比如同步写远程日志），会拖慢每一次模型调用。
|
||
- 面试常问：线上 Agent 出问题怎么排查（看调用链 → 定位节点和输入 → 用评测用例复现 → 修复后加进回归测试）；要监控哪些指标（成功率、延迟、每次任务的模型调用次数和 token、工具错误率、人工介入率）。
|
||
|
||
### 要建的文件
|
||
|
||
- `observability.py`、`tests/test_observability.py`；`docs/observability.md`：LangSmith 或 Langfuse 的截图（用假数据或公开题目）和“哪些情况不能开追踪”
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
uv run pytest tests/test_observability.py -v
|
||
LANGSMITH_TRACING=true LANGSMITH_API_KEY=... uv run agent-mini run "选技术栈" # 有账号时体验
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 能回答“线上 Agent 出问题你怎么排查”：看追踪里的调用树 → 找到出错的节点和输入 → 用评测用例复现 → 修复后加进回归测试。
|
||
|
||
---
|
||
|
||
## 第 4 天：文档（架构、设计取舍、映射表）
|
||
|
||
### 学习内容
|
||
|
||
- 说明：文档是面试的“讲稿底本”。映射表把**项目里的实现**和**生产系统里的概念**一一对齐——面试官听到的是“这个人知道生产系统要什么，并且自己实现过”。生产一栏只写通用概念，不写公司接口名、表名。
|
||
- 案例：`docs/mapping.md`
|
||
|
||
```markdown
|
||
| 生产系统里的概念（通用叫法） | 本项目里的实现 | 对应重点 |
|
||
| --- | --- | --- |
|
||
| 席位 / 专家角色 | `proposer`、`critic` 节点 | A |
|
||
| 主持人派活 | `coordinator` 节点 + 条件边 | A |
|
||
| 共识 / 结束条件 | `VERDICT: APPROVE` + 最大轮数，写在 `routing.py` | A |
|
||
| 防止同一任务被跑两次 | `runs` 表租约 + 带条件的 UPDATE | A |
|
||
| 草稿（编排中间状态） | LangGraph 检查点（按 `thread_id`） | B |
|
||
| 任务状态（对外权威） | `runs` 表 | B |
|
||
| 后台作业 | `RunManager` + 看门狗 | C |
|
||
| 流式输出、断线续传 | `GET /runs/{id}/events` + `Last-Event-ID` | C |
|
||
| 意图澄清 / 等待用户输入 | `interrupt` → `awaiting_input` → `POST /input` | D |
|
||
| 换模型恢复 | 错误码 + `POST /retry {model}`，从检查点继续 | D |
|
||
| 过程报告 | `report.py` | E |
|
||
| 外部能力接入 | MCP 服务 `knowledge_server` | E |
|
||
| 用户偏好 | LangGraph Store 长期记忆 | — |
|
||
```
|
||
|
||
`docs/architecture.md` 里放四张图：组件图、业务状态机（第 12 周）、LangGraph 图结构（`agent-mini graph` 导出）、崩溃恢复时序图：
|
||
|
||
```mermaid
|
||
sequenceDiagram
|
||
participant C as 客户端
|
||
participant A as API 进程 1
|
||
participant DB as runs / events / 检查点
|
||
participant B as API 进程 2（重启后）
|
||
C->>A: POST /runs
|
||
A->>DB: 创建 run（queued），领取租约
|
||
A->>DB: 节点完成 → 写检查点、写事件、续租
|
||
Note over A: kill -9（不再续租）
|
||
B->>DB: 看门狗：发现租约过期的 run
|
||
B->>DB: 领取租约，读取最近检查点
|
||
B->>DB: 从中断的节点继续 → 完成
|
||
C->>B: GET /events（Last-Event-ID）补齐事件
|
||
```
|
||
|
||
### 要建的文件
|
||
|
||
- `docs/architecture.md`、`docs/mapping.md`、`docs/design-decisions.md`（每条写“选择 / 理由 / 代价”）
|
||
|
||
### 完成标准
|
||
|
||
- 映射表至少 5 个概念对齐生产；架构图和代码一致。
|
||
|
||
---
|
||
|
||
## 第 5 天：README、Docker、从零验证、讲解稿
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- README 的目标：**别人 10 分钟内跑通**（假模型模式，不需要密钥）。结构：一句话介绍 + 动图；功能；快速开始；切换真实模型；接口表；评测结果；崩溃恢复演示；映射表；项目结构；设计取舍。
|
||
- 从零验证：在新目录（最好换一台电脑或请朋友）`git clone` 后按 README 操作并计时，卡住的地方都改进。
|
||
- 讲解稿：3 分钟，“问题（为什么多 Agent 难做可靠）→ 设计（两层状态、后台作业、租约、中断、记忆）→ 演示 → 评测和取舍”，引用 1~2 张案例卡说明“这些设计来自我在工作中踩过的坑”。
|
||
- 案例：
|
||
|
||
```dockerfile
|
||
FROM python:3.12-slim
|
||
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
|
||
WORKDIR /app
|
||
COPY pyproject.toml uv.lock README.md ./
|
||
RUN uv sync --frozen --no-dev --no-install-project
|
||
COPY src ./src
|
||
COPY mcp_servers ./mcp_servers
|
||
COPY web ./web
|
||
RUN uv sync --frozen --no-dev
|
||
ENV PATH="/app/.venv/bin:$PATH" AGENT_DATA=/app/data AGENT_MODEL=fake
|
||
EXPOSE 8100
|
||
CMD ["uvicorn", "agent_mini.api.app:create_app", "--factory", "--host", "0.0.0.0", "--port", "8100"]
|
||
```
|
||
|
||
```yaml
|
||
services:
|
||
agent:
|
||
build: .
|
||
env_file: .env # AGENT_MODEL、LLM_BASE_URL、LLM_API_KEY 等；不提交
|
||
volumes:
|
||
- agent-data:/app/data # 检查点、业务表、记忆都在这里，重建容器不丢
|
||
ports:
|
||
- "8100:8100"
|
||
restart: unless-stopped
|
||
volumes:
|
||
agent-data:
|
||
```
|
||
|
||
### 要建的文件
|
||
|
||
- `README.md`、`Dockerfile`、`docker-compose.yml`、`.dockerignore`、`docs/talk.md`；录一段 2~3 分钟演示视频
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
docker compose up -d --build && open http://localhost:8100 # Linux 用 xdg-open
|
||
python3 ../tools/check_sensitive.py --self-test && git add -A && git diff --cached --stat
|
||
```
|
||
|
||
### 完成标准
|
||
|
||
- 别人按 README 10 分钟内跑通；讲解稿计时 ≤ 3 分钟。
|
||
|
||
---
|
||
|
||
## 第 6~7 天：Dify / Coze 体验（低代码平台）
|
||
|
||
### 学习内容
|
||
|
||
- 说明：
|
||
- 招聘里经常出现 Dify、Coze（扣子）、RAGFlow、FastGPT。不需要精通，但要**亲手搭过**，能讲清它们能做什么、和写代码比各有什么优缺点。
|
||
- **Dify**：开源，可以自己部署（官方提供 Docker Compose，注意服务器内存，以官方要求为准：https://docs.dify.ai/ ），也有云版。
|
||
- **Coze（扣子）**：字节跳动的在线平台，直接用网页版（https://www.coze.cn/ ）。
|
||
- 二选一为主，另一个看文档即可。
|
||
- 案例：要搭的东西
|
||
|
||
1. **知识库问答**：把 `rag-qa` 用的同一批公开文档导入平台知识库，做一个问答应用；用第 6 周评测集里挑 10 个问题，对比平台和你自己实现的效果（记录命中情况，数字以实际为准）。
|
||
2. **工作流**：做一个带条件分支的工作流，包含一个 HTTP 请求节点调用你的 `agent-mini` 的 `POST /runs`（服务需要能被平台访问；云版平台访问不到你的本机，可以用自己部署的 Dify，或只调用公网可访问的接口）。平台支持 MCP 的话，也可以接你的 MCP 服务。
|
||
3. **导出配置**：平台支持导出（如 Dify 的 DSL 文件）时，放到 `lowcode/`（去掉密钥）。
|
||
|
||
对比笔记的骨架：
|
||
|
||
```markdown
|
||
| 维度 | Dify / Coze | 自己写（rag-qa / agent-mini） |
|
||
| --- | --- | --- |
|
||
| 开发速度 | | |
|
||
| 可控性（提示词、检索参数、状态、错误处理） | | |
|
||
| 复杂编排（循环、中断、续跑） | | |
|
||
| 可观测性和评测 | | |
|
||
| 部署和成本 | | |
|
||
| 谁来维护（业务人员 / 工程师） | | |
|
||
|
||
结论：什么场景用平台（快速验证、业务人员自己搭、标准问答），什么场景写代码（复杂编排、强可靠性、深度集成业务系统）。
|
||
```
|
||
|
||
### 要建的文件
|
||
|
||
- `notes/lowcode-dify-coze.md`（步骤、截图、对比表、结论）、`lowcode/`（导出的配置，去掉密钥）
|
||
|
||
### 运行方式
|
||
|
||
```bash
|
||
# Dify 自己部署（命令以官方文档为准）
|
||
git clone https://github.com/langgenius/dify.git && cd dify/docker && cp .env.example .env && docker compose up -d
|
||
```
|
||
|
||
### 完成标准（M4 验收，也是阶段 3 出门条件）
|
||
|
||
- [ ] 别人按 README 10 分钟内跑通（不需要真实模型密钥）
|
||
- [ ] 演示：启动 →（中途杀进程）→ 重启 → 最终结果可查
|
||
- [ ] 映射表至少 5 个概念对齐生产
|
||
- [ ] 评测集 ≥ 20 条，至少两次对比结果；回归测试在 CI 里跑
|
||
- [ ] 能讲清可观测性怎么做，知道 LangSmith / Langfuse
|
||
- [ ] 架构图和 README 一致；测试、ruff、pyright 通过；脱敏检查完成
|
||
- [ ] Dify 或 Coze 至少搭过一个知识库应用和一个工作流，写了对比笔记
|
||
- [ ] 在总纲进度表勾掉第 13 周和“阶段 3”，填上“项目三路径：`zhuangang/agent-mini/`”
|
||
|
||
<!-- week-topics:start -->
|
||
|
||
## 本周相关专题
|
||
|
||
> 本节由 `tools/gen_week_topics.py` 根据专题文件的 `week` 字段自动生成，请勿手改；想调整就修改对应专题文件的 frontmatter 后重新生成。
|
||
|
||
- **AI 编程练习（每周）**：本周至少一个功能用 AI 编程工具（Cursor / Claude Code / Codex 等）完成，按 [AI 编程：工作流](../topics/ai/ai-coding/01-workflow-rules-skills.md) 写清需求和规则，按 [验证与证据](../topics/ai/ai-coding/02-verify-evidence.md) 保存测试结果和会话记录（脱敏后），第 15 周整理成“AI 编程 + 验证方法”的应聘案例。
|
||
|
||
| 技术栈 | 课程 / 场景 / 面试题 |
|
||
| --- | --- |
|
||
| [Agent 设计](../topics/ai/agent-design/index.md) | 课程：[上下文工程、Skills、多 Agent 和 A2A](../topics/ai/agent-design/02-context-skills-multiagent.md)；场景：[运营报表 Agent](../topics/ai/agent-design/scenarios/report-agent.md)；[面试题](../topics/ai/agent-design/interview.md) |
|
||
| [Agent 评测与运维](../topics/ai/agent-ops/index.md) | 课程：[Agent 评测和质量门禁](../topics/ai/agent-ops/01-eval-quality-gate.md)；课程：[Agent 可观测性、成本和延迟](../topics/ai/agent-ops/02-observability-cost.md)；场景：[客服 Agent 上线流程](../topics/ai/agent-ops/scenarios/agent-quality-gate.md)；[面试题](../topics/ai/agent-ops/interview.md) |
|
||
| [模型侧基础（了解级）](../topics/ai/llm-serving/index.md) | 课程：[模型侧基础](../topics/ai/llm-serving/01-routing-vllm-lora.md)；场景：[公司级大模型网关](../topics/ai/llm-serving/scenarios/multi-provider-gateway.md)；[面试题](../topics/ai/llm-serving/interview.md) |
|
||
| [低代码 AI 平台（Dify / Coze）](../topics/ai/lowcode/index.md) | 课程：[Dify / Coze 低代码 AI 平台](../topics/ai/lowcode/01-dify-coze.md)；场景：[从 Dify 原型到生产](../topics/ai/lowcode/scenarios/poc-to-production.md)；[面试题](../topics/ai/lowcode/interview.md) |
|
||
| [提示词与上下文工程](../topics/ai/prompt-engineering/index.md) | 课程：[提示词注入和防护](../topics/ai/prompt-engineering/02-context-injection.md)；场景：[读邮件的办公助手](../topics/ai/prompt-engineering/scenarios/injection-defense.md)；[面试题](../topics/ai/prompt-engineering/interview.md) |
|
||
| [Docker 与 Compose](../topics/devops/docker/index.md) | 课程：[Compose、网络和数据卷](../topics/devops/docker/02-compose-network-volume.md)；场景：[一条命令拉起开发环境](../topics/devops/docker/scenarios/compose-dev-env.md) |
|
||
| [Nginx](../topics/devops/nginx/index.md) | 场景：[AI 应用网关](../topics/devops/nginx/scenarios/ai-gateway-sse-canary.md) |
|
||
|
||
**跨技术栈场景**：[内容审核](../scenarios/content-moderation.md)、[客服知识库问答](../scenarios/customer-service-kb.md)、[多租户 SaaS](../scenarios/multi-tenant-saas.md)、[报表生成 Agent](../scenarios/report-agent.md)
|
||
|
||
<!-- week-topics:end -->
|