Method Stream

pkg/agent/streaming.go:55–161

Stream 实现流式执行接口 返回 stream.Reader，支持： - 流式生成事件 - 客户端控制的取消 - 与 LLM 流式 API 无缝集成 - 多消费者支持（Copy, Merge, Transform）

(ctx context.Context, message string, opts ...Option)

// - 与 LLM 流式 API 无缝集成
// - 多消费者支持（Copy, Merge, Transform）
func (a *Agent) Stream(ctx context.Context, message string, opts ...Option) *stream.Reader[*session.Event] {
	reader, writer := stream.Pipe[*session.Event](10)

	go func() {
		defer writer.Close()

		// 应用选项
		config := &streamConfig{}
		for _, opt := range opts {
			opt(config)
		}

		streamLog.Info(ctx, "starting stream", map[string]any{"message": truncate(message, 50)})

		// 1. 前置验证
		if err := a.validateMessage(message); err != nil {
			writer.Send(nil, fmt.Errorf("validate message: %w", err))
			return
		}

		// 2. 创建用户消息
		userMsg := types.Message{
			Role:    types.RoleUser,
			Content: message,
			ContentBlocks: []types.ContentBlock{
				&types.TextBlock{Text: message},
			},
		}

		// 3. 检查 Slash Commands
		if a.commandExecutor != nil {
			if handled, err := a.handleSlashCommandForStream(ctx, &userMsg, writer); handled {
				if err != nil {
					writer.Send(nil, fmt.Errorf("slash command: %w", err))
				}
				return
			}
		}

		// 4. 应用 Skills 增强
		if a.skillInjector != nil {
			skillContext := skills.SkillContext{
				UserMessage: userMsg.GetContent(),
			}
			enhancedPrompt := a.skillInjector.EnhanceSystemPrompt(ctx, a.template.SystemPrompt, skillContext)
			a.template.SystemPrompt = enhancedPrompt
		}

		// 5. 入队消息
		a.mu.Lock()
		a.messages = append(a.messages, userMsg)
		a.mu.Unlock()

		// 6. 持久化消息
		if err := a.persistMessage(ctx, &userMsg); err != nil {
			streamLog.Warn(ctx, "failed to persist message", map[string]any{"error": err})
		}