第三章：tRPC-Agent-Go ReAct 循环与 LLM Flow 深度解析

func (f *Flow) Run(ctx context.Context, invocation *agent.Invocation) (<-chan *event.Event, error) {
    eventChan := make(chan *event.Event, f.channelBufferSize)

    go func(ctx context.Context) {
        defer close(eventChan)

        // 可能从挂起的 tool call 恢复（断点续跑）
        f.maybeResumePendingToolCalls(ctx, invocation, eventChan)

        for {
            // === 循环入口 ===

            // 1. 发射 "开始新一轮" 事件
            f.emitStartEventAndWait(ctx, invocation, eventChan)

            // 2. 可能消费队列中的用户插入消息（Steer 功能）
            f.maybeConsumeQueuedUserMessages(ctx, invocation, eventChan)

            // 3. 执行一步完整的 Think→Act→Observe
            lastEvent, err := f.runOneStep(ctx, invocation, eventChan)

            // === 退出条件 ===
            if lastEvent == nil {
                break  // 出错或被取消
            }
            if invocation.EndInvocation {
                break  // Agent 主动标记结束
            }
            if lastEvent.IsFinalResponse() {
                break  // LLM 给出了最终答案（无 tool_call）
            }

            // 否则继续循环
        }
    }(runCtx)

    return eventChan, nil
}

// 伪代码
func (f *Flow) runOneStep(ctx, inv, eventChan) (*event.Event, error) {
    // 1. 构建 LLM 请求
    llmRequest := f.buildLLMRequest(inv)
    // buildLLMRequest 内部做的事：
    //   - 注入系统指令（system prompt）
    //   - 注入 session 历史消息（让 LLM 知道之前聊了什么）
    //   - 注入可用工具列表（让 LLM 知道能调什么）
    //   - 应用 requestProcessors 链（自定义预处理）

    // 2. 调用 LLM（支持流式响应）
    responseChan := f.callLLM(ctx, inv, llmRequest, callModel)

    // 流式处理 LLM 的响应 token
    var fullResponse *LLMResponse
    for chunk := range responseChan {
        eventChan <- &event.Event{Type: "streaming", Data: chunk}
        fullResponse.Append(chunk)
    }

    // 3. 分析 LLM 的完整响应
    if fullResponse.HasToolCalls() {
        // LLM 决定调用工具
        for _, toolCall := range fullResponse.ToolCalls {
            // 执行工具
            result, err := f.executeTool(ctx, inv, toolCall)

            // 把工具调用和结果追加到 session（下次循环 LLM 就能看到）
            inv.Session.Append(toolCallMessage)
            inv.Session.Append(toolResultMessage)

            // 发射事件
            eventChan <- &event.Event{Type: "tool_call", Data: toolCall}
            eventChan <- &event.Event{Type: "tool_result", Data: result}
        }
        return lastToolEvent, nil  // 返回非 nil、非 final → 循环继续
    }

    // LLM 没有调用工具 → 这就是最终答案
    finalEvent := &event.Event{Type: "final_response", Data: fullResponse.Text}
    eventChan <- finalEvent
    return finalEvent, nil  // IsFinalResponse() == true → 循环退出
}

// 循环开始前，检查是否有未完成的 tool call
f.maybeResumePendingToolCalls(ctx, invocation, eventChan)

// 每轮循环开始时，检查用户是否插入了新消息
f.maybeConsumeQueuedUserMessages(ctx, invocation, eventChan)

if len(toolCalls) > 1 && f.parallelToolExecution {
    results := f.executeToolsInParallel(ctx, inv, toolCalls)
} else {
    results := f.executeToolsSequentially(ctx, inv, toolCalls)
}

result, err := f.executeTool(ctx, inv, toolCall)
if err != nil && f.toolRetryPolicy.ShouldRetry(err) {
    // 重试逻辑：指数退避 + 最大重试次数
    result, err = f.retryTool(ctx, inv, toolCall)
}

type RequestProcessor interface {
    Process(ctx context.Context, inv *Invocation, req *LLMRequest) *LLMRequest
}

// 标准 processors（按顺序执行）：
processors := []RequestProcessor{
    &SystemInstructionProcessor{},   // 注入系统指令
    &SessionHistoryProcessor{},      // 注入历史消息
    &ToolsInjectionProcessor{},      // 注入可用工具
    &PlannerProcessor{},             // 如果有 Planner，注入计划
    &ContextWindowProcessor{},       // 截断超出 context window 的历史
    // ...用户自定义 processors
}

for _, p := range processors {
    request = p.Process(ctx, inv, request)
}

if tool := inv.FindTool(toolCall.Name); tool == nil {
    // 工具不存在，把错误信息告诉 LLM，让它重新选择
    errorMsg := fmt.Sprintf("工具 %s 不存在，可用工具：%v", toolCall.Name, toolNames)
    inv.Session.Append(errorMessage(errorMsg))
    continue  // 不退出循环，让 LLM 重试
}

toolCtx, cancel := context.WithTimeout(ctx, f.toolTimeout)
defer cancel()
result, err := tool.Execute(toolCtx, toolCall.Args)
if err == context.DeadlineExceeded {
    // 超时，告诉 LLM 工具执行超时
    inv.Session.Append(timeoutMessage(toolCall.Name))
}

if f.detectRepetitiveToolCalls(inv.Session) {
    // 检测到重复工具调用模式
    inv.Session.Append(systemMessage("你似乎在重复执行相同操作。请分析已有结果，给出最终答案。"))
}

if f.estimateTokenCount(inv.Session.Messages) > f.maxContextTokens * 0.9 {
    // context 快满了，截断早期历史（保留系统指令和最近几轮）
    inv.Session.Messages = f.truncateHistory(inv.Session.Messages)
}