组件接口 – Willow 的开发笔记

Eino

组件接口

Eino 的 Component 接口定义了一组可替换、可组合的能力单元。所有接口都在 eino/components/ 下，具体实现在 eino-ext 中提供。

组件全景

文档流水线（RAG）:                对话流水线:
                                 
  Loader                           ChatTemplate
    │                                │
    ▼                                ▼
  Transformer (分割/过滤)           ChatModel
    │                                │
    ▼                                ▼
  Embedding (转向量)                Tool / ToolsNode
    │                                
    ▼                                
  Indexer (存入向量库)              
    │                                
    ▼                                
  Retriever (语义搜索召回)

ChatModel — LLM 调用入口

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)
}

// 支持 Tool Calling 的模型，可绑定工具
type ToolCallingChatModel interface {
    BaseChatModel
    WithTools(tools []*schema.ToolInfo) ChatModel
}

方法	说明
`Generate`	同步调用，等完整回复
`Stream`	流式调用，逐 chunk 返回

⚡ 优先用 Stream 提升体验；Generate 适合非实时场景。

eino-ext 已支持: OpenAI、Claude、Gemini、Ark（字节方舟）、Ollama、DeepSeek、Qwen 等。

// 同步
msg, _ := model.Generate(ctx, []*schema.Message{
    schema.SystemMessage("你是编程专家"),
    schema.UserMessage("用 Go 写 Hello World"),
})

// 流式
stream, _ := model.Stream(ctx, messages)
defer stream.Close()
for {
    chunk, err := stream.Recv()
    if err == io.EOF { break }
    fmt.Print(chunk.Content)
}

Tool — LLM 可调用的函数

Tool 是连接 LLM 和外部世界的桥梁。Eino 定义了三级接口：

BaseTool          ← 所有 Tool 必须：提供 Info（名称、描述、参数 Schema）
  ├── InvokableTool   ← 同步执行，返回 string
  └── StreamableTool  ← 流式执行，返回 StreamReader[string]

type BaseTool interface {
    Info(ctx context.Context) (*schema.ToolInfo, error)
}

type InvokableTool interface {
    BaseTool
    InvokableRun(ctx context.Context, argumentsInJSON string, opts ...Option) (string, error)
}

type StreamableTool interface {
    BaseTool
    StreamableRun(ctx context.Context, argumentsInJSON string, opts ...Option) (*schema.StreamReader[string], error)
}

💡 至少实现 InvokableTool 和 StreamableTool 之一。

ToolInfo — 给 LLM 看的"说明书"

type ToolInfo struct {
    Name        string        // 工具名称（LLM 用这个指定调哪个）
    Desc        string        // 功能描述（LLM 据此判断何时调用——写不清楚就不会被调用）
    ParamsOneOf *ParamsOneOf  // 参数的 JSON Schema（LLM 据此生成参数）
}

完整自定义 Tool 示例

// ── 1. 定义 Tool 结构体 ──
type WeatherTool struct {
    apiKey string
}

// ── 2. 实现 Info()——最关键：LLM 靠它理解你的 Tool ──
func (t *WeatherTool) Info(ctx context.Context) (*schema.ToolInfo, error) {
    return &schema.ToolInfo{
        Name: "get_weather",
        Desc: "查询指定城市当天的天气情况。返回温度、湿度、天气状况等。",
        // 参数 Schema：告诉 LLM 需要什么输入
        ParamsOneOf: schema.NewParamsOneOfByParams(map[string]*schema.ParameterInfo{
            "city": {
                Type:     schema.String,
                Desc:     "城市名称，如 'Beijing', 'Shanghai'",
                Required: true,
            },
            "unit": {
                Type:     schema.String,
                Desc:     "温度单位：celsius 或 fahrenheit",
                Enum:     []string{"celsius", "fahrenheit"},
                Required: false,
            },
        }),
    }, nil
}

// ── 3. 实现 InvokableRun()——工具的执行逻辑 ──
func (t *WeatherTool) InvokableRun(ctx context.Context, argsJSON string, opts ...tool.Option) (string, error) {
    // 解析 LLM 传入的 JSON 参数
    var args struct {
        City string `json:"city"`
        Unit string `json:"unit"`
    }
    if err := json.Unmarshal([]byte(argsJSON), &args); err != nil {
        return "", fmt.Errorf("参数解析失败: %w", err)
    }
    if args.Unit == "" {
        args.Unit = "celsius"
    }

    // 调用真实 API
    weather, err := t.fetchWeather(ctx, args.City, args.Unit)
    if err != nil {
        // 返回错误信息——LLM 会看到这个错误并尝试修正
        return "", fmt.Errorf("获取天气失败: %w", err)
    }

    // 返回自然语言的工具结果——LLM 会把这段文字作为上下文继续推理
    result := fmt.Sprintf("%s 今天天气：温度 %.1f°%s，湿度 %d%%，%s",
        args.City, weather.Temp, args.Unit, weather.Humidity, weather.Description)
    return result, nil
}

// ── 4. (可选) 实现 StreamableRun()——流式返回结果 ──
func (t *WeatherTool) StreamableRun(ctx context.Context, argsJSON string, opts ...tool.Option) (*schema.StreamReader[string], error) {
    result, err := t.InvokableRun(ctx, argsJSON, opts...)
    if err != nil {
        return nil, err
    }
    return schema.StreamReaderFromArray([]string{result}), nil
}

Tool 中的中断——Human-in-the-Loop

func (t *ApprovalTool) InvokableRun(ctx context.Context, argsJSON string, opts ...tool.Option) (string, error) {
    // 需要人工审批 → 返回中断错误
    return "", compose.NewInterruptAndRerunErr(&adk.InterruptInfo{
        Data: map[string]any{"status": "awaiting_approval", "request": argsJSON},
    })
    // Resume 时会重新执行此 Tool
}

Retriever — 语义检索

type Retriever interface {
    Retrieve(ctx context.Context, query string, opts ...Option) ([]*schema.Document, error)
}

Option	说明
`TopK`	返回文档数上限
`ScoreThreshold`	最低相似度阈值
`Embedding`	提供后先将 query 转向量再检索
`DSLInfo`	传递底层查询 DSL（如 Elasticsearch DSL）

Embedding — 文本转向量

type Embedder interface {
    EmbedStrings(ctx context.Context, texts []string, opts ...Option) ([][]float64, error)
}

每个文本返回一个 []float64，维度由模型决定。换模型后维度可能不同，需重建向量索引。

embedder, _ := openai.NewEmbedder(ctx, &openai.EmbeddingConfig{Model: "text-embedding-3-small"})
vectors, _ := embedder.EmbedStrings(ctx, []string{"Hello", "你好"})
for i, v := range vectors {
    fmt.Printf("text[%d] dimension: %d\n", i, len(v))
}

Indexer — 文档入库

type Indexer interface {
    Store(ctx context.Context, docs []*schema.Document, opts ...Option) (ids []string, err error)
}

返回存储成功的文档 ID 列表。常见实现：Milvus、Elasticsearch、Pinecone 等。

Loader — 文档加载

type Loader interface {
    Load(ctx context.Context, src Source, opts ...LoaderOption) ([]*schema.Document, error)
}

Source 描述文档来源（文件路径、URL、S3 URI 等）。格式解析委派给 Parser：

// 通过 WithParserOptions 配置解析器
docs, _ := loader.Load(ctx, source,
    document.WithParserOptions(parser.WithChunkSize(1000)),
)

Transformer — 文档处理

type Transformer interface {
    Transform(ctx context.Context, docs []*schema.Document, opts ...TransformerOption) ([]*schema.Document, error)
}

常见操作：按 chunk 分割、按规则过滤、跨文档合并、结果重排序。

// 示例：splitter 将长文档拆成小段
splitter, _ := filesplitter.New(ctx, &filesplitter.Config{ChunkSize: 500, Overlap: 50})
chunks, _ := splitter.Transform(ctx, docs)

ChatTemplate — Prompt 模板

// f-string 格式
template := prompt.FromMessages(schema.FString,
    schema.SystemMessage("你是{role}"),
    schema.UserMessage("问题：{question}"),
)
messages, _ := template.Format(ctx, map[string]any{
    "role": "数据分析师", "question": "Q2 营收趋势",
})

占位符 {key} → params 中取值；需输出文字 { 用 {{ 转义。

Lambda — 自定义函数节点

类型	签名	谁用
`InvokableLambda[I,O]`	`func(ctx, I) (O, error)`	Graph.AddLambdaNode
`StreamableLambda[I,O]`	`func(ctx, I) (*StreamReader[O], error)`	同上
`TransformableLambda[I,O]`	`func(ctx, StreamReader[I]) (StreamReader[O], error)`	同上

fn := compose.InvokableLambda(func(ctx context.Context, in string) (string, error) {
    return strings.ToUpper(in), nil
})

组件接入编排

每个 Component 都是 Graph 中的一个 Node：

graph.AddChatModelNode(   "llm",      chatModel)
graph.AddToolsNode(        "tools",    toolsNode)
graph.AddRetrieverNode(    "retrieve", retriever)
graph.AddEmbeddingNode(    "embed",    embedder)
graph.AddIndexerNode(      "index",    indexer)
graph.AddLoaderNode(       "load",     loader)
graph.AddLambdaNode(       "format",   formatFn)

RAG 完整流水线示例

文档 → Loader → Transformer(splitter) → Embedding → Indexer(向量库)
                                                       │
                                                       ▼
用户问题 → Embedding → Retriever(检索) → 拼接上下文 → ChatModel → 回复

func buildRAGPipeline(ctx context.Context, model model.BaseChatModel, embedder embedding.Embedder, indexer indexer.Indexer, retriever retriever.Retriever) (compose.Runnable[string, *schema.Message], error) {
    graph := compose.NewGraph[string, *schema.Message]()

    // 检索节点：query → documents
    graph.AddRetrieverNode("retrieve", retriever)

    // Lambda：拼接 prompt（检索结果 + 原问题）
    graph.AddLambdaNode("format", compose.InvokableLambda(
        func(ctx context.Context, docs []*schema.Document) ([]*schema.Message, error) {
            context := strings.Join(mapDocs(docs), "\n")
            return []*schema.Message{
                schema.SystemMessage("根据以下资料回答问题:\n" + context),
                schema.UserMessage("待回答的问题将作为后续输入"),
            }, nil
        },
    ))

    // LLM 节点
    graph.AddChatModelNode("llm", model)

    graph.AddEdge(compose.START, "retrieve")
    graph.AddEdge("retrieve", "format")
    graph.AddEdge("format", "llm")
    graph.AddEdge("llm", compose.END)

    return graph.Compile(ctx)
}

常见陷阱

陷阱	说明
🚨 Stream 未 Close	`stream.Close()` 必须调用，否则 goroutine 泄漏
🚨 ToolInfo.Desc 不清楚	LLM 靠 `Desc` 判断何时调用工具，模糊则 Tool 白注册
🚨 Tool 参数未标 Required	LLM 可能不传关键参数，导致 InvokableRun 解析失败
🚨 Embedding 维度不匹配	换 Embedding 模型后向量维度变了，旧索引查不出结果，须重建
🚨 Tool 返回的错误信息不合适	错误信息会被 LLM 看到；缺乏上下文时 LLM 无法自行修正

概述与安装 Schema 与消息类型