Eino 的“你好世界“背后发生了什么（第48篇-E34）

作者：袖梨 2026-07-10

读完这篇你会知道

入口：两层薄壳

调用 chatModel.Generate(ctx, messages) 时，进入的是 eino-ext/components/model/openai/chatmodel.go：

// components/model/openai/chatmodel.gofunc (cm *ChatModel) Generate(ctx context.Context, in []*schema.Message, opts ...model.Option) (outMsg *schema.Message, err error) {ctx = callbacks.EnsureRunInfo(ctx, cm.GetType(), components.ComponentOfChatModel)out, err := cm.cli.Generate(ctx, in, opts...)// ← 直接袋里到内层 Clientif err != nil {return nil, convOrigAPIError(err)}return out, nil}

ChatModel 是薄壳，只做两件事：

callbacks.EnsureRunInfo——往 context 里塞一个 RunInfo（组件类型 = ChatModel），让 callback 系统知道"这是哪个组件在运行"
调 cm.cli.Generate——实际的 acl/openai.Client

第一层：genRequest — 三个转换

acl/openai.Client.Generate 第一步调 genRequest，它做三件事：

1. 参数合并（Options 优先级）

options := model.GetCommonOptions(&model.Options{Temperature: c.config.Temperature,// 构建时的默认值Model: &c.config.Model,// ...}, opts...)// 运行时传入的 opts 会覆盖默认值

GetCommonOptions 实现优先级：运行时 opts > 构建时 config。如果你调 Generate(ctx, msgs, model.WithTemperature(0.0))，温度会覆盖配置文件的值。

2. `[]*schema.Message` → `[]openai.ChatCompletionMessage`

Eino 的 schema.Message 是跨框架的统一类型，OpenAI 有自己的 ChatCompletionMessage。转换发生在 genRequest 里：

for _, inMsg := range in {// 纯文本消息（最常见）msg = openai.ChatCompletionMessage{Role:toOpenAIRole(inMsg.Role),// schema.User → "user"Content: inMsg.Content,ToolCalls:toOpenAIToolCalls(inMsg.ToolCalls),ToolCallID: inMsg.ToolCallID,}// 多模态消息（图片/音频）走另外两个函数msgs = append(msgs, msg)}

角色映射：

`schema.RoleType`	OpenAI 字符串
`schema.User`	`"user"`
`schema.Assistant`	`"assistant"`
`schema.System`	`"system"`
`schema.Tool`	`"tool"`

3. `[]*schema.ToolInfo` → JSON Schema

如果绑定了工具，toTools 把 ToolInfo 转成 OpenAI 的工具格式：

func toTools(tis []*schema.ToolInfo) ([]tool, error) {for _, ti := range tis {paramsJSONSchema, _ := ti.ParamsOneOf.ToJSONSchema()// ← 生成 JSON Schematools = append(tools, tool{Function: &functionDefinition{Name:ti.Name,Description: ti.Desc,Parameters:paramsJSONSchema,},})}}

这就是 utils.InferTool 的 struct tag（jsonschema_description）最终变成 HTTP 请求里工具描述的路径。

第二层：HTTP 请求

genRequest 组装好 openai.ChatCompletionRequest 结构体后，Generate 调：

resp, err := c.cli.CreateChatCompletion(ctx, *req, reqOpts...)

c.cli 是 go-openai SDK 的 openai.Client。它把请求体序列化成 JSON（用 bytedance/sonic，比标准库快），发到：

POST https://api.deepseek.com/v1/chat/completionsContent-Type: application/jsonAuthorization: Bearer <api_key>{"model": "deepseek-chat","messages": [{"role": "system", "content": "你是一个助手"},{"role": "user", "content": "北京今天天气怎么样？"}],"temperature": 1.0,"tools": [{"type": "function","function": {"name": "get_weather","description": "查询指定城市的天气","parameters": {"type": "object","properties": {"city": {"type": "string","description": "城市名，如北京、上海"}},"required": ["city"]}}}]}

响应回来后，buildGenerateResponse 取 Choices[0]，把 OpenAI 字符串角色转回 schema.RoleType，填入 FinishReason 和 token 用量，返回 *schema.Message。

callback 插入点

Generate 里有三个 callback 调用点：

func (c *Client) Generate(...) {// 1. 发起请求前ctx = callbacks.OnStart(ctx, cbInput)// cbInput 含消息列表、工具、配置resp, err := c.cli.CreateChatCompletion(...)// ← HTTP 请求在这里// 2. 出错时if err != nil { callbacks.OnError(ctx, err) }// 3. 正常返回时callbacks.OnEnd(ctx, &model.CallbackOutput{Message:outMsg,TokenUsage: toModelCallbackUsage(outMsg.ResponseMeta),})}

这就是 Langfuse / 自定义 callback 能拿到"这次调用的 token 用量"的原因——token 数从 resp.Usage 里读，在 OnEnd 时推给所有已注册的 handler。

Stream 路径：goroutine + Pipe

Stream 模式有所不同：

func (c *Client) Stream(ctx context.Context, in []*schema.Message, opts ...model.Option) {req, _, _, _, _ = c.genRequest(ctx, in, opts...)req.Stream = true // 开关req.StreamOptions = &openai.StreamOptions{IncludeUsage: true}// 最后一帧带用量stream, _ := c.cli.CreateChatCompletionStream(ctx, *req)// 建立 SSE 连接sr, sw := schema.Pipe[*model.CallbackOutput](1) // 内部 channel，容量 1go func() {defer sw.Close()for {chunk, err := stream.Recv()// 逐帧从 SSE 读if err == io.EOF { return }msg, found, _ := builder.build(chunk) // chunk → schema.Messageif !found { continue }sw.Send(&model.CallbackOutput{Message: msg}, nil)}}()// 把 channel 的读端包成 StreamReader 返回outStream = schema.StreamReaderWithConvert(nsr, ...)return outStream, nil}

关键设计：

schema.Pipe 是一个有 buffer 的 channel，sw 写端在 goroutine 里，sr 读端包成 StreamReader 返回给调用方
调用方 for { msg, err := sr.Recv() } 本质上是从 channel 读，每读到一帧就能处理（打字机效果）
defer sr.Close()：必须调用，否则写端 goroutine 卡在 sw.Send() 永远不退出（goroutine 泄漏）

完整调用链

你的代码：chatModel.Generate(ctx, messages)│▼[components/model/openai/ChatModel.Generate]callbacks.EnsureRunInfo(ctx, ...) ← 塞 RunInfo 到 context│▼[libs/acl/openai/Client.Generate]genRequest(ctx, in, opts...)├── GetCommonOptions(...)← 合并运行时/构建时参数├── 消息类型转换: schema.Message → openai.ChatCompletionMessage├── 工具转换: ToolInfo → JSON Schema → openai.Tool└── 组装 openai.ChatCompletionRequestcallbacks.OnStart(ctx, cbInput) ← 触发所有 handler.OnStart│▼[go-openai SDK / http.Client]sonic.Marshal(req) → JSON bodyPOST /v1/chat/completionsAuthorization: Bearer xxx│▼ (HTTP response)[buildGenerateResponse]resp.Choices[0].Message → schema.Message角色字符串 → schema.RoleType提取 FinishReason + TokenUsage│▼callbacks.OnEnd(ctx, output)← 触发所有 handler.OnEnd（含 token 用量）│▼返回 *schema.Message 给你

Stream 路径在"POST"之后换成 SSE + goroutine + Pipe。