第二章：tRPC-Agent-Go 全架构精读

type Agent interface {
    Run(ctx context.Context, invocation *Invocation) (<-chan *event.Event, error)
    Tools() []tool.Tool
    Info() Info
    SubAgents() []Agent
    FindSubAgent(name string) Agent
}

type Invocation struct {
    // 身份：谁在执行
    Agent      Agent
    AgentInfo  *AgentInfo

    // 输入：用户说了什么
    Messages   []*Message
    UserID     string
    SessionID  string

    // 状态：当前进展
    State      *State
    CurrentStep int
    EndInvocation bool   // Agent 主动标记"我做完了"

    // 能力：能用什么
    Tools      []Tool
    SubAgents  []Agent

    // 控制：约束条件
    MaxSteps   int
    Plan       *Plan     // 预规划的步骤

    // 可观测性：追踪信息
    TraceID    string
    SpanID     string
    RequestID  string
}

// State 的核心结构
type State struct {
    // Values 是一个通用 map —— 节点可以往里面写任何 key-value
    Values   map[string]interface{}

    // Metadata 是系统级元数据 —— 用户代码不应直接修改
    Metadata *StateMetadata
}

type StateMetadata struct {
    CreatedAt    time.Time
    UpdatedAt    time.Time
    Version      int64      // 乐观锁版本号
    CheckpointID string    // 关联的 checkpoint
}

// 场景1：节点间传递中间结果
func researchNode(ctx context.Context, state *State) error {
    results := doSearch(state.Values["query"].(string))
    state.Values["research_results"] = results  // 写入 state
    return nil
}

func writeNode(ctx context.Context, state *State) error {
    results := state.Values["research_results"].([]string)  // 读取上游产出
    article := generateArticle(results)
    state.Values["article"] = article
    return nil
}

// 场景2：循环计数器（防止死循环）
func qualityCheckNode(ctx context.Context, state *State) error {
    count, _ := state.Values["revision_count"].(int)
    state.Values["revision_count"] = count + 1
    return nil
}

// 场景3：条件路由的判断依据
func routingDecision(state *State) string {
    score := state.Values["quality_score"].(float64)
    if score > 0.8 {
        return "done"
    }
    return "revise"
}

// 场景4：跨 Agent 传递上下文（Team 场景）
func coordinatorNode(ctx context.Context, state *State) error {
    state.Values["task_for_researcher"] = "搜索AI趋势"
    state.Values["task_for_writer"] = "写一篇2000字文章"
    return nil
}

// 场景5：存储工具执行的副作用结果
func toolCallback(state *State, toolName string, result string) {
    key := fmt.Sprintf("tool_%s_%d", toolName, time.Now().UnixNano())
    state.Values[key] = result
}

// Agent.Run 返回的是一个事件通道，不是一个结果值
func (a *MyAgent) Run(ctx context.Context, inv *Invocation) (<-chan *event.Event, error) {
    eventChan := make(chan *event.Event, bufferSize)
    go func() {
        // 异步执行，产出事件流
        eventChan <- &event.Event{Type: "thinking", Data: "正在分析..."}
        eventChan <- &event.Event{Type: "tool_call", Data: toolCallData}
        eventChan <- &event.Event{Type: "tool_result", Data: resultData}
        eventChan <- &event.Event{Type: "final_response", Data: answer}
        close(eventChan)
    }()
    return eventChan, nil
}

// 伪代码：RunWithPlugins 的执行顺序
func RunWithPlugins(ctx, inv, agent) {
    for _, plugin := range plugins {
        plugin.BeforeRun(ctx, inv)  // 前置处理
    }

    eventChan := agent.Run(ctx, inv)  // 实际执行

    for _, plugin := range plugins {
        plugin.AfterRun(ctx, inv)   // 后置处理
    }
}

// Plugin 接口定义
type Plugin interface {
    BeforeRun(ctx context.Context, inv *Invocation) error
    AfterRun(ctx context.Context, inv *Invocation, err error) error
    OnEvent(ctx context.Context, ev *event.Event) error
    Name() string
}

// 示例：Token 计费 Plugin
type TokenAccountingPlugin struct {
    store   TokenStore
    limiter TokenLimiter
}

func (p *TokenAccountingPlugin) Name() string {
    return "token_accounting"
}

func (p *TokenAccountingPlugin) BeforeRun(ctx context.Context, inv *Invocation) error {
    // 检查用户配额
    remaining, err := p.limiter.GetRemaining(ctx, inv.UserID)
    if err != nil {
        return fmt.Errorf("check quota failed: %w", err)
    }
    if remaining <= 0 {
        return ErrQuotaExhausted  // 返回错误 → 中止执行
    }
    return nil
}

func (p *TokenAccountingPlugin) OnEvent(ctx context.Context, ev *event.Event) error {
    // 监听 LLM 响应事件，实时累计 token
    if ev.Type == event.TypeLLMResponse {
        usage := ev.Data.(*LLMResponseData).TokenUsage
        inv := GetInvocationFromContext(ctx)
        p.store.AddUsage(ctx, inv.UserID, usage.PromptTokens, usage.CompletionTokens)
    }
    return nil
}

func (p *TokenAccountingPlugin) AfterRun(ctx context.Context, inv *Invocation, runErr error) error {
    // 执行结束后记录计费
    totalUsage := p.store.GetSessionUsage(ctx, inv.SessionID)
    cost := calculateCost(totalUsage, inv.AgentInfo.Model)
    p.store.RecordBilling(ctx, inv.UserID, inv.RequestID, cost)
    return nil
}

// Plugin 之间通过 Invocation.State 传递数据
func (auth *AuthPlugin) BeforeRun(ctx context.Context, inv *Invocation) error {
    user, err := authenticate(ctx)
    if err != nil {
        return ErrUnauthorized
    }
    // 写入 State，下游 Plugin 和 Agent 都能读到
    inv.State.Values["auth_user"] = user
    inv.State.Values["user_tier"] = user.Tier
    return nil
}

func (rateLimit *RateLimitPlugin) BeforeRun(ctx context.Context, inv *Invocation) error {
    // 读取上游 Auth Plugin 写入的用户信息
    tier := inv.State.Values["user_tier"].(string)
    limit := map[string]int{"free": 10, "pro": 100, "enterprise": 1000}[tier]
    if !p.allow(inv.UserID, limit) {
        return ErrRateLimited
    }
    return nil
}

// 把多个 Plugin 打包成一个逻辑单元
func NewProductionSuite(cfg Config) []Plugin {
    return []Plugin{
        NewAuthPlugin(cfg.Auth),
        NewRateLimitPlugin(cfg.RateLimit),
        NewTracingPlugin(cfg.Jaeger),
        NewMetricsPlugin(cfg.Prometheus),
        NewTokenAccountingPlugin(cfg.Billing),
        NewLoggingPlugin(cfg.Logger),  // 放最后：能记录其他 Plugin 的效果
    }
}

// 条件 Plugin：只在特定条件下激活
type ConditionalPlugin struct {
    cond  func(*Invocation) bool
    inner Plugin
}

func (c *ConditionalPlugin) BeforeRun(ctx context.Context, inv *Invocation) error {
    if !c.cond(inv) {
        return nil  // 条件不满足，跳过
    }
    return c.inner.BeforeRun(ctx, inv)
}

// 使用：只对支付相关 Agent 启用审计
auditPlugin := &ConditionalPlugin{
    cond:  func(inv *Invocation) bool { return inv.AgentInfo.Name == "payment_agent" },
    inner: NewAuditPlugin(auditStore),
}

// tRPC-Agent 的状态合并：简单覆盖
state.Values["messages"] = append(state.Values["messages"].([]string), newMsg)
// 如果两个并行节点同时写同一个 key → 竞态条件！
// 需要开发者自己加锁或用不同的 key

// tRPC 配置中心支持热更新 Agent 配置
// 不需要重新部署就能切换 prompt / model
type AgentConfig struct {
    Model       string  `yaml:"model"`        // gpt-4o → claude-3.5-sonnet
    SystemPrompt string `yaml:"system_prompt"` // prompt 版本切换
    MaxSteps    int     `yaml:"max_steps"`
    Temperature float64 `yaml:"temperature"`
}

// 灰度策略：按用户ID hash 分流
func (r *Runner) selectAgentVersion(userID string) *AgentConfig {
    hash := fnv32(userID) % 100
    if hash < 10 {
        return r.configs["canary"]   // 10% 用户走新版本
    }
    return r.configs["stable"]        // 90% 用户走稳定版
}