Pi Agent Core 实战（3/10）：40 行跑通第一个 Agent

上一代框架最容易劝退人的地方是脚手架：配置文件、目录约定、中间件注册，跑通一个 hello world 要读半小时文档。pi-agent-core 反过来——核心就是 new Agent() 加一个 streamFn，剩下都是你自己决定。这一篇把最小可运行版本拆开，顺便看清一次 prompt 到底发生了什么。

为了不依赖任何 API 密钥且可复现，实验注入了一个确定性 streamFn 取代真实模型。循环、工具调度、事件发射全部是真实运行时，只有模型输出是脚本化的。

最小骨架：三个东西就能跑

一个能跑起来的 Agent 只需要三样：初始状态（系统提示、模型、工具列表）、一个 streamFn（怎么把上下文变成模型的输出流）、以及一次 prompt 调用。没有 registry，没有 pipeline，没有 chain 组合器。

streamFn 是关键设计。它的签名是拿到模型对象和上下文，返回一个异步事件流。这意味着循环不关心你用什么模型、走什么协议——真实 provider 用它，测试用假的也用它，两者走同一条路径。

这个设计带来的直接好处是可测性：你可以把模型行为完整脚本化，然后精确断言循环在每种情况下做了什么。后面九篇的实验能复现，靠的就是这一点。

最小可运行版本

import { Agent } from '@earendil-works/pi-agent-core'
import { createAssistantMessageEventStream } from '@earendil-works/pi-ai'

const agent = new Agent({
  initialState: {
    systemPrompt: '你是一个排障助手。',
    model,
    tools: [],
  },
  streamFn: (m, context) => {
    const stream = createAssistantMessageEventStream()
    const text = '收到。'
    stream.push({ type: 'start', partial: partialMessage() })
    stream.push({ type: 'text_delta', contentIndex: 0, delta: text, partial: partialMessage(text) })
    stream.push({ type: 'done', reason: 'stop', message: finalMessage(text) })
    return stream
  },
})

agent.subscribe((e) => {
  if (e.type === 'message_update') process.stdout.write(e.assistantMessageEvent.delta ?? '')
})

await agent.prompt('帮我看看服务状态')

背景：注意事件顺序里的一个细节：先 push start，再 push 增量，最后 push done。done 事件里带的是完整消息，而不是“结束标记”。循环拿这个完整消息继续判断要不要调工具、要不要再来一轮。

一次 prompt 的完整事件序列

我跑了一次带两个工具调用的会话：模型先决定同时读日志和查负载，两个工具并行执行，然后基于结果给出结论。整个过程发出 27 个事件，总耗时 314.7 毫秒。

事件顺序是：agent_start → turn_start → 用户消息开始结束 → 助手消息开始 → 若干 text_delta → 助手消息结束 → 两个 tool_execution_start → 工具结束（按完成顺序）→ 工具结果消息 → turn_end → 下一轮 turn_start → 助手消息 → turn_end → agent_end。

关键观察有两点。第一，单个 turn 内可以有多次工具调用，但只有一次模型请求；第二轮请求是因为工具结果需要模型解读，不是循环自己加的。第二，工具结束事件的顺序是完成顺序，不是声明顺序——180 毫秒的那个先结束，300 毫秒的后结束。

一次 prompt 的耗时构成：两个工具并行执行占 300 毫秒，框架自身调度开销不到 5 毫秒。所谓 Agent 慢，慢在模型和工具，不在循环。

这个数据推翻了一个常见直觉：以为 Agent 框架本身很重。实际上循环的开销是毫秒级，你感知到的所有延迟都来自模型往返和工具执行。

推论也很直接：优化 Agent 性能，不要去优化循环。要么并行化工具，要么减少轮次，要么换更快的模型。循环本身没有可优化的空间。

状态是你唯一需要读的东西

Agent 实例上挂着一个 state 对象，里面是当前模型、思考等级、工具列表、完整消息历史，以及几个只读标记：是否正在流式、正在流式的消息、待完成的工具调用、错误信息。

这个 state 是可变的，而且可以直接改。想临时换模型，改 state.model；想在两轮之间塞一条消息，push 进 state.messages。这在很多框架里是禁忌，在这里是设计。

代价是责任：你改了 state，就要保证它仍然是一段合法上下文。宽松的接口换来的是自由，不是零成本。

读状态做展示

agent.subscribe((e) => {
  if (e.type === 'turn_start') {
    console.log('轮次', agent.state.messages.length, '待办工具', agent.state.pendingToolCalls.size)
  }
  if (e.type === 'agent_end') {
    console.log('会话结束，共', e.messages.length, '条消息')
  }
})

还有一个容易被忽略的方法：waitForIdle。prompt 返回不代表界面已经稳定，因为订阅 agent_end 的异步监听器可能还在跑。需要“确定一切结束”的时候，等它，而不是等 prompt。

不配真实模型也能开发

这个系列所有实验都不需要 API 密钥，原因是 streamFn 可以整个替换。把模型行为写成脚本，你就得到了一个完全确定的 Agent——同样的输入永远产生同样的工具调用、同样的事件序列。

这在工程上价值很大。真实模型做回归测试几乎不可能：温度、采样、上游版本变化都会让结果漂移。但把模型换成脚本之后，你可以断言“第三次工具调用必须被钩子拦住”“队列里的消息必须在下一轮被消费”这类循环级行为。

结论：把模型当外部依赖，把循环当自己的代码。前者用集成测试抽查，后者用单元测试锁死。

下一篇讲工具。这是整个系列里最容易被低估的部分——大多数人以为工具就是“注册一个函数”，实际上工具的 schema 设计直接决定了模型能不能用对，执行模式决定了任务要跑多久。

引用与来源

• pi-agent-core README：Quick Start 与事件流 — 核对 2026-10-31
• AssistantMessageEventStream 工厂函数 — createAssistantMessageEventStream 为公开导出