Eino 框架完整源码剖析手册

graph TB
    subgraph "Eino 核心框架"
        A[Schema 基础数据层] --> B[Components 组件抽象层]
        B --> C[Compose 编排引擎层]
        C --> D[Flow 流程预制层]
        C --> E[ADK 智能体开发套件]
        F[Callbacks 回调切面层] -.-> C
    end
    
    subgraph "组件类型"
        G[ChatModel 聊天模型] --> B
        H[Tool 工具] --> B
        I[ChatTemplate 模板] --> B
        J[Retriever 检索器] --> B
        K[Embedding 嵌入] --> B
        L[Indexer 索引器] --> B
    end
    
    subgraph "编排方式"
        M[Chain 链式编排] --> C
        N[Graph 图式编排] --> C
        O[Workflow 工作流编排] --> C
    end
    
    subgraph "执行模式"
        P[Invoke 同步执行] --> Q[Runnable 可执行接口]
        R[Stream 流式执行] --> Q
        S[Collect 收集执行] --> Q
        T[Transform 转换执行] --> Q
    end

graph TB
    subgraph "应用层 Application Layer"
        A1[用户应用] --> A2[ReAct Agent]
        A2 --> A3[自定义 Agent]
    end
    
    subgraph "流程层 Flow Layer"
        F1[Agent 流程] --> F2[多代理协作]
        F3[检索增强] --> F4[自定义流程]
    end
    
    subgraph "编排层 Compose Layer"
        C1[Chain 链式编排] --> C2[Graph 图式编排]
        C2 --> C3[Workflow 工作流]
        C4[Runnable 执行接口] --> C5[流式处理引擎]
    end
    
    subgraph "组件层 Components Layer"
        CM1[ChatModel] --> CM2[ChatTemplate]
        CM2 --> CM3[Tool]
        CM3 --> CM4[Retriever]
        CM4 --> CM5[Embedding]
        CM5 --> CM6[Indexer]
    end
    
    subgraph "基础层 Schema Layer"
        S1[Message 消息体系] --> S2[StreamReader 流处理]
        S2 --> S3[ToolInfo 工具信息]
        S3 --> S4[Document 文档]
    end
    
    subgraph "回调层 Callbacks Layer"
        CB1[OnStart/OnEnd] --> CB2[OnError]
        CB2 --> CB3[Stream Callbacks]
    end
    
    A1 --> F1
    F1 --> C1
    C1 --> CM1
    CM1 --> S1
    
    CB1 -.-> C1
    CB1 -.-> CM1

sequenceDiagram
    participant User as 用户应用
    participant Compose as 编排层
    participant Component as 组件层
    participant Schema as 基础层
    participant Callback as 回调层
    
    User->>Compose: 创建编排对象
    Compose->>Component: 注册组件
    Component->>Schema: 使用数据结构
    
    User->>Compose: 编译执行
    Compose->>Callback: 触发开始回调
    Compose->>Component: 执行组件逻辑
    Component->>Schema: 处理消息流
    Schema-->>Component: 返回处理结果
    Component-->>Compose: 返回组件输出
    Compose->>Callback: 触发结束回调
    Compose-->>User: 返回最终结果

// Runnable 是所有可执行对象的核心接口
// 位置: compose/runnable.go:32
type Runnable[I, O any] interface {
    // Invoke 同步执行：单输入 -> 单输出
    Invoke(ctx context.Context, input I, opts ...Option) (output O, err error)
    
    // Stream 流式执行：单输入 -> 流输出
    Stream(ctx context.Context, input I, opts ...Option) (output *schema.StreamReader[O], err error)
    
    // Collect 收集执行：流输入 -> 单输出
    Collect(ctx context.Context, input *schema.StreamReader[I], opts ...Option) (output O, err error)
    
    // Transform 转换执行：流输入 -> 流输出
    Transform(ctx context.Context, input *schema.StreamReader[I], opts ...Option) (output *schema.StreamReader[O], err error)
}

// NewChain 创建链式编排
// 位置: compose/chain.go:37
func NewChain[I, O any](opts ...NewGraphOption) *Chain[I, O] {
    ch := &Chain[I, O]{
        gg: NewGraph[I, O](opts...),  // 底层使用 Graph 实现
    }
    
    ch.gg.cmp = ComponentOfChain  // 标记为链式组件
    
    return ch
}

// Chain 链式编排结构
// 位置: compose/chain.go:47
type Chain[I, O any] struct {
    gg *Graph[I, O]  // 底层图实现
    
    nodeIdx     int      // 节点索引
    preNodeKeys []string // 前置节点键
    hasEnd      bool     // 是否已添加结束边
}

// NewGraph 创建图式编排
// 位置: compose/generic_graph.go:68
func NewGraph[I, O any](opts ...NewGraphOption) *Graph[I, O] {
    options := &newGraphOptions{}
    for _, opt := range opts {
        opt(options)
    }

    g := &Graph[I, O]{
        newGraphFromGeneric[I, O](
            ComponentOfGraph,
            options.withState,
            options.stateType,
            opts,
        ),
    }

    return g
}

// graph 内部图结构
// 位置: compose/graph.go
type graph struct {
    // 节点管理
    nodes        map[string]*graphNode     // 节点映射表
    controlEdges map[string][]string      // 控制依赖边
    dataEdges    map[string][]string      // 数据流边
    branches     map[string][]*GraphBranch // 分支条件
    
    // 执行控制
    startNodes   []string                 // 起始节点
    endNodes     []string                 // 结束节点
    
    // 类型系统
    expectedInputType  reflect.Type       // 期望输入类型
    expectedOutputType reflect.Type       // 期望输出类型
    genericHelper      *genericHelper     // 泛型助手
    
    // 状态管理
    stateType      reflect.Type           // 状态类型
    stateGenerator func(ctx context.Context) any // 状态生成器
    
    // 编译状态
    compiled   bool                       // 是否已编译
    buildError error                      // 构建错误
}

// NewWorkflow 创建工作流编排
// 位置: compose/workflow.go:61
func NewWorkflow[I, O any](opts ...NewGraphOption) *Workflow[I, O] {
    options := &newGraphOptions{}
    for _, opt := range opts {
        opt(options)
    }

    wf := &Workflow[I, O]{
        g: newGraphFromGeneric[I, O](
            ComponentOfWorkflow,
            options.withState,
            options.stateType,
            opts,
        ),
        workflowNodes: make(map[string]*WorkflowNode),
        dependencies:  make(map[string]map[string]dependencyType),
    }

    return wf
}

// Workflow 工作流结构
// 位置: compose/workflow.go:45
type Workflow[I, O any] struct {
    g                *graph                    // 底层图
    workflowNodes    map[string]*WorkflowNode  // 工作流节点
    workflowBranches []*WorkflowBranch         // 工作流分支
    dependencies     map[string]map[string]dependencyType // 依赖关系
}

// WorkflowNode 工作流节点
type WorkflowNode struct {
    g                *graph
    key              string
    addInputs        []func() error           // 输入添加函数
    staticValues     map[string]any           // 静态值
    dependencySetter func(fromNodeKey string, typ dependencyType)
    mappedFieldPath  map[string]any           // 字段映射路径
}

graph TB
    subgraph "Schema 模块"
        A[Message 消息体系] --> B[StreamReader 流处理]
        A --> C[ToolInfo 工具信息]
        A --> D[Document 文档]
        B --> E[流式操作接口]
        C --> F[工具调用结构]
        D --> G[文档处理]
    end
    
    subgraph "核心数据结构"
        H[Message] --> I[ToolCall]
        H --> J[ChatMessagePart]
        H --> K[ResponseMeta]
        L[StreamReader] --> M[流读取接口]
        N[ToolInfo] --> O[函数签名]
    end

// Message 核心消息结构
// 位置: schema/message.go
type Message struct {
    Role    RoleType `json:"role"`    // 消息角色：user、assistant、system、tool
    Content string   `json:"content"` // 消息内容
    
    // 多媒体内容支持
    MultiContent []ChatMessagePart `json:"multi_content,omitempty"`
    
    // 工具调用相关
    ToolCalls  []ToolCall `json:"tool_calls,omitempty"`
    ToolCallID string     `json:"tool_call_id,omitempty"`
    ToolName   string     `json:"tool_name,omitempty"`
    
    // 响应元数据
    ResponseMeta *ResponseMeta `json:"response_meta,omitempty"`
    
    // 推理内容（思维链）
    ReasoningContent string `json:"reasoning_content,omitempty"`
    
    // 扩展字段
    Extra map[string]any `json:"extra,omitempty"`
}

// ConcatMessages 合并消息流 - 核心流处理函数
// 位置: schema/message.go
func ConcatMessages(msgs []*Message) (*Message, error) {
    var (
        contents            []string
        contentLen          int
        reasoningContents   []string
        reasoningContentLen int
        toolCalls           []ToolCall
        ret                 = Message{}
        extraList           = make([]map[string]any, 0, len(msgs))
    )

    // 1. 验证消息一致性
    for idx, msg := range msgs {
        if msg == nil {
            return nil, fmt.Errorf("unexpected nil chunk in message stream, index: %d", idx)
        }

        // 验证角色一致性
        if msg.Role != "" {
            if ret.Role == "" {
                ret.Role = msg.Role
            } else if ret.Role != msg.Role {
                return nil, fmt.Errorf("cannot concat messages with different roles: '%s' '%s'", ret.Role, msg.Role)
            }
        }
        
        // 收集内容
        if msg.Content != "" {
            contents = append(contents, msg.Content)
            contentLen += len(msg.Content)
        }
        
        // 收集工具调用
        if len(msg.ToolCalls) > 0 {
            toolCalls = append(toolCalls, msg.ToolCalls...)
        }
    }

    // 2. 合并内容
    if len(contents) > 0 {
        var sb strings.Builder
        sb.Grow(contentLen)
        for _, content := range contents {
            sb.WriteString(content)
        }
        ret.Content = sb.String()
    }

    // 3. 合并工具调用
    if len(toolCalls) > 0 {
        merged, err := concatToolCalls(toolCalls)
        if err != nil {
            return nil, err
        }
        ret.ToolCalls = merged
    }

    return &ret, nil
}

// StreamReader 流式读取器
// 位置: schema/stream.go
type StreamReader[T any] struct {
    // 内部通道和状态管理
    ch     <-chan streamFrame[T]
    closed bool
    mu     sync.Mutex
}

// streamFrame 流数据帧
type streamFrame[T any] struct {
    data T
    err  error
}

// Recv 接收下一个流元素
// 位置: schema/stream.go
func (sr *StreamReader[T]) Recv() (T, error) {
    sr.mu.Lock()
    defer sr.mu.Unlock()
    
    if sr.closed {
        var zero T
        return zero, io.EOF
    }
    
    frame, ok := <-sr.ch
    if !ok {
        sr.closed = true
        var zero T
        return zero, io.EOF
    }
    
    if frame.err != nil {
        return frame.data, frame.err
    }
    
    return frame.data, nil
}

// StreamReaderFromArray 从数组创建流读取器
func StreamReaderFromArray[T any](items []T) *StreamReader[T] {
    ch := make(chan streamFrame[T], len(items))
    
    for _, item := range items {
        ch <- streamFrame[T]{data: item}
    }
    close(ch)
    
    return &StreamReader[T]{ch: ch}
}

sequenceDiagram
    participant App as 应用层
    participant Msg as Message
    participant Stream as StreamReader
    participant Concat as ConcatMessages
    
    App->>Msg: 创建消息
    Msg->>Stream: 转换为流
    
    loop 流处理
        Stream->>Stream: Recv()
        Stream-->>App: 返回消息块
    end
    
    App->>Concat: 合并消息流
    Concat->>Concat: 验证一致性
    Concat->>Concat: 合并内容
    Concat->>Concat: 处理工具调用
    Concat-->>App: 返回合并结果

graph TB
    subgraph "Components 组件层"
        A[ChatModel] --> B[BaseChatModel]
        A --> C[ToolCallingChatModel]
        D[Tool] --> E[InvokableTool]
        D --> F[StreamableTool]
        G[ChatTemplate] --> H[MessagesTemplate]
        I[Retriever] --> J[检索接口]
        K[Embedding] --> L[嵌入接口]
        M[Indexer] --> N[索引接口]
    end
    
    subgraph "组件特性"
        O[类型安全] --> P[编译时检查]
        Q[流式支持] --> R[自动转换]
        S[回调机制] --> T[切面注入]
    end

// BaseChatModel 基础聊天模型接口
// 位置: components/model/interface.go
type BaseChatModel interface {
    Generate(ctx context.Context, input []*schema.Message, opts ...Option) (*schema.Message, error)
    Stream(ctx context.Context, input []*schema.Message, opts ...Option) (*schema.StreamReader[*schema.Message], error)
}

// ToolCallingChatModel 支持工具调用的聊天模型
type ToolCallingChatModel interface {
    BaseChatModel
    WithTools(tools []*schema.ToolInfo) (ToolCallingChatModel, error)
}

sequenceDiagram
    participant User as 用户代码
    participant Model as ChatModel
    participant Callback as 回调系统
    participant Stream as 流处理器
    
    User->>Model: Generate(messages)
    Model->>Callback: OnStart
    Model->>Model: 处理输入消息
    Model->>Model: 调用底层API
    Model->>Callback: OnEnd
    Model-->>User: 返回消息
    
    User->>Model: Stream(messages)
    Model->>Callback: OnStartWithStreamInput
    Model->>Stream: 创建流
    
    loop 流式输出
        Model->>Stream: 发送消息块
        Stream-->>User: 接收消息块
    end
    
    Model->>Callback: OnEndWithStreamOutput

// InvokableTool 可调用工具接口
// 位置: components/tool/interface.go
type InvokableTool interface {
    BaseTool
    InvokableRun(ctx context.Context, argumentsInJSON string, opts ...Option) (string, error)
}

// StreamableTool 流式工具接口  
type StreamableTool interface {
    BaseTool
    StreamableRun(ctx context.Context, argumentsInJSON string, opts ...Option) (*schema.StreamReader[string], error)
}

// BaseTool 基础工具接口
type BaseTool interface {
    Info(ctx context.Context) (*schema.ToolInfo, error)
}

// ToolsNode 工具节点结构
// 位置: compose/tool_node.go
type ToolsNode struct {
    tuple                   []tool.BaseTool
    unknownToolsHandler     func(ctx context.Context, name, input string) (string, error)
    executeSequentially     bool
    toolArgumentsHandler    func(ctx context.Context, name, arguments string) (string, error)
}

// Invoke 工具执行的核心逻辑
func (tn *ToolsNode) Invoke(ctx context.Context, input *schema.Message, opts ...ToolsNodeOption) ([]*schema.Message, error) {
    // 1. 解析工具调用
    tasks, err := tn.genToolCallTasks(ctx, tn.tuple, input, opt.executedTools, false)
    if err != nil {
        return nil, err
    }

    // 2. 执行工具（并行或串行）
    if tn.executeSequentially {
        sequentialRunToolCall(ctx, runToolCallTaskByInvoke, tasks, opt.ToolOptions...)
    } else {
        parallelRunToolCall(ctx, runToolCallTaskByInvoke, tasks, opt.ToolOptions...)
    }

    // 3. 收集结果
    output := make([]*schema.Message, len(tasks))
    for i, task := range tasks {
        if task.err != nil {
            return nil, fmt.Errorf("tool execution failed: %w", task.err)
        }
        output[i] = schema.ToolMessage(task.output, task.callID, schema.WithToolName(task.name))
    }

    return output, nil
}

// ChatTemplate 聊天模板接口
// 位置: components/prompt/interface.go
type ChatTemplate interface {
    Format(ctx context.Context, vs map[string]any, opts ...Option) ([]*schema.Message, error)
}

// 支持的模板格式
type FormatType uint8

const (
    FString    FormatType = 0  // Python 风格格式化
    GoTemplate FormatType = 1  // Go 标准模板
    Jinja2     FormatType = 2  // Jinja2 模板
)

// formatContent 格式化内容的核心函数
// 位置: components/prompt/chat_template.go
func formatContent(content string, vs map[string]any, formatType FormatType) (string, error) {
    switch formatType {
    case FString:
        return pyfmt.Fmt(content, vs)
    case GoTemplate:
        parsedTmpl, err := template.New("template").
            Option("missingkey=error").
            Parse(content)
        if err != nil {
            return "", err
        }
        sb := new(strings.Builder)
        err = parsedTmpl.Execute(sb, vs)
        return sb.String(), err
    case Jinja2:
        env, err := getJinjaEnv()
        if err != nil {
            return "", err
        }
        tpl, err := env.FromString(content)
        if err != nil {
            return "", err
        }
        return tpl.Execute(vs)
    default:
        return "", fmt.Errorf("unknown format type: %v", formatType)
    }
}

graph TD
    subgraph "编排接口层"
        Chain[Chain 链式编排]
        Graph[Graph 图编排]
        Workflow[Workflow 工作流]
    end
    
    subgraph "核心抽象层"
        Runnable[Runnable 可执行接口]
        ComposableRunnable[ComposableRunnable 可组合执行器]
    end
    
    subgraph "执行引擎层"
        Runner[Runner 执行器]
        GraphManager[GraphManager 图管理器]
        ChannelManager[ChannelManager 通道管理器]
        TaskManager[TaskManager 任务管理器]
    end
    
    subgraph "支撑组件层"
        State[State 状态管理]
        Stream[Stream 流式处理]
        Branch[Branch 分支逻辑]
        FieldMapping[FieldMapping 字段映射]
    end
    
    subgraph "基础设施层"
        TypeSystem[Type System 类型系统]
        ErrorHandling[Error Handling 错误处理]
        Callbacks[Callbacks 回调机制]
    end
    
    %% 编排接口层关系
    Chain --> Runnable
    Graph --> Runnable
    Workflow --> Runnable
    
    %% 核心抽象层关系
    Runnable --> ComposableRunnable
    ComposableRunnable --> Runner
    
    %% 执行引擎层关系
    Runner --> GraphManager
    Runner --> ChannelManager
    Runner --> TaskManager
    
    %% 支撑组件层关系
    Runner --> State
    Runner --> Stream
    Runner --> Branch
    Workflow --> FieldMapping
    
    %% 基础设施层关系
    ComposableRunnable --> TypeSystem
    Runner --> ErrorHandling
    Runner --> Callbacks

// runner 执行引擎结构
// 位置: compose/graph_run.go:41
type runner struct {
    // 图结构
    chanSubscribeTo     map[string]*chanCall      // 通道订阅映射
    controlPredecessors map[string][]string       // 控制前驱
    dataPredecessors    map[string][]string       // 数据前驱
    successors          map[string][]string       // 后继节点
    
    // 执行控制
    inputChannels *chanCall                      // 输入通道
    eager         bool                           // 是否急切执行
    dag           bool                           // 是否为DAG模式
    
    // 类型信息
    inputType     reflect.Type                   // 输入类型
    outputType    reflect.Type                   // 输出类型
    genericHelper *genericHelper                 // 泛型助手
    
    // 处理器管理
    preBranchHandlerManager *preBranchHandlerManager // 分支前处理器管理
    preNodeHandlerManager   *preNodeHandlerManager   // 节点前处理器管理
    edgeHandlerManager      *edgeHandlerManager      // 边处理器管理
    
    // 运行时配置
    runCtx        func(ctx context.Context) context.Context // 运行时上下文
    chanBuilder   chanBuilder                               // 通道构建器
    mergeConfigs  map[string]FanInMergeConfig              // 合并配置
    
    // 中断和检查点
    checkPointer          *checkPointer    // 检查点管理
    interruptBeforeNodes  []string         // 前置中断节点
    interruptAfterNodes   []string         // 后置中断节点
    options              graphCompileOptions // 编译选项
}

// run 核心执行引擎 - 框架性能的关键热点
// 位置: compose/graph_run.go:107
func (r *runner) run(ctx context.Context, isStream bool, input any, opts ...Option) (result any, err error) {
    // 🔥 热点 1: 回调处理 - 每次执行都会调用
    ctx, input = onGraphStart(ctx, input, isStream)
    defer func() {
        if err != nil {
            ctx, err = onGraphError(ctx, err)  // 🔥 错误处理热点
        } else {
            ctx, result = onGraphEnd(ctx, result, isStream)  // 🔥 结束处理热点
        }
    }()
    
    // 🔥 热点 2: 管理器初始化 - 每次执行都需要
    cm := r.initChannelManager(isStream)     // 🔥 通道管理器创建
    tm := r.initTaskManager(runWrapper, getGraphCancel(ctx), opts...)  // 🔥 任务管理器创建
    
    // 计算初始任务
    nextTasks, result, isEnd, err := r.calculateNextTasks(ctx, []*task{{
        nodeKey: START,
        call:    r.inputChannels,
        output:  input,
    }}, isStream, cm, optMap)
    
    if isEnd {
        return result, nil
    }

    // 🔥 热点 3: 主执行循环 - 最大的性能瓶颈
    for step := 0; step < maxSteps; step++ {
        // 检查上下文取消
        select {
        case <-ctx.Done():
            return nil, newGraphRunError(ctx.Err())
        default:
        }
        
        // 🔥 热点 3.1: 任务调度
        if len(nextTasks) == 0 {
            break
        }
        
        // 🔥 热点 3.2: 并发任务执行
        err := tm.submit(nextTasks)  // 🔥🔥 最大热点
        if err != nil {
            return nil, newGraphRunError(err)
        }

        // 🔥 热点 3.3: 等待任务完成
        completedTasks, canceled, canceledTasks := tm.wait()  // 🔥 同步等待开销
        if canceled {
            return nil, r.handleInterrupt(ctx, canceledTasks, cm, optMap)
        }

        // 🔥 热点 3.4: 计算下一批任务
        nextTasks, result, isEnd, err = r.calculateNextTasks(ctx, completedTasks, isStream, cm, optMap)
        if err != nil {
            return nil, newGraphRunError(err)
        }
        
        if isEnd {
            return result, nil
        }
    }
    
    return result, nil
}

sequenceDiagram
    participant U as 用户
    participant C as Chain/Graph
    participant G as graph (内部)
    participant CR as ComposableRunnable
    participant R as Runner
    
    U->>C: NewChain/NewGraph()
    C->>G: 创建内部图结构
    
    U->>C: AddNode/AppendXX()
    C->>G: 添加节点到图中
    G->>G: 类型检查与验证
    
    U->>C: Compile(ctx)
    C->>G: compile(ctx, options)
    G->>G: 构建执行计划
    G->>G: 优化图结构
    G->>CR: 创建可组合执行器
    CR->>R: 创建运行器
    G-->>C: 返回 Runnable
    C-->>U: 返回编译结果

sequenceDiagram
    participant U as 用户
    participant R as Runnable
    participant Runner as Runner
    participant CM as ChannelManager
    participant TM as TaskManager
    participant Node as GraphNode
    
    U->>R: Invoke(ctx, input)
    R->>Runner: run(ctx, false, input)
    Runner->>CM: 初始化通道管理器
    Runner->>TM: 初始化任务管理器
    
    loop 执行步骤
        Runner->>TM: 获取就绪任务
        TM-->>Runner: 返回可执行节点列表
        
        par 并行执行节点
            Runner->>Node: 执行节点1
            Runner->>Node: 执行节点2
        end
        
        Node-->>Runner: 返回执行结果
        Runner->>CM: 更新通道数据
        CM->>CM: 检查后续节点就绪状态
    end
    
    Runner-->>R: 返回最终结果
    R-->>U: 返回执行结果

graph TD
    subgraph "智能体接口层"
        Agent[Agent 智能体接口]
        Runner[Runner 运行器]
        ResumableAgent[ResumableAgent 可恢复智能体]
    end
    
    subgraph "智能体实现层"
        ChatModelAgent[ChatModelAgent 聊天模型智能体]
        FlowAgent[FlowAgent 流程智能体]
        WorkflowAgent[WorkflowAgent 工作流智能体]
        ReactAgent[ReactAgent ReAct智能体]
    end
    
    subgraph "协调机制层"
        SubAgentManager[SubAgent Manager 子智能体管理]
        TransferMechanism[Transfer Mechanism 转移机制]
        HistoryRewriter[History Rewriter 历史重写]
    end
    
    subgraph "状态管理层"
        State[State 状态管理]
        SessionValues[Session Values 会话值]
        CheckPointStore[CheckPoint Store 检查点存储]
        InterruptInfo[Interrupt Info 中断信息]
    end
    
    subgraph "工具集成层"
        AgentTool[Agent Tool 智能体工具]
        ToolsConfig[Tools Config 工具配置]
        ExitTool[Exit Tool 退出工具]
    end
    
    subgraph "异步处理层"
        AsyncIterator[AsyncIterator 异步迭代器]
        AsyncGenerator[AsyncGenerator 异步生成器]
        EventStream[Event Stream 事件流]
    end
    
    %% 接口层关系
    Agent --> ChatModelAgent
    Agent --> FlowAgent
    Runner --> Agent
    ResumableAgent --> Agent
    
    %% 实现层关系
    FlowAgent --> WorkflowAgent
    FlowAgent --> ReactAgent
    ChatModelAgent --> State
    
    %% 协调机制关系
    FlowAgent --> SubAgentManager
    FlowAgent --> TransferMechanism
    FlowAgent --> HistoryRewriter
    
    %% 状态管理关系
    Runner --> CheckPointStore
    Agent --> SessionValues
    ResumableAgent --> InterruptInfo
    
    %% 工具集成关系
    ChatModelAgent --> ToolsConfig
    AgentTool --> Agent
    
    %% 异步处理关系
    Runner --> AsyncIterator
    Agent --> AsyncGenerator
    AsyncIterator --> EventStream

// Agent 智能体核心接口
// 位置: adk/interface.go
type Agent interface {
    Name(ctx context.Context) string
    Description(ctx context.Context) string
    Run(ctx context.Context, input *AgentInput, options ...AgentRunOption) *AsyncIterator[*AgentEvent]
}

// AgentInput 智能体输入
type AgentInput struct {
    Messages        []Message
    EnableStreaming bool
}

// AgentEvent 智能体事件
type AgentEvent struct {
    AgentName string
    RunPath   []RunStep
    Output    *AgentOutput
    Action    *AgentAction
    Err       error
}

// ChatModelAgent 聊天模型智能体
// 位置: adk/chatmodel.go
type ChatModelAgent struct {
    config *ChatModelAgentConfig
    model  model.BaseChatModel
    tools  []tool.BaseTool
}

// ChatModelAgentConfig 配置结构
type ChatModelAgentConfig struct {
    Model         model.BaseChatModel  // 聊天模型
    Tools         []tool.BaseTool      // 工具列表
    MaxIterations int                  // 最大迭代次数
    Temperature   float64              // 生成温度
    SystemPrompt  string               // 系统提示
}

// NewChatModelAgent 创建聊天模型智能体
// 位置: adk/chatmodel.go:179
func NewChatModelAgent(_ context.Context, config *ChatModelAgentConfig) (*ChatModelAgent, error) {
    // 验证配置
    if config.Model == nil {
        return nil, errors.New("model is required")
    }

    // 创建智能体实例
    agent := &ChatModelAgent{
        config: config,
        model:  config.Model,
        tools:  config.Tools,
    }

    // 设置默认配置
    if agent.config.MaxIterations == 0 {
        agent.config.MaxIterations = 10
    }

    return agent, nil
}

// Run 执行智能体，返回异步事件流
// 位置: adk/chatmodel.go
func (a *ChatModelAgent) Run(ctx context.Context, input *AgentInput, opts ...Option) *AsyncIterator[*AgentEvent] {
    // 创建异步生成器
    generator := NewAsyncGenerator[*AgentEvent]()

    // 启动执行协程
    go func() {
        defer generator.Close()

        // 初始化状态
        state := &State{
            Messages:    input.Messages,
            Iterations:  0,
            MaxIterations: a.config.MaxIterations,
        }

        // 执行循环
        for state.Iterations < state.MaxIterations {
            // 生成回复
            response, err := a.model.Generate(ctx, state.Messages)
            if err != nil {
                generator.SendError(err)
                return
            }

            // 检查工具调用
            if hasToolCalls(response) {
                // 执行工具
                toolResults, err := a.executeTools(ctx, response.ToolCalls)
                if err != nil {
                    generator.SendError(err)
                    return
                }

                // 更新消息历史
                state.Messages = append(state.Messages, response)
                state.Messages = append(state.Messages, toolResults...)
                state.Iterations++

                // 发送工具执行事件
                generator.Send(&AgentEvent{
                    Type: EventTypeToolExecution,
                    Data: toolResults,
                })
            } else {
                // 发送最终回复事件
                generator.Send(&AgentEvent{
                    Type: EventTypeResponse,
                    Data: response,
                })
                return
            }
        }
    }()

    return generator.Iterator()
}

// Agent ReAct智能体结构
// 位置: flow/agent/react/agent.go
type Agent struct {
    runnable         compose.Runnable[[]*schema.Message, *schema.Message]
    graph            *compose.Graph[[]*schema.Message, *schema.Message]
    graphAddNodeOpts []compose.GraphAddNodeOpt
}

// state ReAct状态结构
type state struct {
    Messages                 []*schema.Message
    ReturnDirectlyToolCallID string
}

// NewAgent 创建ReAct智能体
// 位置: flow/agent/react/agent.go
func NewAgent(ctx context.Context, config *AgentConfig) (*Agent, error) {
    // 1. 准备组件
    chatModel, err := agent.ChatModelWithTools(config.Model, config.ToolCallingModel, toolInfos)
    if err != nil {
        return nil, err
    }

    toolsNode, err := compose.NewToolNode(ctx, &config.ToolsConfig)
    if err != nil {
        return nil, err
    }

    // 2. 构建图
    graph := compose.NewGraph[[]*schema.Message, *schema.Message](
        compose.WithGenLocalState(func(ctx context.Context) *state {
            return &state{Messages: make([]*schema.Message, 0, config.MaxStep+1)}
        }))

    // 3. 添加模型节点
    modelPreHandle := func(ctx context.Context, input []*schema.Message, state *state) ([]*schema.Message, error) {
        state.Messages = append(state.Messages, input...)
        if config.MessageModifier != nil {
            return config.MessageModifier(ctx, state.Messages), nil
        }
        return state.Messages, nil
    }

    err = graph.AddChatModelNode(nodeKeyModel, chatModel, 
        compose.WithStatePreHandler(modelPreHandle))
    if err != nil {
        return nil, err
    }

    // 4. 添加工具节点
    toolsNodePreHandle := func(ctx context.Context, input *schema.Message, state *state) (*schema.Message, error) {
        if input != nil {
            state.Messages = append(state.Messages, input)
            state.ReturnDirectlyToolCallID = getReturnDirectlyToolCallID(input, config.ToolReturnDirectly)
        }
        return input, nil
    }

    err = graph.AddToolsNode(nodeKeyTools, toolsNode, 
        compose.WithStatePreHandler(toolsNodePreHandle))
    if err != nil {
        return nil, err
    }

    // 5. 添加分支逻辑
    modelPostBranchCondition := func(ctx context.Context, sr *schema.StreamReader[*schema.Message]) (string, error) {
        isToolCall, err := config.StreamToolCallChecker(ctx, sr)
        if err != nil {
            return "", err
        }
        if isToolCall {
            return nodeKeyTools, nil
        }
        return compose.END, nil
    }

    err = graph.AddBranch(nodeKeyModel, compose.NewStreamGraphBranch(
        modelPostBranchCondition, 
        map[string]bool{nodeKeyTools: true, compose.END: true}))
    if err != nil {
        return nil, err
    }

    // 6. 编译图
    runnable, err := graph.Compile(ctx, compileOpts...)
    if err != nil {
        return nil, err
    }

    return &Agent{
        runnable: runnable,
        graph:    graph,
    }, nil
}