July 2026
从模型调用到生产级 Agent：Vibe Coding 之外的工程实践
从一次模型调用出发，讨论控制流、运行时、生产服务、可观测、评测与 AI 友好的开发体系
现在做一个 Agent Demo 已经很容易了。
准备一个 API Key，让编码 Agent 写几十行代码，模型就能对话、调用 Tool，甚至自己规划下一步。继续通过 Vibe Coding 追加需求，很快还能得到多步骤流程、状态保存、流式输出和一个看起来不错的界面。
真正困难的部分通常在 Demo 之后：执行中断了能否恢复，多个实例如何协调，线上问题能否还原，新版本是不是真的更好，以及半年后 AI 和人还能不能安全地维护这套系统。
过去一段时间，我参与开发过两种形态不同的 Agent：一个是基于 Python 和 LangGraph 的有状态长任务服务，另一个是基于 Go 和 Eino Compose 的短 Workflow。它们让我越来越确定一件事：
写出一个 Agent，可能只需要几十行代码；让它安全、稳定地运行几个月，需要的是完整的软件工程。
这篇文章不讨论如何训练模型，也不试图罗列所有 Agent 概念。它沿着一个 Agent 从 Idea 到生产系统的真实演进过程，讨论什么时候该使用框架、框架解决什么、框架之外还缺什么，以及 Vibe Coding 最容易替我们悄悄造出哪些不完整的轮子。
第一章：Agent 从一个函数开始
1.1 用 Python 完成一次模型调用
先不引入 Agent 框架。下面这段 Python 代码通过 OpenAI SDK 调用一个 OpenAI-compatible 模型服务，发送一条消息并返回模型生成的文本。服务可以直接来自模型厂商，也可以经过 LiteLLM 等统一网关。运行前配置 OPENAI_BASE_URL、OPENAI_API_KEY 和 OPENAI_MODEL；密钥只保存在环境变量或密钥系统中。
import os
from openai import OpenAI
client = OpenAI(
base_url=os.environ["OPENAI_BASE_URL"],
api_key=os.environ["OPENAI_API_KEY"],
)
def ask_ai(prompt: str) -> str:
response = client.chat.completions.create(
model=os.environ["OPENAI_MODEL"],
messages=[{"role": "user", "content": prompt}],
)
return response.choices[0].message.content or ""
print(ask_ai("用一句话解释什么是 Agent。"))
OpenAI SDK 负责发起请求，网关根据 model 将请求路由到实际的模型服务。调用链路很简单：
Python 函数 → OpenAI SDK → 模型网关 → 模型
文本进去，模型生成的文本出来，这就是一次完整的 AI 调用。从程序结构上看，ask_ai 仍然只是一个普通函数，模型是函数内部的远程依赖。
OpenAI SDK 也支持图像生成。改用图像生成接口并选择对应的图像模型，就可以用类似的方式生成图片。
1.2 当一个函数逐渐不够用
单次调用适合分类、摘要、改写和一次性生成。业务复杂后，一个函数往往会开始承担更多职责：
- 按照多个步骤理解、生成和校验。
- 根据中间结果选择不同分支。
- 调用搜索、数据库和其他外部服务。
- 保留中间状态，支持多轮交互。
- 在某个步骤失败后重试或恢复。
这些逻辑仍然可以继续写在普通函数里，但控制流、状态和错误边界会越来越难以管理。当一个函数演变成多步骤、有状态的程序时，就需要 Workflow 和 Agent 框架来显式组织它。
参考资料：OpenAI API Quickstart、LiteLLM Getting Started。
第二章：图执行与控制流选择
不要造轮子
完成一次模型调用后，如果在 Vibe Coding 中继续追加“再加一步”、“再加一个分支”、“保存状态”和“失败后恢复”，AI 很容易沿着当前函数继续堆代码，最后无意中重新实现一套不完整的 Workflow / Agent Runtime。当问题开始涉及多步骤、分支、状态和恢复时，应先识别它已经进入框架的能力边界，优先使用成熟框架。Python 项目可以考虑 LangGraph，Go 项目可以考虑 Eino；具体选择还要看团队技术栈、状态模型和运行时需求。
2.1 从多步骤函数到 Graph
业务变复杂后，一次模型调用会演变成多个步骤。例如，先生成文本，再判断是否需要图片：
def generate_content(user_input):
text = generate_text(user_input)
image = generate_image(text) if need_image(text) else None
return {"text": text, "image": image}
函数本身没有问题。但步骤、分支、循环和中间状态继续增加后，执行结构就会隐藏在嵌套的函数和条件语句中。
将同一段逻辑表示成图，控制流会更直接：
START → 生成文本 → 是否需要图片？
├─ 是 → 生成图片 → END
└─ 否 → END
图将程序拆成 Node，用 Edge 和 Branch 表达转移，并由 Runtime 推进执行。Node 仍然可以是普通函数，也可以调用模型或外部服务。
LangGraph 和 Eino Compose 都提供了这类图编排能力：
LangGraph StateGraph → add_node → add_edge / add_conditional_edges → compile
Eino Compose NewGraph → Add*Node → AddEdge / AddBranch → Compile
我在一个内容生成服务中使用的就是这种形态：用 Eino Compose 组织“生成文本 → 按需生图”的短 Workflow。
2.2 Workflow 与 ReAct
Graph 描述程序如何执行，但没有规定“下一步由谁决定”。Workflow 和 ReAct 可以看作控制权的两端，中间还有程序与模型共同控制的 Hybrid Agent：
| 模式 | 下一步由谁决定 | 适合场景 | 框架入口 |
|---|---|---|---|
| Workflow | 程序和预定义边 | 流程明确、结果可约束的业务 | LangGraph StateGraph / Eino Compose Graph |
| Hybrid Agent | 程序固定边界，模型在局部规划、路由或选择 Tool | 路径部分明确，但仍需要动态决策的任务 | LangGraph / Eino Compose 自定义 Graph |
| ReAct | 模型根据当前消息和 Tool 结果 | 路径难以预先枚举的开放任务 | LangChain create_agent / Eino ADK ChatModelAgent |
ReAct 的核心是一个模型与 Tool 之间的循环：
Model → 是否调用 Tool？
├─ 是 → Tool → 结果返回 Model
└─ 否 → END
LangChain create_agent 和 Eino ADK ChatModelAgent 封装了这个循环。开发者主要提供 Model、Tool 和运行选项，不需要自己编写每一条循环边。
ReAct 将大量控制权交给模型，因此也最容易超出业务预期。模型不仅可能偏离当前目标；连接文件系统、Shell 和网络等 Tool 后，意外决策还可能变成真实操作。
ReAct 的循环轮数和 Tool 调用路径也更难预估，因此 Token、延迟和成本的波动通常比固定 Workflow 更大。
Hybrid Agent 本质上仍是一类 Workflow，但没有唯一的结构。程序定义执行骨架和能力边界，模型在局部完成规划、路由或决策。Plan-and-Execute 是常见实现，也可以组合模型路由、结果评审和重试等模式。它比固定 Workflow 更灵活，又比 ReAct 更受约束，执行过程也更容易观测、测试和控制。
Workflow、Hybrid Agent 和 ReAct 不是从低到高的成熟度阶段，而是不同的控制流选择。主要区别是业务路径能否提前定义，以及模型可以决定多少执行过程。
下面两个案例展示了这种不可预期性：客服 Agent 偏离当前订单场景，主动延续了历史 RAG 话题；通用 Agent 则借助文件系统和执行工具，自主搭建了游戏服务器与内网穿透。它们既体现了 ReAct 的通用性，也说明需要限制 Tool 权限、工作目录、执行预算和人工确认边界。
2.3 LangGraph 与 Eino 的框架分层
以 LangGraph 为底层的 Python 生态和 Eino 都覆盖了图编排与高层 Agent 模式，但分层方式不完全相同。可以先建立一个近似映射：
| 抽象层 | Python 主线 | Go 主线 |
|---|---|---|
| 自定义 Graph / Workflow | LangGraph | Eino Compose |
| 高层 ReAct Agent | LangChain create_agent |
Eino ADK ChatModelAgent |
LangGraph 是 State-first 的图运行时，重点是共享 State、状态转移和可持久执行。LangChain create_agent 在它之上预制了模型与 Tool 的循环。
Eino Compose 更强调 Go 类型化 Input / Output 和组件数据流；Eino ADK 则提供 ChatModelAgent、Runner 和常见 Agent 协作模式。Compose 与 ADK 可以组合，但不是同一套图 API 的上下层封装。
本文后续会继续对照两种实现：LangGraph 的有状态 Graph 与自建 Run Service，以及 Eino Compose 的类型化短 Workflow。
参考资料：LangGraph Graph API、LangChain Agents、Eino Overview、Eino: Agent or Graph?。
第三章：Agent 代码设计
3.1 控制流：Node、Edge、Branch 与 Loop
框架首先解决的是“哪一步在什么时候执行”。
| 概念 | 解决的问题 | LangGraph | Eino Compose |
|---|---|---|---|
| Node | 将复杂任务拆成可独立执行的步骤 | 普通 Python 函数 + add_node |
Component / Lambda + Add*Node |
| Edge | 表达确定的执行顺序 | add_edge |
AddEdge |
| Branch | 在运行时选择下一步 | add_conditional_edges / Command |
AddBranch |
| Loop | 重复执行，直到满足退出条件 | 条件边返回前序 Node | Branch 返回前序 Node，形成有向环 |
下面这段经过简化的 Eino Compose 代码直接对应了这些概念：
func BuildGraph(ctx context.Context) (compose.Runnable[*GenerateInput, *GenerateOutput], error) {
g := compose.NewGraph[*GenerateInput, *GenerateOutput]()
textNode := NewTextNode()
imageNode := NewImageNode()
if err := g.AddLambdaNode("text", compose.InvokableLambda(textNode.Invoke)); err != nil {
return nil, err
}
if err := g.AddLambdaNode("image", compose.InvokableLambda(imageNode.Invoke)); err != nil {
return nil, err
}
if err := g.AddEdge(compose.START, "text"); err != nil {
return nil, err
}
if err := g.AddBranch("text", newImageBranch()); err != nil {
return nil, err
}
if err := g.AddEdge("image", compose.END); err != nil {
return nil, err
}
return g.Compile(ctx, compose.WithGraphName("content_generation"))
}
AddLambdaNode 注册文本和图片两个 Node，AddEdge 定义固定顺序，AddBranch 根据文本结果决定进入生图节点还是直接结束，最后 Compile 将图变成可执行的 Runnable。
Node 不需要对应每个小函数。适合成为 Node 的通常是一个有明确输入输出，并且需要独立观测、测试、重试或路由的业务步骤。
分支应当尽量只负责路由，不要在路由函数里同时完成大量业务逻辑。循环则必须有明确的退出条件和步数上限，避免模型决策导致无限执行。
在另一个长任务 Agent 中，我把 Planner、Coordinator、Capability 和等待人工输入等步骤显式建模为 LangGraph Node。Graph 变复杂了，但每一步的责任和实际执行路径反而更清楚。
3.2 数据流：State、Input / Output 与 Message
控制流决定“执行哪一步”，数据流决定“这一步能看到什么”。
| 框架 | 核心数据模型 | Node 执行 | Fan-in |
|---|---|---|---|
| LangGraph | 共享 State | 读取 State，返回 Partial State Update | 并行 Node 的 Update 由 Reducer 聚合到 State，下游再读取聚合后的 State |
| Eino Compose | 类型化 Input / Output | 上游 Output 沿 Edge 传给下游 Input，Compile 时检查类型对齐 | 多个 Output 通过 Map Merge 或 Field Mapping 组成下游 Input |
LangGraph 虽然可以定义 Graph 的 Input / Output Schema，但它们只是对外调用边界：Input 初始化 State，Output 从最终 State 中筛选字段。Graph 内部仍然通过 State Update 传递数据，Edge 不传递独立的 Node Output。
Eino 的 Fan-in 需要同时解决“何时执行”和“如何合并”。确定性 DAG 可使用 AllPredecessor，等待所有前驱完成；上游通过 WithOutputKey 生成不同 Map Key，由框架合并后传给下游。使用 Eino Workflow 时，还可以把多个上游字段显式映射到下游 Struct。
Message 表达模型对话、Tool Call 和 Tool Result。LangGraph 常用 MessagesState 和 Message Reducer 管理，Eino 使用 schema.Message 或 ADK []*schema.Message。调用模型前，只应从当前数据中选择必要信息组装成 Message，不要把全部 State 直接放进 Prompt。
参考资料：LangGraph Graph API、Eino 编排设计理念。
3.3 运行时设施：Checkpoint、Interrupt、Stream 与 Callback
Graph Runtime 除了推进 Node，还需要处理执行快照、暂停恢复、过程输出和扩展钩子。
| 概念 | 解决的问题 | LangGraph | Eino |
|---|---|---|---|
| Checkpoint | 保存可恢复的执行快照 | Checkpointer 按 Thread 和 Super-step 保存 State | Compose / ADK 提供 CheckPointStore 接口，存储由业务实现 |
| Interrupt | 暂停执行，等待确认或补充输入 | interrupt() + Command(resume=...) |
Compose Interrupt / ADK Interrupt + Resume |
| Stream | 在最终结果之前持续输出文本片段、State 和业务事件 | stream / astream，支持 values、updates、messages 和 custom |
Compose Stream；ADK AgentEvent |
| Callback | 统一注入日志、Trace、Metric 和用量统计 | Runnable Callback / LangSmith Tracing | callbacks.Handler，支持 Start、End、Error 和 Stream Hook |
Checkpoint 不是 Memory。它保存的是执行快照，而不是经过挑选、适合未来检索的长期信息。
Checkpoint 也不能自动保证副作用只执行一次。Interrupt 或失败恢复时，当前 Node 可能被重新执行，所以外部写操作仍然需要幂等保护。
这些能力解决的仍是 Graph / Agent 一次执行内的问题。Run 元数据、后台任务、队列、Worker 和多实例控制属于更外层的 Agent Service，是下一章的主题。
3.4 为什么要使用框架
LangGraph 和 Eino 的核心图运行时并不只能编排 AI。Node 可以调用模型，也可以执行普通程序；框架负责的是节点调度、状态推进和暂停恢复等通用 Workflow 能力。
框架的价值不只是把函数组织成图，而是让控制流、数据流、Tool、Checkpoint、Interrupt、Stream 和 Callback 运行在同一套 Runtime 中。这些设施共享执行语义和扩展点，组合起来也更稳定。
控制流显式化后，也更容易生成 Graph 可视化、查看实际执行路径，并围绕 Node 做局部调试。
完成第一章的最小函数后，如果继续通过 Vibe Coding 逐项增加循环、状态、事件、暂停恢复和流式输出，AI 通常会沿着现有代码直接补实现。功能可能暂时可用，但很容易无意中重新实现一套不完整的 Agent Runtime。
| 层级 | 主要实现内容 |
|---|---|
| Framework Runtime | Graph 调度、状态推进、Tool 执行、Stream、Callback、Checkpoint 和 Interrupt / Resume |
| Agent 业务代码 | Input / State Schema、Node、Tool、路由规则和业务策略 |
| Agent Service | Run 生命周期、API、权限、Queue / Worker 和集群运行 |
尽量使用框架已有的能力。准备自己实现一项设施前，先了解框架是否已经提供，以及如何使用和扩展；确认不能满足需求后，再决定是否自行实现。
参考资料：LangGraph Graph API、LangGraph Persistence、LangGraph Interrupts、LangGraph Streaming、Eino Orchestration Design Principles、Eino Stream Programming Essentials、Eino Callback User Manual、Eino Interrupt & CheckPoint。
第四章：从 Agent 程序到生产服务
4.1 Thread、Run 与暂停恢复
一个 Graph 能在本地运行，不等于已经是生产 Agent 服务。这里先使用两个概念：
| 概念 | 含义 |
|---|---|
| Thread | 跨多个 Run 保存历史和状态的持久容器 |
| Run | 从 Input 或 Resume 开始，到 Completed、Failed、Cancelled 或 Interrupted 的有限执行 |
Thread
├─ Run 1 → Completed
├─ Run 2 → Completed
└─ Run 3 → Interrupted
└─ Resume → Run 4 → Completed
普通多轮对话通常由一个 Thread 串联多个 Run。每次用户输入启动一个 Run，Run 结束后释放 Worker，历史和状态由 Thread 保留。
Checkpoint 保存执行快照，Interrupt 暂停执行，Resume 从快照继续：
Run 执行
→ Interrupt
→ 保存 Checkpoint，释放 Worker
→ 等待用户确认或补充输入
→ Resume
→ 从 Checkpoint 继续
LangGraph 使用 thread_id 和 Checkpointer 保存状态。Eino 支持 Checkpoint 和 Interrupt，但 CheckPointStore、对话历史和 Thread 需要业务持久化。普通对话不需要用 Interrupt 表示每轮结束，它主要用于审批、缺失参数和人工决策。
读取 Thread 历史与状态
→ 组装本轮 Context / Messages
→ 执行一次 Runner / Graph
→ 保存消息和结果
→ 结束本轮 Run
4.2 Run Service：状态与操作
在一个基于 LangGraph OSS 的长任务 Agent 中，我在 Graph Runtime 之外实现了一层 Run Service：
API → RunService → Background Task → LangGraph Runtime
| 职责 | 一种实现方式 |
|---|---|
| Run 元数据 | PostgreSQL RunRecord 保存状态、Metadata 和更新时间 |
| 生命周期 | 状态机管理 Running、Interrupted、Completed、Failed、Cancelled |
| Graph 状态 | PostgreSQL Checkpointer 保存 LangGraph 执行快照 |
| 服务操作 | Start、Resume、Cancel、Recover、Rollback、Fork 和 Timeline |
Run Service 不执行具体 Node，而是管理一次执行的生命周期，再将任务交给 LangGraph Runtime。
4.3 Queue、Worker 与集群控制
这个服务当前通过 Pod 内的 asyncio.Task 执行 Run。集群中的 Cancel 请求不一定落到持有 Task 的 Pod，因此使用 Redis Pub/Sub 广播控制信号，再由目标 Pod 取消本地任务：
Cancel API → Redis Pub/Sub → 目标 Pod → Local Task
PostgreSQL CAS 防止多个实例同时 Resume，Checkpoint 支持故障恢复；Redis Stream 推送实时事件，PostgreSQL Event 保存历史记录。Checkpoint 外部化后，Run 可以在一个 Pod 执行，再由另一个 Pod 恢复；Worker 可以持有执行期间的临时状态，但不应持有无法恢复的唯一状态。
状态外部化不等于把每个 Node 拆成独立微服务。Agent 的 Graph 会随业务频繁调整，过度拆分会增加接口和部署成本，也会降低 Workflow 的迭代灵活性。
完整的异步执行服务通常会进一步拆分：
Client
→ API Server / Run Service
→ Run Record / Durable Queue
→ Worker
→ Eino / LangGraph Runtime
→ Event / Checkpoint / Result
| 层级 | 主要职责 |
|---|---|
| Framework Runtime | Node 调度、State 推进、Tool 执行、Stream、Checkpoint、Interrupt / Resume |
| Run Service | Run 元数据和状态机、API、Cancel、Recover、查询与 Timeline |
| Queue / Worker | 后台执行、并发上限、优先级、租约、心跳、故障接管和积压治理 |
这套实现已经支持跨实例控制，但还没有 Durable Queue；Redis Stream 在这里承载过程事件，不是 Run Queue。任务规模继续增长后，可以再引入 MQ 和独立 Worker。LangGraph Agent Server 已经提供类似设施；只使用 LangGraph OSS 或 Eino 时，则需要在框架之外建设。
另一个同步、短生命周期的 Eino Workflow 则不需要这套外层设施。生产能力也应该按真实复杂度增加，不必让每个 Agent 从第一天就拥有同一套架构。
参考资料：LangGraph Persistence、LangGraph Agent Server、Agent Server Runs、Agent Server Threads、Eino Memory and Session。
第五章：可观测与故障排查
5.1 Log、Metric 与 Trace
Agent 的一次运行可能包含多个 Node、Model 和 Tool 调用。只记录最终输出，无法说明中间发生了什么。
| 信号 | 简单理解 | 主要用途 |
|---|---|---|
| Log | 每一次具体操作留下的明细记录 | 查看某次 Run 的参数、返回值、状态和错误 |
| Metric | 按时间聚合的数值 | 观察成功率、延迟、Token、成本和队列积压 |
| Trace | 由多个 Span 组成的一次请求调用树 | 查看父子调用、耗时和失败位置 |
结构化 Log 可以使用稳定的 event_type 表达关键生命周期，例如 step_failed 和 model_completed。这类 Log 适合重建 Run Timeline；调试文本、连接池信息和错误栈则是普通 Log。
小提示：结构化日志。 字段应作为键值参数传入，不要先拼成一整句文本，这样 SLS 才能直接按 run_id、model 等字段查询。
logger.InfoContext(ctx, "model_completed",
slog.String("run_id", runID),
slog.String("model", model),
slog.Int64("duration_ms", elapsed.Milliseconds()),
)
logger = structlog.get_logger()
structlog.contextvars.bind_contextvars(run_id=run_id)
logger.info("model_completed", model=model, duration_ms=elapsed_ms)
Go 可以通过 context.Context 传递链路信息，并由结构化日志组件统一补充公共字段；Python 可以在请求或 Run 入口绑定 contextvars，让后续 Log 自动携带相同上下文。
以一次模型调用完成为例，同一件事可以同时产生三类信号：
Log → 写入 model_completed，记录 run_id、model、Token、耗时和返回摘要
Metric → 调用次数 +1，并更新耗时和 Token 统计
Trace → 结束 Model Span，并与上层 Node Span 关联
在 LLM 领域，最值得优先落地的是两份具体记录：
| 领域记录 | 归类 | 主要价值 |
|---|---|---|
| Run Trace | Trace：以 Run 为根 Span，下面挂载 Node、Model 和 Tool Span | 输入 run_id 即可还原执行过程，快速定位问题 |
| LLM Call Record | 结构化 Log；在 Trace 中同时对应一个 Model Span | 分析 Token 和成本，支持 Replay、评测和数据沉淀 |
LLM Call Record 在可观测语境中可以算作 Log，但它比普通调试日志更稳定，更像一份调用明细或审计记录。它包含模型、参数、Message、Tool Call、返回值、Token、耗时和请求 ID。
Log 描述的是逐条明细记录，不代表数据只能保存到日志平台。LLM Call 元数据可以进入 Trace 和 SLS 用于在线排查；如果要长期用于 Replay、评测或训练，则可以用 llm_call 表保存索引和元数据，用对象存储或数据湖保存完整 Message 和返回值。这些数据不应依赖 Log 或 Trace 的采样和保留周期；经过授权、脱敏、质量筛选和去重后，可以沉淀为评测集或训练样本。
我在一个生产内容服务中将模型调用明细先写入数据库，再定期同步到数据仓库。这样不仅能排查单个失败 Case，也能分析线上结果分布。例如，更换模型后固定 Benchmark 可能全部通过，但线上文本却整体变短；保留 Prompt、返回值、Token、耗时和业务结果后，才能发现这种分布漂移。
结构化 Log 也可以作为数据收集入口，由 Consumer 将不同类型的记录分别写入 SLS、Metric 系统或独立存储。
Eino 可以通过 callbacks.Handler 在组件的 Start、End、Error 和 Stream 阶段统一采集信号，例如统计具名 Lambda Node 的调用量和耗时。LangGraph / LangChain 也可以通过 LangSmith 或 OpenTelemetry 记录 Trace。
Prompt、Message、Tool 参数和返回值可能包含隐私数据。默认应当记录元数据，对原文做脱敏、采样和权限控制。
5.2 Run Timeline 与全链路 ID
结构化 Log 和 Trace 需要共用关联 ID，才能还原一次完整执行。Metric 则使用服务、Workflow、Node 和模型等公共维度对齐整体趋势。
thread_id
└─ run_id
├─ step_id
│ └─ attempt_id
├─ model_call_id → provider_request_id
└─ tool_call_id
run_id ↔ trace_id
step / call ↔ span_id
run_id 是业务排查的主入口，trace_id 是可观测系统生成的技术链路。两者需要互相记录，但不需要合并成同一个概念。
一条最小可用的 Run Timeline 至少覆盖以下结构化日志：
run_started
step_started / step_completed / step_failed
model_started / model_completed
tool_started / tool_completed
run_interrupted / run_resumed
run_completed / run_failed / run_cancelled
每条日志记录时间、类型、状态、关联 ID 和必要摘要。排查时先按时间重建 Timeline，定位第一个异常记录，避免把后续的连锁失败当成根因。
5.3 SLS 自动查询、Replay 与根因定位
故障排查应当从 Run 记录开始，SLS 用于查找详细证据，不用日志反推业务状态。
输入 run_id
→ 查询 Run 状态
→ 重建 Timeline
→ 定位第一个异常 Step
→ 根据 step_id / call_id 查询 SLS
→ Replay
→ 输出根因和证据
SLS 查询可以做成可复用的排查工具：根据服务、环境、Run 时间范围和关联 ID 生成字段查询，再逐层下钻到 Step、Model 或 Tool。查询模板应当限定时间范围和返回数量，不让模型无约束地搜索整个 Logstore。
Replay 依赖前面保留的 LLM Call Record，用来回答“相同输入在另一个环境中会发生什么”：
| 方式 | 做法 | 用途 |
|---|---|---|
| 结果回放 | 使用已记录的 Model / Tool 返回值继续执行 | 隔离排查解析、Node 和 Workflow 问题 |
| 请求重放 | 使用原始 Message、Tool Schema、模型和参数重新请求 | 比较模型、Prompt、代码或服务商的差异 |
Checkpoint 还可以固定上游状态，只重跑目标 Node，避免每次都从 Graph 起点执行。对于概率性问题，可以对同一输入重复请求重放：如果经常失败，应当作为稳定问题处理；如果极难复现，则需要权衡是否值得为低频边界 Case 增加 Prompt 和流程复杂度。
请求重放不保证结果完全一致。有外部副作用的 Tool 应当使用 Mock、沙箱或幂等键，避免重复写入真实系统。
一份有用的根因结果只需要说清四件事：第一个失败点、相关证据、影响范围和修复建议。如果证据不足，应当保留未知项，不由模型补全根因。
5.4 Go 与 Python 的可观测技术栈
Log、Trace 和 Metric 不需要强制使用同一套 SDK。需要统一的是字段语义和关联 ID，例如 run_id、step_id、model_call_id 和 trace_id。
| 能力 | Go | Python |
|---|---|---|
| Log | log/slog 或 Zap |
structlog + 标准 logging |
| Trace | OpenTelemetry、otelgin、otelhttp、otelgorm；Eino Callback 补充 Agent Span |
OpenTelemetry SDK 及 FastAPI / HTTPX Instrumentation；LangSmith 可选 |
| Metric | prometheus/client_golang |
prometheus-client |
在实际项目中，这几套系统很容易分别接入，却没有形成统一语义。更完整的做法是以 Framework Callback 为 Agent 采集入口，将结构化 Log、OpenTelemetry Span 和 Prometheus Metric 与同一组链路 ID 关联起来。
Framework Callback
├─ Structured Log → SLS / 独立存储
├─ OpenTelemetry Trace → ARMS
└─ Prometheus Metric → 监控与告警
可观测解决“这次 Run 发生了什么”，不直接回答“Agent 是否真的变好了”。后者需要将 Run 产物进入测试和评测闭环。
参考资料：LangSmith Observability、LangSmith Observability Concepts、Eino Callback & Trace、OpenTelemetry Go、OpenTelemetry Python、OpenTelemetry Semantic Conventions、Go slog、structlog、Prometheus Go Client、Prometheus Python Client、SLS 查询语法。
第六章：如何判断 Agent 真的变好了
6.1 Test：Agent 程序是否正确
Test 检查程序是否满足明确约束。测试失败时，代码不应发布。
Agent 的自然语言输出存在不确定性，测试不应断言它必须生成某句固定文本，而应断言稳定的程序事实：
- 输入输出是否符合 Schema。
- 是否进入正确的 Branch，选择正确的 Tool。
- Tool 参数、State Update 和错误类型是否正确。
- Loop 是否在步数上限内结束。
- Interrupt / Resume、Retry 和 Checkpoint 是否符合预期。
| 层级 | 测试对象 | 常见做法 |
|---|---|---|
| Node Test | 单个 Node、路由函数、Parser 或 Tool | 直接调用，使用 Fake Model / Tool，断言契约和不变量 |
| Graph Test | 多个 Node 的路径、State 和恢复语义 | 使用内存 Checkpointer 和固定模型返回，调用编译后的 Graph |
| Scenario Test | 一个完整用户场景 | 从用户输入运行到最终状态，检查结果、关键路径和副作用 |
Scenario Test 和 Eval 可以使用相同的案例。前者检查必须通过的硬性条件，后者对质量进行评分和对比。
Frozen Response Replay 可以将历史 Model / Tool 返回值注入测试，让 Node 和 Graph Test 可以稳定重复，同时不产生真实调用成本。只有需要验证模型、凭证、Schema 和网络集成时，才使用真实模型运行 Integration Test。
LangGraph 编译后的 Graph 可以整体调用，也可以单独调用 graph.nodes 中的 Node。Eino 中底层 Component / Lambda 直接用 go test 测试，编译后的 Runnable 用于 Graph Test。
6.2 Eval：Agent 结果是否足够好
Test 通过只能说明 Agent 没有违反明确约束。Eval 需要先定义“好”的标准，再在一批固定案例上测量质量。
不要让测试 Case 和质量标准只存在于产品或开发人员的脑中。至少将 Input 固化为文件或 Dataset，将期望的结果写成 Reference、Rule 或 Rubric，AI 才能自动批量执行并比较结果。
Dataset 是评测的基础。它可以先从少量人工案例、真实用户场景和历史失败开始：
| 字段 | 内容 |
|---|---|
| Input | 用户输入、必要 Context 和运行环境 |
| Reference | 参考结果、必须满足的条件或预期路径，可以为空 |
| Metadata | 场景、难度、用户群、语言和失败类型等分组标签 |
Agent 不只需要评价最终文本：
| 维度 | 回答的问题 | 示例 |
|---|---|---|
| Outcome | 最终任务是否成功 | 输出是否完整，外部系统是否真的产生了目标结果 |
| Process | 执行过程是否合理 | Tool 选择、参数、调用顺序、Retry 和停止时机 |
| Performance | 付出的代价是否可接受 | 延迟、Token、调用次数和成本 |
一个 Case 可以同时使用多种 Evaluator：
| 方法 | 适合的问题 | 特点 |
|---|---|---|
| Code / Rule | Schema、必填字段、Tool 路径、禁止词、延迟和成本阈值 | 便宜、稳定，应当优先使用 |
| LLM Judge | 正确性、完整性、指令遵循、风格和路径合理性 | 能处理语义问题，但结果也存在偏差和波动 |
| Human | 主观质量、高风险结果和 Judge 难以决定的案例 | 最接近业务判断，但成本高，适合抽样和校准 |
LLM Judge 需要明确 Rubric，固定 Judge Model 和版本，并输出分数与理由。比较两个 Release 时，可以隐去版本信息做 Pairwise 判断，再定期用人评检查 Judge 是否仍与业务标准一致。
同一 Case 的结果可能每次不同。重要场景需要重复运行，同时看平均质量和成功的稳定性，不用单次得分代表整个 Agent。
6.3 Release：新版本是否真正更好
只记录代码 Commit 不足以复现 Agent 行为。一个 Agent Release 至少需要确定 Graph / Agent 代码、Prompt、Model 与参数、Tool Schema、Policy 和运行配置。
将某个 Release 在一个 Dataset 上完整运行一遍，得到一次 Experiment。新旧版本应当使用相同 Dataset、运行条件和 Evaluator：
Dataset
├─ Baseline Release → Experiment A
└─ Candidate Release → Experiment B
↓
逐 Case 对比质量、成本和延迟
每条评测结果至少关联 case_id、release_id、run_id 和 evaluator_version，这样一个分数才能追溯到具体输入、Agent 行为和评分标准。
发布门禁不应只看总平均分，还要检查：
- Test 全部通过，关键场景没有回归。
- 新版本在目标质量上更好，或者至少不更差。
- Token、成本和延迟没有超出可接受范围。
- 失败 Case 已经按场景和根因分类，而不是被平均分掩盖。
实际发布可以分成三层验证：固定 Case 的离线评测检查已知能力，线上流量的影子回放检查模型、Prompt 或代码是否明显劣化，灰度或 A/B 再判断真实业务指标是否改善。
离线 Eval 通过后，仍需要通过灰度或 A/B 观察真实用户的任务成功、反馈、延迟和成本。线上失败、差评和新出现的边界场景再进入 Dataset，成为下一版的 Regression Set：
线上 Run / 用户反馈
→ 筛选典型成功与失败 Case
→ 加入 Dataset
→ 修复并产生 Candidate Release
→ 离线 Experiment
→ 灰度 / A/B
→ 继续收集线上信号
参考资料：LangGraph Test、LangChain Agent Test、LangSmith Evaluation Concepts、LangSmith: Evaluate a Complex Agent、OpenAI: Working with Evals、Anthropic: Demystifying Evals for AI Agents、Eino Overview。
第七章：Agent 开发指南
7.1 构建一个 AI 友好的仓库
AI 友好的仓库不只让 AI 能读懂代码，还要让它能找到正确流程、执行常用操作，并验证执行结果。原则上，仓库内的开发、测试、排查和运维操作都应该能由 AI 完成，人只需要描述目标、查看结果并做必要决策。
| 内容 | 作用 |
|---|---|
AGENTS.md / CLAUDE.md |
说明仓库结构、开发规则、权限边界和验证命令 |
ARCHITECTURE.md |
作为经过人工确认的系统设计入口，说明核心边界、数据流和关键决策 |
| Skills | 将启动、开发、测试、排障和 Replay 等任务固化为可重复工作流 |
| API / CLI / Script | 将手动点击和临时命令变成稳定的可执行入口 |
| Test / Log / Timeline / Metric | 让 AI 根据证据判断操作是否成功 |
一个 Skill 至少说清何时使用、需要哪些前置条件、执行什么命令、如何验证，以及失败后如何排查。复杂命令尽量封装成脚本，Skill 负责选择和组织流程。对于结果难以用代码直接断言的任务，可以由脚本负责执行、Skill 提供验收规则，再由 AI 判断结果。
文档需要有明确索引和可信入口。核心架构文档应由人确认准确性；面向具体操作的详细文档可以主要服务于 AI，但代码和流程变化后必须同步更新。仓库外还可以提供 Bootstrap 指引，先完成 Git、运行环境、权限和 Clone，再由仓库内的 Skills 接管后续操作。
我在自己的 Agent 仓库里建立了这类体系：开发 Skill 覆盖环境启动、API、测试、Benchmark、故障注入、Metric 和排障；Replay Skill 负责模型调用重放；诊断 Skill 将真实 Case 排查收敛成固定步骤。
配套的 Dashboard 是面向人的结果查看入口：可以按 run_id 搜索 Run，查看状态、实时 Event、Timeline、Checkpoint 和 Fork Tree，也可以触发 Cancel、Recover 和 Rollback。这些能力同时通过 Agent API 暴露，Dashboard 只是可视化客户端；AI 可以调用同一套 API 完成操作，人负责查看过程和结果。
用户提出任务
→ AI 选择 Skill
→ 调用 API / CLI / Script
→ 获取 run_id 和运行结果
→ 查询 Timeline / Log / Metric
→ 返回结论与证据
Skills 也是仓库的一部分。API、脚本和开发流程变化时，对应 Skill 需要同步更新和验证。
7.2 从一个 Idea 开始
Agent 开发通常从一个 Idea 开始。先确定想解决的问题和最小验证方式，不在第一天建设完整的 Agent 平台。
选择一个模型服务或 OpenAI-compatible 网关，配置 OPENAI_BASE_URL、OPENAI_API_KEY 和 OPENAI_MODEL，先使用第一章的 OpenAI SDK 代码跑通最小 Demo。Key 只保存在本地环境或密钥系统，不写入代码和文档。
Demo 的复杂度真实增长后，再按照本文的章节逐步补充能力：
| 出现的需求 | 对应章节 |
|---|---|
| 验证 Idea，完成第一次模型调用 | 第一章：一个普通函数 |
| 出现多步骤、分支、Tool 和状态 | 第二、三章：使用 Eino 或 LangGraph 组织代码 |
| 出现多轮、异步长任务、恢复或集群运行 | 第四章：建设 Agent Service |
| 线上问题难以定位 | 第五章：补充 Log、Metric、Trace 和排查工具 |
| 无法证明新版本真的更好 | 第六章：建立 Test、Eval 和 Release 闭环 |
| 希望 AI 参与后续开发和维护 | 第七章：补充仓库规则、Skills 和可执行接口 |
Idea
→ Model API Key
→ 最小 Demo
→ Eino / LangGraph
→ 生产服务
→ 可观测与排查
→ 评测与持续改进
→ AI 可操作的开发体系
原则是先验证 Idea，再为真实出现的复杂度补充工程能力；不为了做 Demo 提前建设所有设施，也不在已经进入框架能力边界后继续造轮子。
结语
Vibe Coding 大幅降低了写代码的成本，但没有消除系统设计。相反，当 AI 可以快速添加循环、状态、Tool 和后台任务时，我们更容易在没有意识到的情况下造出一套新的 Runtime。
我的做法是从最小模型调用开始，只在复杂度真实出现时增加下一层能力：先选择控制流，再依托成熟框架组织执行；需要长期运行时建设 Run Service；上线前补齐可观测和评测；最后把开发、测试和排查也做成 AI 可执行的工作流。
一个生产级 Agent 不一定需要本文列出的全部设施，但应该能够回答几个基本问题：
- 谁决定下一步，模型的控制边界在哪里？
- 执行中断后，状态是否可以恢复？
- 输入一个
run_id，能否还原它真正做过什么？ - 更换模型、Prompt 或 Graph 后，如何证明结果更好了？
- AI 能否自己完成开发、验证和排查，并把证据交给人？
Agent 最特殊的部分当然是模型。但真正决定它能否进入生产的，仍然是这些朴素的软件工程问题。