从零搭建 Agent Harness 系列（十一）REPL 与流式输出

发表于 分类于 Waline：
基于 go-tiny-claw 当前源码，实现连续 Terminal 的 REPL、流式事件协议、ToolCall 聚合和实时输出。

上一讲把连续 Terminal 拆成了四个阶段。本篇实现第一阶段：REPL 负责持续读取用户输入，Provider 负责读取模型流，Engine 负责消费事件，Reporter 负责把文本 Delta 实时显示到终端。

本文源码版本对应后续提交：fb7d7450c0fa9d4d096a06f77e609c4320a55b4f（REPL）和 02dd5b944952fe7a9369f7e6877c4fef4b0881c9（Stream Provider）。

这里有一个重要的架构原则：流式输出是展示协议，不是工具执行协议。文本可以边生成边显示；ToolCall 必须等参数完整后才交给 Engine。

一、先定义 Provider 的流式协议

internal/provider/interface.go 中，非流式接口仍然保留：

1
2
3
type LLMProvider interface {
	Generate(ctx context.Context, messages []schema.Message, availableTools []schema.ToolDefinition) (*schema.Message, error)
}

流式能力通过额外接口表示，而不是修改所有 Provider 的 Generate：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
type StreamEventType string

const (
	StreamTextDelta StreamEventType = "text_delta"
	StreamCompleted StreamEventType = "completed"
	StreamError     StreamEventType = "error"
)

type StreamEvent struct {
	Type    StreamEventType
	Text    string
	Message *schema.Message
	Usage   *schema.Usage
	Err     error
}

type StreamingProvider interface {
	LLMProvider
	GenerateStream(ctx context.Context, messages []schema.Message, tools []schema.ToolDefinition) (<-chan StreamEvent, error)
}

三个事件的含义不同：

1
2
3
StreamTextDelta  一小段可以立即展示的文本
StreamCompleted  完整 Assistant Message，包含完整 ToolCall
StreamError      Provider 或流解析失败

StreamEvent.Message 不应该在每个 Delta 中反复携带。最终消息只在 StreamCompleted 中出现，避免 Engine 把半成品消息写入 Session。

二、OpenAI 兼容 Provider 如何生产事件

当前实现位于 internal/provider/openai.go。它先复用已有的参数构造逻辑，再打开 SDK 的流式请求：

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
29
30
31
32
func (p *OpenAIProvider) GenerateStream(
	ctx context.Context,
	msgs []schema.Message,
	availableTools []schema.ToolDefinition,
) (<-chan StreamEvent, error) {
	params, err := p.buildParams(msgs, availableTools)
	if err != nil {
		return nil, err
	}

	params.StreamOptions.IncludeUsage = openai.Bool(true)

	events := make(chan StreamEvent, 16)

	go func() {
		defer close(events)

		stream := p.client.Chat.Completions.NewStreaming(ctx, params)
		defer stream.Close()

		var content strings.Builder
		toolCalls := make([]*toolCallAccumulator, 0)
		var usage *schema.Usage

		for stream.Next() {
			chunk := stream.Current()
			// 继续处理 chunk。
		}
	}()

	return events, nil
}

这里的 events 是一个带缓冲的 Channel。Provider Goroutine 是生产者，Engine 是消费者。缓冲区大小为 16，意味着短时间内生产者可以先放入少量事件，不必每次都等待消费者。

生产事件时使用当前源码中的 sendStreamEvent：

1
2
3
4
5
6
7
8
9
10
11
12
func sendStreamEvent(
	ctx context.Context,
	events chan<- StreamEvent,
	event StreamEvent,
) bool {
	select {
	case events <- event:
		return true
	case <-ctx.Done():
		return false
	}
}

这比直接写 events <- event 更安全。假如 Engine 因为 Ctrl-C 已经停止消费，Provider 不会永久阻塞在发送操作上，而是从 ctx.Done() 分支退出。

三、文本 Delta 可以立即展示

当前 Provider 对文本 Delta 的处理是：一边累积完整内容，一边发送当前片段。

1
2
3
4
5
6
7
8
9
10
if delta.Content != "" {
	content.WriteString(delta.Content)

	if !sendStreamEvent(ctx, events, StreamEvent{
		Type: StreamTextDelta,
		Text: delta.Content,
	}) {
		return
	}
}

content.WriteString(delta.Content) 不能省略。发送给 Terminal 的 Delta 只是 UI 用的增量；后续要写入 schema.Message.Content 的完整答案仍然需要由 Provider 自己拼起来。

最终消息使用：

1
2
3
4
5
finalMessage := &schema.Message{
	Role:    schema.RoleAssistant,
	Content: content.String(),
	Usage:   usage,
}

如果只把 Delta 发给 Reporter，却不在 Provider 内累计，Engine 最终就拿不到完整内容，下一轮上下文会丢失模型回答。

四、ToolCall 为什么不能直接流式执行

模型返回 ToolCall 时，名称、ID 和参数都可能分散在多个 Chunk 中。当前代码为每个 ToolCall 准备一个累加器：

1
2
3
4
5
type toolCallAccumulator struct {
	id        string
	name      string
	arguments strings.Builder
}

在每个 Delta 中，根据模型给出的 Index 找到对应累加器：

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
for _, deltaToolCall := range delta.ToolCalls {
	index := int(deltaToolCall.Index)

	for len(toolCalls) <= index {
		toolCalls = append(toolCalls, nil)
	}

	if toolCalls[index] == nil {
		toolCalls[index] = &toolCallAccumulator{}
	}

	accumulator := toolCalls[index]

	if deltaToolCall.ID != "" {
		accumulator.id = deltaToolCall.ID
	}

	if deltaToolCall.Function.Name != "" {
		accumulator.name = deltaToolCall.Function.Name
	}

	if deltaToolCall.Function.Arguments != "" {
		accumulator.arguments.WriteString(
			deltaToolCall.Function.Arguments,
		)
	}
}

参数片段可能是：

1
2
3
{"com
mand":"go test
 ./..."}

每一段单独拿出来都不是合法 JSON，所以不能在收到第一段时调用工具。流结束后再转换为正式的 schema.ToolCall：

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
for _, accumulator := range toolCalls {
	if accumulator == nil {
		continue
	}

	arguments := accumulator.arguments.String()
	if arguments == "" {
		arguments = "{}"
	}

	if !json.Valid([]byte(arguments)) {
		sendStreamEvent(ctx, events, StreamEvent{
			Type: StreamError,
			Err: fmt.Errorf(
				"工具 %s 返回非法 JSON 参数: %s",
				accumulator.name,
				arguments,
			),
		})
		return
	}

	finalMessage.ToolCalls = append(finalMessage.ToolCalls, schema.ToolCall{
		ID:        accumulator.id,
		Name:      accumulator.name,
		Arguments: json.RawMessage(arguments),
	})
}

五、结束、错误和 Usage 事件

当前实现从流 Chunk 中提取 Usage：

1
2
3
4
5
6
7
if chunk.Usage.PromptTokens > 0 ||
	chunk.Usage.CompletionTokens > 0 {
	usage = &schema.Usage{
		PromptTokens:     int(chunk.Usage.PromptTokens),
		CompletionTokens: int(chunk.Usage.CompletionTokens),
	}
}

底层流结束后必须检查 stream.Err()：

1
2
3
4
5
6
7
8
9
if err := stream.Err(); err != nil {
	if !sendStreamEvent(ctx, events, StreamEvent{
		Type: StreamError,
		Err:  fmt.Errorf("流式响应失败: %w", err),
	}) {
		return
	}
	return
}

没有错误时发送最终事件：

1
2
3
4
5
sendStreamEvent(ctx, events, StreamEvent{
	Type:    StreamCompleted,
	Message: finalMessage,
	Usage:   usage,
})

这里要区分“Channel 关闭”和“任务完成事件”：

1
2
StreamCompleted：Provider 已经构造出最终消息
close(events)：生产者不会再发送任何事件

二者都需要。Engine 可能先收到 Completed，再收到 Channel 关闭；如果 Channel 直接关闭却没有最终消息，Engine 应该报错，而不是静默结束。

六、Engine 如何消费流

internal/engine/stream.go 用类型断言兼容不支持流式的 Provider：

1
2
3
4
5
streamProvider, ok := e.provider.(provider.StreamingProvider)
if !ok {
	message, err := e.provider.Generate(ctx, messages, tools)
	return message, false, err
}

支持流式时，Engine 监听 Context 和事件 Channel：

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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
events, err := streamProvider.GenerateStream(ctx, messages, tools)
if err != nil {
	return nil, false, err
}

streamReporter, canStream := reporter.(StreamReporter)
emittedText := false
var finalMessage *schema.Message

for {
	select {
	case <-ctx.Done():
		return nil, false, ctx.Err()

	case event, ok := <-events:
		if !ok {
			if finalMessage == nil {
				return nil, false, fmt.Errorf("流式响应未返回最终消息")
			}
			return finalMessage, emittedText, nil
		}

		switch event.Type {
		case provider.StreamTextDelta:
			if emitText && canStream {
				streamReporter.OnTextDelta(ctx, event.Text)
				emittedText = true
			}

		case provider.StreamCompleted:
			finalMessage = event.Message
			if emitText && canStream {
				streamReporter.OnTextComplete(ctx)
			}

		case provider.StreamError:
			if event.Err == nil {
				return nil, false, fmt.Errorf("流式响应失败")
			}
			return nil, false, event.Err
		}
	}
}

StreamReporter 只关心文本展示：

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

ToolCall 不放到这里，因为它不是“已经可以展示的普通文本”，而是 Engine 下一步状态转移的控制数据。工具调用仍然在 StreamCompleted 携带的最终消息里进入审批和执行。

七、REPL 负责把 Run 串起来

internal/cli/repl.go 的主体是一个明确的输入循环：

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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
func (r *REPL) Run(ctx context.Context) error {
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, os.Interrupt)
	defer signal.Stop(signals)

	for {
		fmt.Fprint(r.out, "\nclaw>")

		line, err := r.reader.ReadString('\n')
		if err != nil {
			if errors.Is(err, io.EOF) {
				return nil
			}
			return err
		}

		prompt := strings.TrimSpace(line)
		if prompt == "" {
			continue
		}

		switch prompt {
		case "/exit", "/quit":
			return nil
		case "/clear":
			r.session.Clear()
			fmt.Fprintln(r.out, "会话已清空。")
			continue
		case "/help":
			printHelp(r.out)
			continue
		}

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

		if err := r.runTurn(ctx, signals); err != nil {
			fmt.Fprintf(r.out, "引擎运行出错: %v\n", err)
		}
	}
}

同一个 Session 被反复传入 Run，因此 /clear 清理的是上下文而不是重新创建一套 Engine。中断的 runTurn 细节留到下一篇讲。

八、Terminal Reporter 的职责

当前 TerminalReporter 对文本增量只做一件事：直接写到 stdout。

1
2
3
4
5
6
7
func (r *TerminalReporter) OnTextDelta(ctx context.Context, delta string) {
	fmt.Printf("%s", delta)
}

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

工具状态和普通消息仍然有独立回调：

1
2
3
4
5
6
7
8
9
10
11
func (r *TerminalReporter) OnToolCall(ctx context.Context, toolName string, args string) {
	fmt.Printf("[🛠️ 调用工具] %s\n", toolName)
}

func (r *TerminalReporter) OnToolResult(ctx context.Context, toolName string, result string, isError bool) {
	if isError {
		fmt.Printf("[❌ 执行失败] %s\n", toolName)
		return
	}
	fmt.Printf("[✅ 执行成功] %s\n", toolName)
}

Reporter 这样设计后，换成 WebSocket、飞书或测试用的内存 Reporter，不需要改 Provider 和 Engine。

九、这一篇的验证清单

1
2
3
4
5
6
7
8
不支持 StreamingProvider 时仍能走 Generate
支持流式时文本 Delta 能实时显示
多个 ToolCall 的参数能按 Index 聚合
非法 JSON 参数会产生 StreamError
流结束但没有 Completed 会报错
Provider 发送事件时能响应 ctx.Done()
Session 中保存的是完整 Assistant Message
Reporter 不负责执行工具

下一篇继续处理用户体验中最重要的控制能力：如何让 Ctrl-C 取消当前任务，而不是直接杀掉整个 Terminal 进程。