事件接口：把执行过程变成可消费的事件流 · 生产级 Agent

假设你手上已经有一个最小可用的 Agent：拿到任务后循环问模型下一步做什么，模型要调工具就执行工具，把结果塞回去继续问，直到模型给出最终回答。它能跑，但过程全靠 print 打在终端上。这篇笔记讲一个改动极小、影响极大的重构——让主循环不再打印，而是逐步产出结构化事件。改完之后，同一套主循环可以同时喂给命令行、Web 前端和自动化测试，而且调试能力会上一个台阶。

为什么 print 撑不起工程演进

print 在开发初期是最顺手的观察手段，但它有一个致命特性：它把"发生了什么"和"怎么展示"焊死在了一起。

具体表现为三个问题：

只有一个消费者。 打印出来的字符只存在于终端。想加一个 Web 界面实时展示执行过程？想在 CI 里断言"这个任务确实调用了某个工具"？拿不到数据——过程信息在打印的那一刻就被消费掉了，没有第二个人能用。
展示逻辑侵入核心逻辑。 每多一种展示需求（换个格式、加个颜色、只看工具调用），都得回去改 Agent 主循环。主循环本该只关心"任务怎么推进"，却越来越像一个排版函数。
过程即丢弃。 打印完就没了，不能落盘、不能回放、不能统计。线上出了问题，你只能靠用户复述"它当时好像说了什么"。

结论：终端输出适合验证原型，但一旦要往产品方向演进，就必须把"过程数据"从"展示方式"里剥出来。

关键改动：主循环变成事件生成器

改法只有一句话：把主循环里所有 print(...) 换成 yield {"type": ..., ...}。

Python 里一个函数只要包含 yield，就变成生成器函数。它的执行模式从"一次跑完、返回结果"变成"跑到 yield 就暂停，把值交给调用方，调用方处理完再恢复执行"。用一个极简示意感受一下：

def demo_stream():
    yield {"type": "token", "content": "让我查一下清单。"}
    yield {"type": "tool_start", "name": "list_todos", "args": {}}
    yield {"type": "tool_result", "name": "list_todos", "output": "[1] 交周报", "is_error": False}
    yield {"type": "done"}

for event in demo_stream():
    handle(event)   # 每产出一个事件，这里就被调用一次

注意执行顺序是交替的：生成器跑到第一个 yield → 暂停 → 外层 for 拿到事件并处理 → 生成器从暂停点继续 → 下一个 yield。不是等函数跑完才开始消费。

这一下就把角色切开了：

Agent 主循环是事件生产者。 它只负责在正确的时机产出正确的事件，完全不关心事件最终被谁、以什么形式消费。
消费者可以有任意多个形态。 CLI 拿到事件逐行渲染；Web 服务把每个事件序列化成 SSE 推给浏览器；测试代码把事件收进列表逐条断言。三者共用同一个主循环，主循环一行不用改。

以后想做中断、续跑、事件持久化，也都是在消费端或事件层做文章，核心循环保持稳定。

六种核心事件

事件不需要一开始就设计得很全，先覆盖一次执行中最关键的节点。以下六种是最小可用集：

事件	何时产出	主要用途
`iteration_start`	每轮模型调用前	监控轮数。简单任务跑到第 8 轮还没停，基本可以断定循环出了问题
`token`	模型流式吐出一小段文本时	实时展示模型正在说什么，CLI 直接打印，Web 推给前端
`tool_start`	工具执行前	暴露模型的"意图"：选了哪个工具、参数长什么样。工具选错、参数编错都在这一步现形
`tool_result`	工具执行后	携带输出和 `is_error` 标志，通过 `call_id` 与对应的 `tool_start` 配对
`done`	模型不再调用工具、给出最终回答后	标记任务正常收尾
`error`	系统级故障时	模型 API 调不通、超过最大轮数等，任务被迫终止

其中有一条边界值得单独强调：工具失败不等于系统错误。

read_file 找不到文件、complete_todo 传了不存在的编号——这类失败应该包装进 tool_result 并把 is_error 置为 True，然后照常把错误文本写回模型上下文。模型看到错误信息后往往能自己纠偏：换个参数重试，或者直接告诉用户"这条不存在"。
只有当循环本身无法继续时才发 error 事件：比如模型接口抛异常、跑满轮数上限还没收敛。这时任务直接结束。

判断标准一句话：模型还有机会自救的，走 tool_result；模型已经没有下一轮的，走 error。

两条线：messages 给模型看，事件给外部看

改造过程中最容易混淆的一点：主循环里同时维护着两套"记录"，它们服务的对象完全不同。

messages 是模型的上下文。 每轮 assistant 的输出（含 tool_calls）、每个工具的执行结果，都必须按 API 协议原样追加回去，否则模型下一轮就"失忆"，甚至直接被接口拒绝（缺少与 tool_calls 配对的 tool 消息是常见的 400 错误来源）。
事件是外部世界的观测流。 它不参与模型推理，只供 CLI、Web、测试、日志系统消费。

所以代码里会出现"同一个动作干两件事"的模式：工具跑完之后，既要 yield 一个 tool_result 事件（给外部），又要 messages.append 一条 tool 消息（给模型）。删掉任何一边，另一条线就断了——少了 yield 外面看不到过程，少了 append 模型不知道结果。

代码模式

下面用一个待办清单 Agent 完整走一遍。模型接口统一走 OpenAI 兼容协议，base_url 可配置，DeepSeek、Kimi、GLM 等都能直接换：

export LLM_API_KEY=sk-xxx
export LLM_BASE_URL=https://api.deepseek.com   # 换厂商只改这两行
export LLM_MODEL=deepseek-chat

工具模块

工具级失败用一个专门的异常类型表达，让调度层能把它和真正的程序 bug 区分开：

# agent/toolkit.py
import json
from pathlib import Path

STORE = Path(__file__).parent / "todos.json"


class ToolError(Exception):
    """工具级失败：写进 tool_result 交回模型，不中断任务。"""


def _load() -> list[dict]:
    if not STORE.exists():
        return []
    return json.loads(STORE.read_text(encoding="utf-8"))


def _save(items: list[dict]) -> None:
    STORE.write_text(json.dumps(items, ensure_ascii=False, indent=2), encoding="utf-8")


def list_todos() -> str:
    items = _load()
    if not items:
        return "清单目前是空的"
    return "\n".join(
        f"[{it['id']}]{'✓' if it['done'] else ' '} {it['title']}" for it in items
    )


def add_todo(title: str) -> str:
    items = _load()
    new_id = max((it["id"] for it in items), default=0) + 1
    items.append({"id": new_id, "title": title, "done": False})
    _save(items)
    return f"已添加 #{new_id}: {title}"


def complete_todo(todo_id: int) -> str:
    items = _load()
    for it in items:
        if it["id"] == todo_id:
            it["done"] = True
            _save(items)
            return f"#{todo_id} 已标记完成"
    raise ToolError(f"找不到编号为 {todo_id} 的待办")


TOOL_REGISTRY = {
    "list_todos": list_todos,
    "add_todo": add_todo,
    "complete_todo": complete_todo,
}

TOOL_SCHEMAS = [
    {
        "type": "function",
        "function": {
            "name": "add_todo",
            "description": "往待办清单添加一条新待办",
            "parameters": {
                "type": "object",
                "properties": {"title": {"type": "string", "description": "待办内容"}},
                "required": ["title"],
            },
        },
    },
    # list_todos / complete_todo 的 schema 同理，略
]

主循环：事件生产者

流式模式下工具调用的参数不是一次到齐的，模型会分多个增量片段吐出来，需要按 index 把同一个调用的片段拼完整，等本轮流结束后再执行：

# agent/core.py
import json
import os
from typing import Iterator

from openai import OpenAI

from agent.toolkit import TOOL_REGISTRY, TOOL_SCHEMAS, ToolError

client = OpenAI(
    api_key=os.environ["LLM_API_KEY"],
    base_url=os.environ["LLM_BASE_URL"],  # OpenAI 兼容接口，厂商随意换
)
MODEL = os.environ["LLM_MODEL"]
MAX_ROUNDS = 12


def _merge_fragment(pending: dict[int, dict], frag) -> None:
    """按 index 累积流式返回的 tool_call 增量片段。"""
    slot = pending.setdefault(frag.index, {"id": "", "name": "", "args_json": ""})
    if frag.id:
        slot["id"] += frag.id
    if frag.function:
        if frag.function.name:
            slot["name"] += frag.function.name
        if frag.function.arguments:
            slot["args_json"] += frag.function.arguments


def _decode_args(raw: str) -> tuple[dict, str | None]:
    """解析工具参数；失败时返回错误说明而不是抛异常。"""
    if not raw:
        return {}, None
    try:
        parsed = json.loads(raw)
    except json.JSONDecodeError:
        return {"raw": raw}, f"模型生成的参数不是合法 JSON: {raw}"
    if not isinstance(parsed, dict):
        return {"value": parsed}, f"参数应为 JSON 对象，实际是: {raw}"
    return parsed, None


def _run_tool(name: str, kwargs: dict) -> tuple[str, bool]:
    """执行工具，返回 (输出文本, 是否失败)。任何失败都不往外抛。"""
    fn = TOOL_REGISTRY.get(name)
    if fn is None:
        return f"未注册的工具: {name}", True
    try:
        return fn(**kwargs), False
    except ToolError as exc:
        return str(exc), True
    except Exception as exc:  # 工具内部 bug 也降级为工具错误
        return f"工具执行异常: {exc}", True


def agent_stream(task: str) -> Iterator[dict]:
    # 两条线：history 给模型看，yield 出去的事件给外部看。
    history: list[dict] = [
        {"role": "system", "content": "你是待办清单助手，操作清单时使用提供的工具。"},
        {"role": "user", "content": task},
    ]

    for round_no in range(1, MAX_ROUNDS + 1):
        yield {"type": "iteration_start", "iteration": round_no}

        try:
            stream = client.chat.completions.create(
                model=MODEL,
                messages=history,
                tools=TOOL_SCHEMAS,
                stream=True,
            )
        except Exception as exc:
            # 模型接口都调不通了，任务无法继续 —— 这才是系统级 error
            yield {"type": "error", "message": f"模型调用失败: {exc}"}
            return

        text_parts: list[str] = []
        pending_calls: dict[int, dict] = {}

        for chunk in stream:
            if not chunk.choices:
                continue
            delta = chunk.choices[0].delta
            if delta.content:
                text_parts.append(delta.content)
                yield {"type": "token", "content": delta.content}
            if delta.tool_calls:
                for frag in delta.tool_calls:
                    _merge_fragment(pending_calls, frag)

        # 本轮 assistant 输出写回模型上下文（含 tool_calls，协议要求）
        assistant: dict = {"role": "assistant"}
        if text_parts:
            assistant["content"] = "".join(text_parts)
        if pending_calls:
            assistant["tool_calls"] = [
                {
                    "id": c["id"],
                    "type": "function",
                    "function": {"name": c["name"], "arguments": c["args_json"]},
                }
                for c in pending_calls.values()
            ]
        history.append(assistant)

        # 没有工具调用 = 模型给出了最终回答，正常收尾
        if not pending_calls:
            yield {"type": "done"}
            return

        for call in pending_calls.values():
            kwargs, parse_err = _decode_args(call["args_json"])

            yield {
                "type": "tool_start",
                "call_id": call["id"],
                "name": call["name"],
                "args": kwargs,
            }

            if parse_err:
                output, failed = parse_err, True
            else:
                output, failed = _run_tool(call["name"], kwargs)

            # 给外部看：tool_result 事件（工具失败走 is_error，不是 error 事件）
            yield {
                "type": "tool_result",
                "call_id": call["id"],
                "name": call["name"],
                "output": output,
                "is_error": failed,
            }

            # 给模型看：tool 消息写回上下文，模型下一轮才知道结果
            history.append(
                {"role": "tool", "tool_call_id": call["id"], "content": output}
            )

    yield {"type": "error", "message": f"超过最大轮数 {MAX_ROUNDS}，任务被截断"}

消费者一：CLI

CLI 完全不知道 history 怎么维护、工具怎么执行，它只做一件事——决定每种事件怎么显示：

# agent/cli.py
from agent.core import agent_stream


def render(ev: dict) -> None:
    kind = ev["type"]
    if kind == "iteration_start":
        print(f"\n=== 第 {ev['iteration']} 轮 ===")
    elif kind == "token":
        print(ev["content"], end="", flush=True)
    elif kind == "tool_start":
        print(f"\n→ 调用 {ev['name']}  参数: {ev['args']}")
    elif kind == "tool_result":
        mark = "✗" if ev["is_error"] else "✓"
        print(f"← {mark} {ev['name']}: {ev['output']}")
    elif kind == "done":
        print("\n[任务完成]")
    elif kind == "error":
        print(f"\n[系统错误] {ev['message']}")


if __name__ == "__main__":
    while True:
        task = input("\n任务> ").strip()
        if task in {"q", "quit"}:
            break
        for ev in agent_stream(task):
            render(ev)

试一个任务：「记一条待办：周五下班前交周报，然后把清单读给我看」。终端上你会看到轮次、token、两次工具调用和最终确认依次滚出来——表面观感和以前 print 版没区别，但数据通路已经完全不同。

消费者二：Web SSE（示意）

将来接 Web 时，只是换一个消费者，主循环零改动：

import json

def sse_stream(task: str):
    for ev in agent_stream(task):
        yield f"data: {json.dumps(ev, ensure_ascii=False)}\n\n"

消费者三：测试断言

事件流让"过程"第一次变得可测试。以前你只能断言最终回答里包含某个词，现在可以直接断言行为：

def test_add_then_confirm():
    events = list(agent_stream("记一条待办：给服务器续费"))

    tools_used = [e["name"] for e in events if e["type"] == "tool_start"]
    assert "add_todo" in tools_used

    results = [e for e in events if e["type"] == "tool_result"]
    assert all(not e["is_error"] for e in results)

    assert events[-1]["type"] == "done"

调试时，事件流远比最终回答有信息量

一次任务失败，用户看到的可能只是模型的一句"抱歉，我没能完成"。这句话对排障几乎没有价值。但事件流把整个过程摊开了：

任务绕了太多轮？看 iteration_start 的编号涨到了多少。
模型选错了工具？看 tool_start.name。
工具选对了但参数编错了？看 tool_start.args。
工具本身挂了？看 tool_result.is_error 和 output。
到底是正常结束还是被兜底截断？看结尾是 done 还是 error。

最终回答只告诉你结果，事件流告诉你过程在哪一步偏离了预期。这也是为什么值得把事件序列落盘保存：线上问题拿着事件日志回放一遍，比追问用户"它当时说了啥"可靠得多。

实践要点与常见坑

事件结构要当对外 API 对待。 一旦有了多个消费者，随手改事件字段名就是破坏性变更。加字段安全，改名和删字段要走废弃流程。
call_id 必须贯穿始终。 一轮里模型可能发起多个工具调用，tool_start 和 tool_result 只有靠 call_id 才能配对，前端和测试都依赖这一点。
别忘了 messages 的协议义务。 有 tool_calls 的 assistant 消息和对应的 tool 消息必须成对写回，缺一边下一轮请求多半直接报错。
生成器是惰性的。 不迭代就不执行；消费端中途 break，生成器会停在半路。这既是特性（天然支持提前中断）也是坑（以为任务跑完了其实没有）。
事件要可 JSON 序列化。 事件里只放基础类型，不要塞异常对象、文件句柄之类的东西，否则 SSE 和日志落盘都会卡壳。
别把工具报错升级成系统 error。 工具失败是 Agent 的日常，把它交回模型往往下一轮就自我修正了；动不动就终止任务，Agent 的容错能力会大幅退化。

动手清单

给每个事件加一个 ts 时间戳字段，跑几个任务后统计：一次执行里模型生成和工具执行各占多少时间。
写一个日志消费者，把每次运行的事件序列以 JSONL 追加写入文件；再写一个回放脚本，读取日志离线重演当时的执行过程。
实现一个"安静模式"消费者：只在 tool_result.is_error 为真或出现 error 事件时输出，其余事件全部吞掉，适合挂在后台任务上。
用 pytest 写一条行为断言：给定任务「把 3 号待办标记完成」，断言第一轮 tool_start 的工具名是 complete_todo 且参数里 todo_id == 3。
新增第七种事件 usage：每轮结束时上报该轮消耗的 token 数（多数 OpenAI 兼容接口在流式模式下可通过 stream_options 拿到 usage），并让 CLI 在任务结束时汇总打印。

事件接口是这套 Agent 架构里承上启下的一环：往前，它替掉了不可扩展的 print；往后，工具注册器、Web 端 SSE、执行过程持久化都会长在这条事件流上。本系列后面还会展开这些方向。

*本文为学习笔记，知识框架参考自叶小钗「生产级 Agent」系列课程；文字与代码示例为本站原创整理。*