M11 Eino 框架

mkdir eino-hello
cd eino-hello
go mod init eino-hello
go get github.com/cloudwego/eino
go get github.com/cloudwego/eino-ext/components/model/openai

export DEEPSEEK_API_KEY="your-api-key"
export DEEPSEEK_BASE_URL="https://api.deepseek.com"
export DEEPSEEK_MODEL="deepseek-v4-pro"

package main

import (
	"context"
	"fmt"
	"log"
	"os"

	"github.com/cloudwego/eino-ext/components/model/openai"
	"github.com/cloudwego/eino/schema"
)

func main() {
	ctx := context.Background()

	cm, err := openai.NewChatModel(ctx, &openai.ChatModelConfig{
		APIKey:  os.Getenv("DEEPSEEK_API_KEY"),
		BaseURL: os.Getenv("DEEPSEEK_BASE_URL"),
		Model:   os.Getenv("DEEPSEEK_MODEL"),
	})
	if err != nil {
		log.Fatal(err)
	}

	input := []*schema.Message{
		schema.SystemMessage("你是一个Go语言编程助手，擅长回答Go语言相关技术问题。"),
		schema.UserMessage("用两句话介绍 Eino 框架。"),
	}

	output, err := cm.Generate(ctx, input)
	if err != nil {
		log.Fatal(err)
	}

	fmt.Println(output.Content)
}

OpenAI ChatModel 实现
        ↓
[]*schema.Message
        ↓ Generate
*schema.Message

reader, err := cm.Stream(ctx, input)
if err != nil {
	log.Fatal(err)
}
defer reader.Close()

for {
	chunk, err := reader.Recv()
	if errors.Is(err, io.EOF) {
		break
	}
	if err != nil {
		log.Fatal(err)
	}
	fmt.Print(chunk.Content)
}

type BaseModel[M messageType] interface {
	Generate(ctx context.Context, input []M, opts ...Option) (M, error)

	Stream(ctx context.Context, input []M, opts ...Option) (*schema.StreamReader[M], error)
}

type BaseChatModel = BaseModel[*schema.Message]

type ToolCallingChatModel interface {
	BaseChatModel
	WithTools(tools []*schema.ToolInfo) (ToolCallingChatModel, error)
}

schema.SystemMessage("你是客服助手。")
schema.UserMessage("查询订单 ORD-1001")
schema.AssistantMessage("请提供订单号", nil)
schema.ToolMessage(
	"{\"status\":\"shipped\"}",
	"call_123",
	schema.WithToolName("query_order"),
)

answer, err := cm.Generate(
	ctx,
	messages,
	model.WithTemperature(0.2),        // 设置温度
	model.WithMaxTokens(1200),         // 设置最大 token 数
	model.WithTopP(0.9),               // 设置 top_p 值
	model.WithStop([]string{"<END>"}), // 设置停止词
)

type ChatTemplate interface {
	Format(ctx context.Context, vs map[string]any, opts ...Option) ([]*schema.Message, error)
}

tpl := prompt.FromMessages(
	schema.FString,
	schema.SystemMessage(
		"你是{brand}客服。只依据已知资料回答。",
	),
	schema.MessagesPlaceholder("history", true),
	schema.UserMessage(
		"资料：\n{context}\n\n问题：{question}",
	),
)

messages, err := tpl.Format(ctx, map[string]any{
	"brand":    "Aurora",
	"history":  history,
	"context":  retrievedContext,
	"question": userQuestion,
})

chain := compose.NewChain[map[string]any, *schema.Message]()
chain.
	AppendChatTemplate(tpl).
	AppendChatModel(cm)

app, err := chain.Compile(ctx)
if err != nil {
	log.Fatalf("invoke failed, err:%v", err)
}
answer, err := app.Invoke(ctx, variables)

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

// InvokableTool 可调用的工具接口，支持同步调用
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)
}

// EnhancedInvokableTool 是支持返回结构化多模态结果的工具接口
// 与返回字符串的 InvokableTool 不同，此接口返回 *schema.ToolResult
// 可以包含文本、图片、音频、视频和文件
type EnhancedInvokableTool interface {
	BaseTool
	InvokableRun(ctx context.Context, argument *schema.ToolArgument, opts ...Option) (*schema.ToolResult, error)
}

// EnhancedStreamableTool 是支持返回结构化多模态结果的流式工具接口
// 提供流式读取器以逐步访问多模态内容
type EnhancedStreamableTool interface {
	BaseTool
	StreamableRun(ctx context.Context, argument *schema.ToolArgument, opts ...Option) (*schema.StreamReader[*schema.ToolResult], error)
}

type OrderArgs struct {
	OrderID string `json:"order_id" jsonschema:"required" jsonschema_description:"订单号，例如 ORD-1001"`
}

type Order struct {
	OrderID string `json:"order_id"`
	Info    string `json:"info"`
	Status  int    `json:"status"`
}

func QueryOrder(ctx context.Context, args OrderArgs) (*Order, error) {
	customerID := auth.CustomerID(ctx)
	order, err := orderService.Query(
		ctx,
		customerID,
		args.OrderID,
	)
	if err != nil {
		return nil, err
	}
	return order, nil
}

queryOrderTool, err := utils.InferTool(
	"query_order",
	"查询当前登录客户的订单，包含商品信息、订单状态、金额和运单号等。",
	QueryOrder,
)

// compose/tool_node.go

// run tools using `Invoke`
func (tn *ToolsNode) Invoke(ctx context.Context, input *schema.Message,
    opts ...ToolsNodeOption) ([]*schema.Message, error)
    
// run tools using `Stream`
func (tn *ToolsNode) Stream(ctx context.Context, input *schema.Message,
    opts ...ToolsNodeOption) (*schema.StreamReader[[]*schema.Message], error)

// compose/tool_node.go

type ToolsNodeConfig struct {
    Tools []tool.BaseTool

    UnknownToolsHandler func(ctx context.Context, name, input string) (string, error)

    ExecuteSequentially bool

    ToolArgumentsHandler func(ctx context.Context, name, arguments string) (string, error)

    ToolCallMiddlewares []ToolMiddleware
}

// schema/message.go

type Message struct {
    // role should be 'assistant' for tool call message
    Role RoleType `json:"role"`

    // here each `ToolCall` is generated by ChatModel and to be executed by ToolsNode
    ToolCalls []ToolCall `json:"tool_calls,omitempty"`
    
    // other fields...
}

// ToolCall is the tool call in a message.
// It's used in Assistant Message when there are tool calls should be made.
type ToolCall struct {
    // Index is used when there are multiple tool calls in a message.
    // In stream mode, it's used to identify the chunk of the tool call for merging.
    Index *int `json:"index,omitempty"`
    // ID is the id of the tool call, it can be used to identify the specific tool call.
    ID string `json:"id"`
    // Type is the type of the tool call, default is "function".
    Type string `json:"type"`
    // Function is the function call to be made.
    Function FunctionCall `json:"function"`

    // Extra is used to store extra information for the tool call.
    Extra map[string]any `json:"extra,omitempty"`
}

// FunctionCall is the function call in a message.
// It's used in Assistant Message.
type FunctionCall struct {
    // Name is the name of the function to call, it can be used to identify the specific function.
    Name string `json:"name,omitempty"`
    // Arguments is the arguments to call the function with, in JSON format.
    Arguments string `json:"arguments,omitempty"`
}

toolsNode, err := compose.NewToolNode(
	ctx,
	&compose.ToolsNodeConfig{
		Tools: []tool.BaseTool{
			queryOrderTool,
			xxxTool,
		},
		ExecuteSequentially: true,
	},
)

// Mock LLM 输出作为输入
input := &schema.Message{
	Role: schema.Assistant,
	ToolCalls: []schema.ToolCall{
		{
			Function: schema.FunctionCall{
				Name:      "query_order",
				Arguments: `{"order_id": "lwz-1234567"}`,
			},
		},
	},
}

toolMessages, err := toolsNode.Invoke(ctx, input)

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

type Source struct {
	URI string
}

type Document struct {
	ID       string
	Content  string
	MetaData map[string]any
}

Loader：定位、下载或打开数据
Parser：解析 PDF、HTML、DOCX、Markdown
Document：统一输出

docs, err := docLoader.Load(ctx, document.Source{
	URI: "/data/refund-policy.pdf",
})

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

父文档
  → 去除导航、页眉和页脚
  → 按标题或段落切分
  → 控制 chunk token 长度
  → 添加 overlap
  → 继承元数据
  → chunk Documents

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

vectors, err := embedder.EmbedStrings(ctx, []string{
	"七天无理由退货政策",
	"订单物流查询",
})

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

import (
	"github.com/cloudwego/eino-ext/components/embedding/ark"
	"github.com/cloudwego/eino-ext/components/indexer/milvus2"
	"github.com/cloudwego/eino/schema"
	"github.com/milvus-io/milvus/client/v2/milvusclient"
)


// 创建 embedding 模型
emb, err := ark.NewEmbedder(ctx, &ark.EmbeddingConfig{
	APIKey:     os.Getenv("DOUBAO_API_KEY"),
	Model:      "doubao-embedding-vision-251215",
	APIType:    new(ark.APITypeMultiModal),
	Dimensions: &defaultDim,
})
if err != nil {
	log.Fatalf("Failed to create embedding: %v", err)
	return err
}
// 创建索引器
indexer, err := milvus2.NewIndexer(ctx, &milvus2.IndexerConfig{
	ClientConfig: &milvusclient.ClientConfig{
		Address: "localhost:19530",
	},
	Collection: "my_collection",
	Vector: &milvus2.VectorConfig{
		Dimension:    2048,
		MetricType:   milvus2.COSINE,
		IndexBuilder: milvus2.NewAutoIndexBuilder(),
	},
	Embedding: emb,
})
if err != nil {
	slog.WarnContext(ctx, "newIndexer failed", "err", err)
	return err
}

// 索引文档
docs := []*schema.Document{
	{
		ID:      "doc1",
		Content: "EINO 是一个用于构建人工智能应用的框架",
		MetaData: map[string]any{
			"category": "database",
			"year":     2021,
		},
	},
	{
		ID:      "doc2",
		Content: "Aurora品牌耳机拆封后概不退货。",
		MetaData: map[string]any{
			"category": "database",
			"year":     2026,
		},
	},
	{
		ID:      "doc3",
		Content: "《宠物厌食症状识别与干预》 —— 当宠物连续 24 小时拒食或显著减食，常见原因有应激反应、消化系统疾病、口腔问题……",
		MetaData: map[string]any{
			"category": "database",
			"year":     2013,
		},
	},
}
ids, err := indexer.Store(ctx, docs)
if err != nil {
	slog.WarnContext(ctx, "indexer store failed", "err", err)
	return err
}
fmt.Printf("ids:%#v\n", ids)
return nil
}

ids, err := idx.Store(
	ctx,
	docs,
	indexer.WithSubIndexes([]string{"customer_service"}),
	indexer.WithEmbedding(embedder),
)

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

import (
	"github.com/cloudwego/eino-ext/components/embedding/ark"
	"github.com/cloudwego/eino-ext/components/retriever/milvus2"
	"github.com/cloudwego/eino-ext/components/retriever/milvus2/search_mode"
	"github.com/milvus-io/milvus/client/v2/milvusclient"
)
// 创建 embedding 模型
emb, err := ark.NewEmbedder(ctx, &ark.EmbeddingConfig{
	APIKey:     os.Getenv("DOUBAO_API_KEY"),
	Model:      "doubao-embedding-vision-251215",
	APIType:    new(ark.APITypeMultiModal),
	Dimensions: &defaultDim,
})
if err != nil {
	log.Fatalf("Failed to create embedding: %v", err)
	return
}

// 创建 retriever
retriever, err := milvus2.NewRetriever(ctx, &milvus2.RetrieverConfig{
	ClientConfig: &milvusclient.ClientConfig{
		Address: "localhost:19530",
	},
	Collection: "my_collection",
	TopK:       10,
	SearchMode: search_mode.NewApproximate(milvus2.COSINE),
	Embedding:  emb,
})
if err != nil {
	log.Fatalf("Failed to create retriever: %v", err)
	return
}
log.Printf("Retriever created successfully")

// 检索文档
documents, err := retriever.Retrieve(ctx, "耳机拆了试戴不合适能退吗？")

// eino/compose/types_lambda.go

type Lambda struct {
    executor *composableRunnable
}

type Invoke[I, O, TOption any] func(ctx context.Context, input I, opts ...TOption) (O, error)

type Stream[I, O, TOption any] func(ctx context.Context, input I, opts ...TOption) (*schema.StreamReader[O], error)

type Collect[I, O, TOption any] func(ctx context.Context, input *schema.StreamReader[I], opts ...TOption) (O, error)

type Transform[I, O, TOption any] func(ctx context.Context, input *schema.StreamReader[I], opts ...TOption) (*schema.StreamReader[O], error)

import (
	"github.com/cloudwego/eino/compose"
	"github.com/cloudwego/eino/schema"
)
formatDocs := compose.InvokableLambda(
	func(_ context.Context, docs []*schema.Document) (string, error) {
		var b strings.Builder
		for i, doc := range docs {
			fmt.Fprintf(&b, "[%d] %s\n", i+1, doc.Content)
		}
		return b.String(), nil
	},
)

// Lambda 是 Chain/Graph 中的节点，需要先加入 Chain 并编译。
chain := compose.NewChain[[]*schema.Document, string]()
chain.AppendLambda(formatDocs)

runnable, err := chain.Compile(ctx)
if err != nil {
	return fmt.Errorf("compile chain: %w", err)
}

docs := []*schema.Document{
	{ID: "doc-1", Content: "Eino 是一个大模型应用开发框架。"},
	{ID: "doc-2", Content: "Lambda 可以包装自定义的数据处理函数。"},
}

result, err := runnable.Invoke(ctx, docs)
if err != nil {
	return fmt.Errorf("invoke chain: %w", err)
}

fmt.Println(result)

Content
ReasoningContent
UserInputMultiContent / AssistantGenMultiContent
ToolCalls / ToolCallID
ResponseMeta / Extra

type AgenticMessage struct {
	Role          AgenticRoleType
	ContentBlocks []*ContentBlock
	ResponseMeta  *AgenticResponseMeta
	Extra         map[string]any
}

type AgenticModel = BaseModel[*schema.AgenticMessage]

type AgenticChatTemplate interface {
	Format(ctx context.Context, vs map[string]any, opts ...Option) ([]*schema.AgenticMessage, error)
}

Node：Component、Lambda 或子图
Edge：执行依赖与数据通道
State：单次运行共享状态
Branch：运行时选择后继
Runnable：Compile 后的统一执行接口

声明节点 → 连接节点 → Compile → Invoke/Stream

输入 → ChatTemplate → ChatModel → Parser → 输出

tpl := prompt.FromMessages(
	schema.FString,
	schema.SystemMessage(
		"你是售后客服，只依据资料回答。",
	),
	schema.UserMessage(
		"资料：{context}\n问题：{question}",
	),
)

chain := compose.NewChain[
	map[string]any,
	*schema.Message,
]()

chain.
	AppendChatTemplate(tpl).
	AppendChatModel(cm)

app, err := chain.Compile(ctx)
if err != nil {
	return err
}

result, err := app.Invoke(ctx, map[string]any{
	"context":  "激活后的耳机不支持七天无理由退货。",
	"question": "耳机激活后还能无理由退货吗？",
})

map[string]any
  ↓ ChatTemplate
[]*schema.Message
  ↓ ChatModel
*schema.Message

                       ┌→ retrieve_policy ─┐
question → classify ──┼→ query_order ─────┼→ answer
                       └→ ask_detail ──────┘

g := compose.NewGraph[string, string]()

normalize := compose.InvokableLambda(
	func(ctx context.Context, input string) (string, error) {
		return strings.TrimSpace(input), nil
	},
)

reply := compose.InvokableLambda(
	func(ctx context.Context, input string) (string, error) {
		return "已收到：" + input, nil
	},
)

if err := g.AddLambdaNode("normalize", normalize); err != nil {
	return err
}
if err := g.AddLambdaNode("reply", reply); err != nil {
	return err
}

if err := g.AddEdge(compose.START, "normalize"); err != nil {
	return err
}
if err := g.AddEdge("normalize", "reply"); err != nil {
	return err
}
if err := g.AddEdge("reply", compose.END); err != nil {
	return err
}

app, err := g.Compile(ctx)

branch := compose.NewGraphBranch(
	func(
		ctx context.Context,
		msg *schema.Message,
	) (string, error) {
		if len(msg.ToolCalls) > 0 {
			return "tools", nil
		}
		return compose.END, nil
	},
	map[string]bool{
		"tools":     true,
		compose.END: true,
	},
)

if err := g.AddBranch("model", branch); err != nil {
	return err
}

generate → evaluate
   ↑          │
   └─ revise ←┘

rewrite.query ───────────┐
retrieve.documents ──────┼→ AnswerInput
profile.user_profile ────┘

Retriever.Invoke
  ↓ 完整 Documents
FormatDocs.Invoke
  ↓ 完整 context
ChatTemplate.Invoke
  ↓ 完整 Messages
ChatModel.Stream
  ↓ Message chunks
SSE.Transform
  ↓ HTTP chunks

result, err := app.Invoke(
	ctx,
	input,
	compose.WithCallbacks(handler),
)

result, err := app.Invoke(
	ctx,
	input,
	// 所有 ChatModel 节点都能看到该 Option。
	compose.WithChatModelOption(
		model.WithTemperature(0.1),
	),
	// 只传给顶层 key 为 retrieve_policy 的 Retriever。
	compose.WithRetrieverOption(
		retriever.WithTopK(8),
	).DesignateNode("retrieve_policy"),
	// 定位到子图中的具体节点。
	compose.WithCallbacks(traceHandler).DesignateNodeWithPath(
		compose.NewNodePath("rag_subgraph", "rerank"),
	),
	// 覆盖本次有环图运行步数上限。
	compose.WithRuntimeMaxSteps(30),
)

节点请求外部输入
  ↓ Interrupt
保存拓扑位置和 State
  ↓ CheckpointStore
取得审批结果
  ↓ Resume
从中断位置继续

ChatModel / Tool / Retriever / Indexer / Transformer
                         ↓
                 Chain / Graph 编排
                         ↓
                    Flow 模式
                         ↓
       直接调用 / 作为组件 / 作为子图 / 包装为 Lambda

START
  ↓
ChatModel ── 无 ToolCall ─────────────→ END
  │
  └─ 有 ToolCall → ToolsNode
                     │
                     ├─ ReturnDirectly → END
                     └─ 普通结果 → ChatModel

reactFlow, err := react.NewAgent(ctx, &react.AgentConfig{
	ToolCallingModel: cm,
	ToolsConfig: compose.ToolsNodeConfig{
		Tools:               tools,
		ExecuteSequentially: true,
	},
	MessageRewriter: func(
		ctx context.Context,
		messages []*schema.Message,
	) []*schema.Message {
		// 可以在这里压缩历史；返回值会写回 Flow State。
		return messages
	},
	MaxStep: 12,
	ToolReturnDirectly: map[string]struct{}{
		"finish_report": {},
	},
})
if err != nil {
	return err
}

answer, err := reactFlow.Generate(ctx, []*schema.Message{
	schema.SystemMessage("你是企业客服。业务数据必须来自工具。"),
	schema.UserMessage("查询订单 ORD-1001 的物流"),
})
if err != nil {
	return err
}
fmt.Println(answer.Content)

answer, err := reactFlow.Generate(
	ctx,
	messages,
	react.WithChatModelOptions(model.WithTemperature(0.1)),
)

原始 query
   ↓ RewriteLLM 或 RewriteHandler
query_1  query_2  query_3
   ↓        ↓        ↓
并发调用 OrigRetriever
   └────────┼────────┘
            ↓
      FusionFunc / 按 ID 去重

multiRetriever, err := multiquery.NewRetriever(
	ctx,
	&multiquery.Config{
		RewriteLLM:    rewriteModel,
		OrigRetriever: baseRetriever,
		MaxQueriesNum: 3,
		// RewriteTemplate、LLMOutputParser、FusionFunc 为空时使用默认实现。
	},
)
if err != nil {
	return err
}

docs, err := multiRetriever.Retrieve(ctx, "耳机拆封后能否七天无理由退货")

routedRetriever, err := router.NewRetriever(ctx, &router.Config{
	Retrievers: map[string]retriever.Retriever{
		"policy":  policyRetriever,
		"product": productRetriever,
		"order":   orderRetriever,
	},
	Router: func(ctx context.Context, query string) ([]string, error) {
		switch {
		case strings.Contains(query, "订单"):
			return []string{"order"}, nil
		case strings.Contains(query, "退货"):
			return []string{"policy", "product"}, nil
		default:
			return []string{"policy", "product"}, nil
		}
	},
})

入库：父文档 → Transformer → 子文档
                      ├─ Metadata[parent_id] = 父文档 ID
                      ├─ 生成唯一子文档 ID
                      └─ 写入底层 Indexer

查询：query → 召回子文档 → 提取并去重 parent_id
                             ↓
                     OrigDocGetter(ids)
                             ↓
                         完整父文档

chain := compose.NewChain[string, []*schema.Document]()
chain.AppendRetriever(multiRetriever)

app, err := chain.Compile(ctx)
if err != nil {
	return err
}
docs, err := app.Invoke(ctx, query)

reactGraph, graphOpts := reactFlow.ExportGraph()
graphOpts = append(
	graphOpts,
	compose.WithNodeKey("customer_service_react"),
)

outer := compose.NewChain[[]*schema.Message, *schema.Message]()
outer.AppendGraph(reactGraph, graphOpts...)

app, err := outer.Compile(ctx)

                    ┌─ 无 ToolCall → Host 直接回答
用户消息 → Host ────┤
                    ├─ 一个 Specialist → 直接返回专家结果
                    └─ 多个 Specialists → 并行执行 → Summarizer

ChatModel + Tool + ToolsNode
           ↓
编排循环、状态、流、Callback
           ↓
ChatModelAgent + Runner + AgentEvent

type TypedAgent[M MessageType] interface {
	Name(ctx context.Context) string
	Description(ctx context.Context) string
	Run(ctx context.Context, input *TypedAgentInput[M], options ...AgentRunOption) *AsyncIterator[*TypedAgentEvent[M]]
}

type Agent = TypedAgent[*schema.Message]

ChatModelAgent
  = Model
  + Instruction
  + Tools
  + ReAct Loop
  + Middleware

customerService, err := adk.NewChatModelAgent(
	ctx,
	&adk.ChatModelAgentConfig{
		Name:        "customer_service",
		Description: "处理政策、订单、物流与工单问题。",
		Instruction: "你是企业客服。政策问题先检索知识库；" +
			"订单与物流数据必须来自工具；" +
			"创建工单前必须获得明确确认；" +
			"资料不足时追问，禁止编造。",
		Model:         cm,
		MaxIterations: 12,
		ToolsConfig: adk.ToolsConfig{
			ToolsNodeConfig: compose.ToolsNodeConfig{
				Tools:               tools,
				ExecuteSequentially: true,
			},
		},
	},
)

Reason：模型阅读目标、历史、观察和工具定义
  ↓
Action：模型生成 ToolCall
  ↓
Act：ToolsNode 执行 Tool
  ↓
Observation：Tool Result 加入消息
  ↓
模型继续 Reason
  ↓
没有 ToolCall 时输出最终回答

runner := adk.NewRunner(ctx, adk.RunnerConfig{
	Agent:           customerService,
	EnableStreaming: true,
	CheckPointStore: checkpointStore,
})

iter := runner.Query(
	ctx,
	"查询订单 ORD-1001 的物流",
)

for {
	event, ok := iter.Next()
	if !ok {
		break
	}
	if event.Err != nil {
		return event.Err
	}
	consume(event)
}

iter := runner.Run(
	ctx,
	append(history, schema.UserMessage(question)),
	adk.WithSessionValues(map[string]any{
		"customer_id": "CUST-001",
		"tenant_id":   "TENANT-A",
	}),
)

researcher, err := adk.NewChatModelAgent(
	ctx,
	&adk.ChatModelAgentConfig{
		Name:        "researcher",
		Description: "搜索资料并返回带来源的摘要。",
		Instruction: "只返回结论、证据与来源。",
		Model:       cm,
		ToolsConfig: adk.ToolsConfig{
			ToolsNodeConfig: compose.ToolsNodeConfig{
				Tools: []tool.BaseTool{searchTool},
			},
		},
	},
)
if err != nil {
	return err
}

researchTool := adk.NewAgentTool(ctx, researcher)

type AuditHandler struct {
	*adk.BaseChatModelAgentMiddleware
}

func (h *AuditHandler) BeforeModelRewriteState(
	ctx context.Context,
	state *adk.ChatModelAgentState,
	_ *adk.ModelContext,
) (context.Context, *adk.ChatModelAgentState, error) {
	// 可在这里压缩 state.Messages，或依据权限筛选 state.ToolInfos。
	// 修改后的 state 会用于本轮模型调用，并保留到后续迭代。
	return ctx, state, nil
}

audit := &AuditHandler{
	BaseChatModelAgentMiddleware: &adk.BaseChatModelAgentMiddleware{},
}

agent, err := adk.NewChatModelAgent(ctx, &adk.ChatModelAgentConfig{
	Name:     "customer_service",
	Model:    cm,
	Handlers: []adk.ChatModelAgentMiddleware{audit},
})

Agent 准备执行高风险 Tool
  ↓
Tool、Graph 节点或自定义 Agent 产生 Interrupt
  ↓
Runner 将状态写入 CheckPointStore
  ↓
外部审批系统取得用户决定
  ↓
Resume 携带审批结果
  ↓
Agent 从中断点继续

Components / Graph / ADK
  ├─ Callback → Trace / Metrics / Logs
  ├─ Debug protocol → Eino Dev
  ├─ Run result → Evaluation
  └─ Prompt version → Prompt Platform

run customer_service
├─ model decide
│  ├─ input/output token
│  └─ time_to_first_token
├─ tool search_knowledge
│  ├─ tenant/filter/top_k
│  └─ returned_documents
├─ model answer
└─ ticket confirmation

go get github.com/cloudwego/eino-ext/devops@v0.1.9

if err := devops.Init(ctx); err != nil {
	return err
}

// devops.Init 在 Compile 之前调用。
app, err := graph.Compile(ctx)

客户输入
  ↓
Runner
  ↓
ChatModelAgent
  ├─ search_knowledge
  │    └─ Milvus Hybrid Search
  │         ├─ dense vector search
  │         ├─ BM25 sparse search
  │         ├─ tenant filter
  │         └─ RRF + lexical rerank
  ├─ query_order
  ├─ query_logistics
  ├─ draft_ticket
  └─ confirm_ticket
  ↓
AgentEvent / MessageStream
  ↓
当前 CLI；可扩展为 HTTP/SSE

code/11-customer-service-assistant/
├── cmd/
│   ├── assistant/       # 多轮客服
│   ├── indexer/         # 知识入库
│   └── search/          # 独立测试检索
├── fixtures/
│   ├── orders.json
│   └── shipments.json
├── internal/
│   ├── agent/           # Agent 和事件消费
│   ├── business/        # 订单、物流、工单领域逻辑
│   ├── config/          # 环境配置
│   ├── embedding/       # hash 与 OpenAI-compatible embedding
│   ├── knowledge/       # Markdown 加载和切块
│   ├── rag/             # Milvus 入库与混合检索
│   └── tools/           # Eino Tool 适配
├── knowledge/           # 政策和产品知识
└── docker-compose.yml   # Milvus Standalone

type Chunk struct {
	ChunkID string
	DocID   string
	Title   string
	Source  string
	Content string
}

Markdown
  → Load
  → 按标题切块
  → 清洗空白
  → 生成稳定 ID
  → dense embedding
  → Milvus Upsert
  → BM25 Function 生成 sparse vector

export EMBEDDING_PROVIDER=openai
export EMBEDDING_API_KEY='your-api-key'
export EMBEDDING_BASE_URL='https://api.openai.com/v1'
export EMBEDDING_MODEL='text-embedding-3-small'
export EMBEDDING_DIMENSION=1536

analyzer := map[string]any{
	"tokenizer": "jieba",
}

contentField := entity.NewField().
	WithName("content").
	WithDataType(entity.FieldTypeVarChar).
	WithMaxLength(8192).
	WithEnableAnalyzer(true).
	WithAnalyzerParams(analyzer).
	WithEnableMatch(true)

bm25 := entity.NewFunction().
	WithName("content_bm25").
	WithType(entity.FunctionTypeBM25).
	WithInputFields("content").
	WithOutputFields("sparse_vector")

denseReq := milvusclient.NewAnnRequest(
	"dense_vector",
	candidates,
	entity.FloatVector(queryVector),
).WithFilter(filter)

sparseReq := milvusclient.NewAnnRequest(
	"sparse_vector",
	candidates,
	entity.Text(query),
).WithFilter(filter)

results, err := client.HybridSearch(
	ctx,
	milvusclient.NewHybridSearchOption(
		collection,
		candidates,
		denseReq,
		sparseReq,
	).
		WithOutputFields(
			"doc_id",
			"title",
			"source",
			"content",
		).
		WithReranker(
			milvusclient.NewRRFReranker(),
		),
)

func (r *Repository) Order(
	customerID string,
	orderID string,
) (Order, error) {
	order, ok := r.orders[orderID]
	if !ok || order.CustomerID != customerID {
		return Order{}, ErrOrderNotFound
	}
	return order, nil
}

draft_ticket
  → 生成分类、优先级、摘要和 draft_id
  → Agent 展示草稿
  → 用户确认或取消
  → confirm_ticket
  → 创建 ticket_id 或返回取消状态

type Dependencies struct {
	TenantID   string
	CustomerID string
	Knowledge  rag.Store
	Business   *business.Repository
	Tickets    *business.TicketService
}

searchKnowledge, err := utils.InferTool(
	"search_knowledge",
	"检索企业知识库中的退换货、保修、物流、产品使用和工单 SLA 资料，返回引用来源。涉及政策和产品说明时优先调用。",
	func(ctx context.Context, args searchKnowledgeArgs) (string, error) {
		limit := args.TopK
		if limit <= 0 {
			limit = 4
		}
		if limit > 6 {
			limit = 6
		}

		hits, err := deps.Knowledge.Search(
			ctx, deps.TenantID, args.Query, limit,
		)
		if err != nil {
			return "", err
		}
		return encode(struct {
			Query string    `json:"query"`
			Hits  []rag.Hit `json:"hits"`
		}{Query: args.Query, Hits: hits})
	},
)

serviceAgent, err := adk.NewChatModelAgent(
	ctx,
	&adk.ChatModelAgentConfig{
		Name:        "enterprise_customer_service",
		Description: "企业智能客服与工单助手。",
		Instruction: instruction,
		Model:       cm,
		MaxIterations: 12,
		ToolsConfig: adk.ToolsConfig{
			ToolsNodeConfig: compose.ToolsNodeConfig{
				Tools:               tools,
				ExecuteSequentially: true,
			},
		},
	},
)
if err != nil {
	return nil, err
}

runner := adk.NewRunner(ctx, adk.RunnerConfig{
	Agent:           serviceAgent,
	EnableStreaming: true,
})

messages := append(clone(history), schema.UserMessage(query))
responses, err := serviceagent.ConsumeEvents(
	runner.Run(ctx, messages),
	output,
	debug,
)
if err != nil {
	return history, err
}
history = append(messages, responses...)

cd code/11-customer-service-assistant
docker compose up -d
docker compose ps

go run ./cmd/indexer --recreate

go run ./cmd/search \
  -query 'Aurora Pro 耳机拆封使用后，七天内可以无理由退货吗？' \
  -top-k 3

export OPENAI_API_KEY='your-api-key'
export OPENAI_BASE_URL='https://api.deepseek.com'
export OPENAI_MODEL_NAME='deepseek-chat'

go run ./cmd/assistant -customer-id CUST-001

go test -race ./...
go vet ./...

docker compose config --quiet

query
  ├─ rewrite
  ├─ dense embedding → dense search
  └─ original query → BM25 search
            ↓
        RRF merge
            ↓
          rerank
            ↓
      citation formatter