从零搭建 Agent Harness 系列（十四）抽象 Reporter 与 Runtime：把 Agent Engine 变成可复用内核

发表于 分类于 Waline：
基于 go-tiny-claw 当前源码，拆解 Reporter 和 Runtime 两个边界，让 Agent Engine 从 Terminal 入口中解耦，支持连续对话、流式输出和可取消任务。

前面几篇已经让 go-tiny-claw 具备了连续对话、流式输出、Ctrl-C 中断和工具审批能力。但如果继续把这些能力都写在 main.go 或 REPL 中，系统很快会遇到一个问题：Agent Engine 只能被 Terminal 使用。

一个真正可复用的 Agent Framework，至少需要把下面两个问题拆开：

1
2
Agent 如何把状态和事件交给外部渠道？
一条用户任务如何被启动、等待、取消和结束？

第一个问题由 Reporter 解决，第二个问题由 Runtime 解决。

本文对应当前仓库中 Reporter 和 Runtime 的抽象实现，重点不是增加一个新功能，而是把已经存在的功能重新放到正确的架构边界中。

一、为什么 Engine 不能直接依赖 Terminal

在最初的实现中，Terminal 入口大致负责下面所有事情：

1
2
3
4
5
6
7
8
9
10
11
12
13
读取 stdin
    ↓
追加 User Message
    ↓
创建 Context
    ↓
启动 Engine.Run
    ↓
等待 done Channel
    ↓
处理 Ctrl-C
    ↓
把文本和工具状态打印到 stdout

这种实现可以快速跑起来，但它把三种完全不同的职责混在了一起：

1
2
3
输入：用户从哪里提交 Prompt
执行：Agent 当前这一轮如何运行
输出：模型和工具事件应该如何展示

如果下一步接入 HTTP、WebSocket 或飞书，直接复制 REPL 代码会产生多个问题：

  1. 每个渠道都要重新实现一遍任务生命周期。
  2. Engine 需要知道当前输出是 Terminal 还是消息卡片。
  3. Provider 产生的流式 Chunk 可能直接进入 fmt.Printf。
  4. Ctrl-C、HTTP 取消、飞书撤回实际上都是“取消任务”，却没有统一模型。

因此我们希望得到下面的依赖关系：

1
2
3
4
5
6
7
Terminal / HTTP / 飞书
        ↓
      Runtime
        ↓
    Agent Engine
        ↓
 Provider / Approval / Tool Registry

渠道负责适配输入和展示，Runtime 负责一次任务的生命周期，Engine 只负责完成 Agent Loop。

二、Reporter 是什么

当前 Reporter 定义在 internal/reporter/reporter.go：

1
2
3
4
5
6
7
8
9
10
package reporter

import "context"

type Reporter interface {
	OnThinking(ctx context.Context)
	OnToolCall(ctx context.Context, toolName string, args string)
	OnToolResult(ctx context.Context, toolName string, result string, isError bool)
	OnMessage(ctx context.Context, content string)
}

它表达的不是“如何打印文字”，而是 Agent 运行过程中对外发布的高层事件：

1
2
3
4
OnThinking     开始思考
OnToolCall     准备执行工具
OnToolResult   工具执行完成
OnMessage      得到一条完整回复

Engine 只调用这些方法，不关心具体实现：

1
2
3
if rep != nil {
	rep.OnToolCall(ctx, item.call.Name, string(item.call.Arguments))
}

这段代码既可以把事件打印到 Terminal，也可以把事件转换成 WebSocket 消息、HTTP SSE 事件或飞书卡片更新。

1. Reporter 不是 Logger

Logger 主要记录系统诊断信息，例如：

1
2
3
Provider 请求耗时 2.4 秒
工具执行失败
Trace 已写入文件

Reporter 面向的是 Agent 使用者，表达的是用户需要看到的过程：

1
2
3
4
正在思考
正在调用 bash
工具执行成功
Agent 回复

两者的生命周期和消费对象不同，不能用一个接口混合。Logger 可以写文件，Reporter 通常需要绑定当前会话和当前渠道。

2. Reporter 不应该读取输入

审批看起来也是用户交互，但它与 Reporter 的方向相反：

1
2
Reporter：Agent → 渠道
Approval Handler：渠道 → Agent

如果 Reporter 同时读取 stdin，就会和 REPL 的 Prompt 读取、Terminal Approval 的输入读取互相竞争。因此当前架构把审批保留在 internal/approval，Reporter 只负责输出。

三、为什么 StreamReporter 要单独定义

流式文本不是一条完整消息，而是一系列增量事件。当前源码用另一个接口表达这个能力：

1
2
3
4
type StreamReporter interface {
	OnTextDelta(ctx context.Context, delta string)
	OnTextComplete(ctx context.Context)
}

这样设计有两个好处。

第一，原有 Reporter 实现不需要立刻支持流式输出。一个只支持完整消息的渠道仍然可以实现：

1
2
3
4
5
type SimpleReporter struct{}

func (r *SimpleReporter) OnMessage(ctx context.Context, content string) {
	// 只在完整消息产生后展示
}

第二，Engine 可以在运行时检查渠道是否支持流式能力。当前 internal/engine/stream.go 中的逻辑是：

1
streamReporter, canStream := rep.(reporter.StreamReporter)

如果 canStream 为 true，就把文本 Delta 实时交给 Reporter；否则只等待 StreamCompleted，再通过普通的 OnMessage 输出完整内容。

这是一种 Go 中常见的能力接口设计：基础接口保证最低能力，扩展接口表达可选能力。

四、TerminalReporter 如何实现两个接口

当前 Terminal 实现位于 internal/reporter/terminal_reporter.go：

1
2
3
4
5
type TerminalReporter struct{}

func NewTerminalReporter() *TerminalReporter {
	return &TerminalReporter{}
}

它同时实现普通 Reporter 和 StreamReporter：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
func (r *TerminalReporter) OnMessage(ctx context.Context, content string) {
	if content == "" {
		return
	}
	fmt.Printf("\n🤖 Agent 回复:\n%s\n\n", content)
}

func (r *TerminalReporter) OnTextDelta(ctx context.Context, delta string) {
	fmt.Printf(delta)
}

func (r *TerminalReporter) OnTextComplete(ctx context.Context) {
	fmt.Print("\n\n")
}

因此同一个 Terminal Reporter 可以处理两类响应：

1
2
非流式 Provider：OnMessage
流式 Provider：OnTextDelta + OnTextComplete

Engine 的调用方不需要判断具体 Provider 是哪一种，只需要判断 Reporter 是否有流式展示能力。

不过，这个实现还有一个生产化问题：它直接使用 fmt.Printf，而工具调用是并发执行的。多个 goroutine 同时输出时，文本可能互相穿插。后续应该把输出对象注入 Reporter，并在 Reporter 内部增加互斥或事件队列：

1
2
3
4
type TerminalReporter struct {
	out io.Writer
	mu  sync.Mutex
}

这样测试时也可以传入 bytes.Buffer，不必真的写 stdout。

五、Reporter 抽象如何接入 Engine

Reporter 抽取后，AgentEngine.Run 的参数不再属于 Engine 包，而是依赖独立的 reporter.Reporter：

1
2
3
4
5
func (e *AgentEngine) Run(
	ctx context.Context,
	session *ctxpkg.Session,
	rep reporter.Reporter,
) error

Engine 在关键阶段发布事件：

1
2
3
if e.EnableThinking && rep != nil {
	rep.OnThinking(ctx)
}
1
2
3
if rep != nil {
	rep.OnToolCall(ctx, item.call.Name, string(item.call.Arguments))
}
1
2
3
if rep != nil {
	rep.OnToolResult(ctx, item.call.Name, displayOutput, result.IsError)
}

流式文本则由 generate 负责消费 Provider 的 StreamEvent，再转发给 StreamReporter。因此事件链路是：

1
2
3
4
5
6
7
OpenAI SDK Chunk
    ↓
Provider StreamEvent
    ↓
Engine 处理和聚合
    ↓
Reporter 展示

Provider 不知道终端存在，Reporter 也不需要知道 OpenAI SDK 的 Chunk 结构。

六、Runtime 是什么

Reporter 解决了“事件发给谁”，Runtime 解决的是“任务如何运行”。当前 Runtime 位于 internal/runtime/runtime.go，先定义了一个可以被 Engine 实现的运行接口：

1
2
3
type AgentRunner interface {
	Run(ctx context.Context, session *ctxpkg.Session, rep reporter.Reporter) error
}

AgentEngine 天然满足这个接口，因为它已经有同样签名的 Run 方法。Runtime 不需要依赖 *engine.AgentEngine 这个具体类型，只依赖 AgentRunner：

1
2
3
4
type Runtime struct {
	runner  AgentRunner
	session *ctxpkg.Session
}

构造函数也因此很简单：

1
2
3
4
5
6
func NewRuntime(runner AgentRunner, session *ctxpkg.Session) *Runtime {
	return &Runtime{
		runner:  runner,
		session: session,
	}
}

这里的关键不是代码量，而是依赖方向发生了变化：

1
2
之前：REPL → AgentEngine
现在：REPL → Runtime → AgentRunner

未来可以为测试构造一个假的 AgentRunner，也可以为不同渠道创建不同 Runtime 适配层，而不需要让 REPL 了解 Engine 的细节。

七、Start 如何创建一条任务

当前 Runtime 的异步入口是 Start：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
func (r *Runtime) Start(
	parent context.Context,
	prompt string,
	rep reporter.Reporter,
) (*Task, error) {
	prompt = strings.TrimSpace(prompt)
	if prompt == "" {
		return nil, errors.New("用户提示不能为空")
	}

	r.session.Append(schema.Message{
		Role:    schema.RoleUser,
		Content: prompt,
	})

	runCtx, cancel := context.WithCancel(parent)
	done := make(chan error, 1)

	go func() {
		defer close(done)
		done <- r.runner.Run(runCtx, r.session, rep)
	}()

	return &Task{
		done:   done,
		cancel: cancel,
	}, nil
}

这段代码完成了四件事。

1. Runtime 负责写入用户消息

以前是 REPL 直接调用 session.Append。现在 Prompt 由 Runtime 统一追加：

1
2
3
4
5
6
7
Terminal Prompt
      ↓
Runtime.Start
      ↓
Session.Append(UserMessage)
      ↓
AgentRunner.Run

这样 HTTP、飞书和 Terminal 都不会各自实现一套消息写入逻辑，也不会出现同一个 Prompt 被追加两次。

2. 每次 Start 都创建独立的 Context

Runtime 接收渠道传入的父 Context，然后创建当前任务自己的子 Context：

1
runCtx, cancel := context.WithCancel(parent)

父 Context 通常代表整个进程或请求；子 Context 代表这一轮 Agent Run。取消边界因此变成：

1
2
3
4
5
6
7
8
9
10
11
程序退出 / HTTP 请求结束
              ↓
          parent cancel
              ↓
          runCtx cancel

Terminal Ctrl-C
              ↓
          task.Cancel()
              ↓
          runCtx cancel

Engine、Provider 和支持 Context 的工具都可以沿着这条链路退出。

3. Runtime 用 Goroutine 隔离 Engine

如果在 Start 中同步调用 runner.Run，调用方就无法及时得到一个任务句柄，也无法同时处理取消信号。因此 Runtime 把 Runner 放进 goroutine，并立即返回 Task。

4. done 使用容量为 1 的 Channel

1
done := make(chan error, 1)

Runner 结束时把结果写入 Channel，调用方通过 Task.Done() 读取。容量为 1 很重要：即使调用方先响应 Ctrl-C、暂时还没有读取结果，Runner 也可以安全地把最终错误写进去，不会因为发送结果而永久阻塞。

八、Task 是一次运行的句柄

当前 Task 定义在 internal/runtime/task.go：

1
2
3
4
5
6
7
8
9
10
11
12
type Task struct {
	done   <-chan error
	cancel context.CancelFunc
}

func (t *Task) Done() <-chan error {
	return t.done
}

func (t *Task) Cancel() {
	t.cancel()
}

调用方不需要接触 Runtime 内部的 Context 和 goroutine，只需要操作两个方法：

1
2
3
4
5
6
7
8
9
10
11
12
task, err := rt.Start(ctx, prompt, rep)
if err != nil {
	return err
}

select {
case err := <-task.Done():
	return err
case <-signals:
	task.Cancel()
	return <-task.Done()
}

这就是任务控制面的最小模型：

1
2
3
Start → 得到 Task
Task.Done → 等待结束
Task.Cancel → 请求取消

未来还可以在 Task 上增加：

1
2
3
4
5
ID()       返回任务 ID
Status()   查看任务状态
Err()      获取最终错误
Wait()     阻塞等待并返回错误
Events()   订阅结构化运行事件

但第一步不应该过早设计一个复杂的任务对象。先把生命周期边界固定下来，再根据渠道需求扩展。

九、同步 Run 只是 Start 的便捷封装

Runtime 同时保留同步入口：

1
2
3
4
5
6
7
8
9
10
11
func (r *Runtime) Run(
	ctx context.Context,
	prompt string,
	rep reporter.Reporter,
) error {
	task, err := r.Start(ctx, prompt, rep)
	if err != nil {
		return err
	}
	return <-task.Done()
}

这里不要再复制一套 Prompt 校验、Session 追加和 goroutine 逻辑。同步 Run 只是：

1
2
3
4
5
Start
  ↓
等待 Done
  ↓
返回错误

这样同步调用方和异步调用方共享完全相同的行为，后续修复取消或错误处理时也只需要改一个地方。

十、REPL 改造后只剩渠道职责

当前 internal/cli/repl.go 定义了一个更窄的 Runtime 接口：

1
2
3
4
type Runtime interface {
	Start(parent context.Context, prompt string, reporter reporter.Reporter) (*runtime.Task, error)
	Clear()
}

REPL 自己只保存输入、输出、Runtime 和 Reporter：

1
2
3
4
5
6
type REPL struct {
	reader   *bufio.Reader
	out      io.Writer
	runtime  Runtime
	reporter reporter.Reporter
}

本地命令由 REPL 处理：

1
2
3
4
5
6
7
8
9
10
11
switch prompt {
case "/exit", "/quit":
	return nil
case "/clear":
	r.runtime.Clear()
	fmt.Fprintln(r.out, "会话已清空。")
	continue
case "/help":
	printHelp(r.out)
	continue
}

普通 Prompt 则交给 Runtime：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
func (r *REPL) runTurn(
	parent context.Context,
	prompt string,
	signals <-chan os.Signal,
) error {
	task, err := r.runtime.Start(parent, prompt, r.reporter)
	if err != nil {
		return err
	}

	select {
	case err := <-task.Done():
		return err
	case <-signals:
		fmt.Fprintln(r.out, "\n正在取消当前任务...")
		task.Cancel()
		return <-task.Done()
	}
}

改造后的 REPL 不再知道：

1
2
3
4
5
Engine 如何循环
Provider 如何流式读取
ToolCall 如何聚合
Session 如何保存历史
工具如何并发执行

它只知道用户输入、命令、信号和任务句柄。这就是渠道层应该拥有的复杂度。

十一、main.go 如何组装对象

完成抽象后，启动入口负责依赖注入：

1
2
3
4
5
6
eng := engine.NewAgentEngine(llmProvider, registry, gate, false, false)
rep := reporter.NewTerminalReporter()
sess := ctxpkg.GlobalSessionMgr.GetOrCreate("terminal_default", workDir)
rt := runtime.NewRuntime(eng, sess)

repl := cli.NewREPL(reader, os.Stdout, rt, rep)

组装关系是：

1
2
3
4
OpenAIProvider ─┐
ToolRegistry ───┼→ AgentEngine ─→ Runtime ─→ REPL
ApprovalGate ───┘                  ↑          ↑
                              Session     Reporter

main.go 仍然负责组装具体实现，但不再让具体依赖渗透进 REPL。将来换成 HTTP 入口时，可以复用：

1
2
rt := runtime.NewRuntime(engine, session)
task, err := rt.Start(request.Context(), prompt, httpReporter)

这就是 Runtime 抽象的实际价值：不是为了多创建一个 package，而是为了让不同入口复用同一条 Agent 执行链路。

十二、一次完整请求的时序

把 Reporter 和 Runtime 放在一起，一次请求的时序如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
用户输入 Prompt
      ↓
REPL.runTurn
      ↓
Runtime.Start
      ├── 校验 Prompt
      ├── Session.Append(UserMessage)
      ├── context.WithCancel
      └── goroutine → AgentRunner.Run
                              ↓
                       Engine.generate
                              ↓
                       StreamEvent
                              ↓
                       Reporter.OnTextDelta
                              ↓
                       Tool / Approval
                              ↓
                       Reporter.OnToolResult
                              ↓
                       Task.Done
                              ↓
                       返回 claw>

Ctrl-C 的路径则是：

1
2
3
4
5
6
7
8
9
10
11
Ctrl-C
  ↓
REPL 收到 os.Interrupt
  ↓
Task.Cancel
  ↓
runCtx.Done
  ├── Engine 停止消费流
  ├── Provider 停止发送事件
  ├── bash 子进程收到取消
  └── Runner 返回 context.Canceled

注意，Reporter 不参与取消。它只展示取消前已经产生的事件；取消控制权属于 Runtime 和 Task。

十三、这次抽象解决了什么

1. Engine 可以脱离 Terminal 测试

可以使用一个假的 Reporter 收集事件：

1
2
3
type RecordingReporter struct {
	messages []string
}

也可以使用假的 AgentRunner 测试 Runtime：

1
2
3
type FakeRunner struct {
	run func(context.Context, *ctxpkg.Session, reporter.Reporter) error
}

不需要启动真实模型，也不需要读取 stdin。

2. 输出渠道可以替换

同一套 Engine 和 Runtime 可以连接不同 Reporter：

1
2
3
4
5
TerminalReporter
SSEReporter
WebSocketReporter
FeishuReporter
RecordingReporter

3. 任务控制统一

Terminal 的 Ctrl-C、HTTP 的请求取消、服务端的超时和未来的“停止任务”按钮，都可以映射到同一个操作：

1
task.Cancel()

4. 连续会话有了明确归属

REPL 不再直接维护消息写入规则，Runtime 负责把 Prompt 转换为 Session 中的一条 User Message，再启动一次 Agent Run。

十四、当前实现还不够生产级

目前的 Runtime 是一个最小可用抽象，还需要继续补齐以下边界。

1. 防止同一 Session 并发运行

当前 Start 可以被连续调用两次。两个 Runner 会同时修改同一个 Session，导致上下文顺序和 Reporter 输出不可预测。

Runtime 至少应该增加：

1
2
3
4
activeTask
互斥锁
忙碌错误或排队策略
任务结束后的清理

2. 任务状态不能只靠 error

context.Canceled、超时、用户拒绝和 Engine 失败应该在外部有清晰区分。后续可以定义：

1
2
3
4
5
6
7
8
type TaskStatus string

const (
	TaskRunning   TaskStatus = "running"
	TaskCompleted TaskStatus = "completed"
	TaskCanceled  TaskStatus = "canceled"
	TaskFailed    TaskStatus = "failed"
)

3. Reporter 需要结构化事件

当前 Reporter 方法已经比 fmt.Printf 好很多，但仍然是多个回调。生产环境还需要统一事件模型：

1
2
3
4
5
6
7
type Event struct {
	TaskID string
	Kind   string
	Text   string
	Data   any
	At     time.Time
}

这样日志、UI、Trace 和远程渠道可以订阅同一份事件，而不必分别实现十几个回调。

4. 终端输入和审批输入需要统一协调

当前 REPL 读取 Prompt，Approval Handler 也读取 Terminal 输入。它们必须保证不会同时消费同一个 stdin。更进一步，可以由 Runtime 统一管理输入请求，并让审批变成一种挂起的任务状态。

5. Session 需要持久化

当前 Session 主要存在内存中。进程重启后，连续对话会丢失。生产级 Runtime 需要抽象：

1
2
3
4
SessionStore
TaskStore
GrantStore
TraceStore

然后根据部署方式选择内存、SQLite、PostgreSQL 或其他存储。

十五、下一步应该怎么做

Reporter 和 Runtime 抽象完成后，下一步不是继续给 Engine 增加更多 if，而是先补齐任务控制面：

1
2
3
4
5
6
7
8
9
Runtime 单任务并发控制
        ↓
Task 状态机和任务 ID
        ↓
结构化 Event 总线
        ↓
Session 持久化
        ↓
跨渠道 Runtime Adapter

等这些边界稳定后，Agent Framework 才能从“一个可以运行的 Terminal Demo”逐渐变成“多个入口共享的执行平台”。

Reporter 让 Engine 的输出可以被替换，Runtime 让 Engine 的生命周期可以被控制。两者结合之后，Agent 的核心能力终于不再绑定某一个终端入口，这也是从 Demo 走向 Framework 的关键一步。