系列「企业级 AI Agent 实现拆解」E29 篇。前面 28 篇把 Eino 的机制讲清楚了。这篇和下篇换个角度：看 DeepFlux 实际怎么用 Eino——具体在server/internal/agent/infrastructure/einoadapter/这个目录里，五个适配器做了什么。

读完这篇你会知道

• 为什么适配器在 infrastructure 层，不在 application 层
• chatModelAdapter：把多租户 LLMClient 变成 Eino model 接口
• toolBrokerAdapter：把 ToolBroker 变成 Eino InvokableTool，顺手解决工具名碰撞
• SessionCheckpointStore：把 SessionRepo 变成 Eino CheckPointStore
• AgentCallback：把 HookRunner + Streamer 变成 callbacks.Handler
• tokenStreamReader：把 Eino StreamReader 变成 application 层的 TokenReader
• ReActFactory：把这五个组装起来，加 Runnable 缓存

一、为什么需要适配器

DDD 的 D4 规则：跨层依赖要倒置。application 层不能直接 import Eino 的具体 API。

application/command/run_turn.go ↓ 只看 port 接口（框架无关） application/port/ports.go LLMClient / ToolBroker / TokenReader / ReActRunnableFactory ↑ 只看 port 接口 infrastructure/einoadapter/ chatModelAdapter / toolBrokerAdapter / SessionCheckpointStore / AgentCallback / tokenStreamReader / ReActFactory ↑ 只这里 import Eino

一句话：Eino 的类型只在einoadapter包里出现，application 层和 domain 层完全不知道 Eino 的存在。换掉 Eino 只需改 infrastructure 层，不碰业务逻辑。

二、chatModelAdapter：多租户 LLM → Eino ChatModel

typechatModelAdapterstruct{client port.LLMClient// DeepFlux 内部 LLM 调用接口profilestring// LLM 配置名，如 "deepseek-v3"tenantIDstring// 计费 / RLS 隔离用sessionIDstring// 绑定到 session（可被 context 覆盖）temperaturefloat32maxTokensintcachedSchemas[]llm.ToolSchema// WithTools 时预计算，避免 per-request 双重 JSON 转换}

实现了einomodel.ToolCallingChatModel——这是 Eino ReAct 需要的接口（比ChatModel多一个WithTools方法）。

WithTools不可变：返回新实例，不修改已有实例。这符合 Eino compose 的期望——图编译时WithTools一次，运行时不再变。同时预计算cachedSchemas，把*schema.ToolInfo → llm.ToolSchema的 JSON 转换做在构建期，不在每次请求时做。

Stream里的 sessionID 读取顺序：

func(a*chatModelAdapter)buildRequest(ctx context.Context,input[]*schema.Message)llm.ChatRequest{sessionID:=a.sessionIDifsid,ok:=ctx.Value(port.ContextKeyAgentSessionID{}).(string);ok&&sid!=""{sessionID=sid}// ...}

Runnable 会被缓存复用——同一份编译好的图可能服务多个 session。a.sessionID是构建时绑定的，如果用它就会把 session A 的 LLM 调用算到 session B 头上。所以运行时优先从 context 读，context 里没有才 fallback 到构建时的值。

Generate是流的消费者：不直接调 LLM 同步接口，而是调Stream然后收集 chunks 合并。DeepFlux 的 LLMClient 只有流式实现，这样两个接口共用同一条路径。

三、toolBrokerAdapter：工具调用路由 + 名字净化

ToolsFromBroker把ToolBroker.Schemas()返回的清单变成[]tool.InvokableTool：

funcToolsFromBroker(ctx context.Context,broker port.ToolBroker,tenantID,sessionID,agentConfigstring)([]tool.InvokableTool,error){schemas,err:=broker.Schemas(ctx,tenantID,agentConfig)// ...seen:=make(map[string]string)// sanitized name → 原始名for_,s:=rangeschemas{sanitized:=sanitizeToolName(s.Name)iforig,ok:=seen[sanitized];ok&&orig!=s.Name{returnnil,fmt.Errorf("tool name collision: %q and %q both → %q",orig,s.Name,sanitized)}seen[sanitized]=s.Name tools=append(tools,&toolBrokerAdapter{...})}}

为什么要净化名字：OpenAI / DeepSeek 的工具名只接受[a-zA-Z0-9_-]，不接受点号。DeepFlux 的工具按<namespace>.<name>命名（如kb.search），点号直接发给 LLM API 会报错。

funcsanitizeToolName(namestring)string{returnstrings.ReplaceAll(name,".","_")}

碰撞检测在构建期：如果kb.search和kb_search同时存在，净化后都是kb_search，Eino 的 tool dispatch 会静默让后者覆盖前者，导致调用 A 实际执行 B。这里在ToolsFromBroker里提前检测，fail fast，而不是等 runtime 出诡异 bug。

InvokableRun里同样优先读 context 的 sessionID，理由和 chatModelAdapter 一样——Runnable 缓存复用时不能用构建时绑定的值。

四、SessionCheckpointStore：把 DB 变成 Eino 的持久化后端

Eino 的compose.CheckPointStore接口只有两个方法：

typeCheckPointStoreinterface{Get(ctx context.Context,checkPointIDstring)([]byte,bool,error)Set(ctx context.Context,checkPointIDstring,data[]byte)error}

SessionCheckpointStore把domain.SessionRepo包一层就能对接：

typeSessionCheckpointStorestruct{repo domain.SessionRepo}func(s*SessionCheckpointStore)Get(ctx context.Context,idstring)([]byte,bool,error){returns.repo.LoadCheckpoint(ctx,model.SessionID(id))}func(s*SessionCheckpointStore)Set(ctx context.Context,idstring,data[]byte)error{returns.repo.SaveCheckpoint(ctx,model.SessionID(id),data)}

checkPointID = session UUID。每次图执行后，Eino 把整个图状态序列化存进sessions.checkpoint_blob列（Postgres）。HITL 中断后恢复时，Eino 从这里加载状态继续执行，不需要从头跑。

五、AgentCallback：把 Hook 和 SSE 变成 callbacks.Handler

NewAgentCallback用callbacks.HandlerBuilder构建，不实现完整 Handler 接口，只注册需要的几个时机：

funcNewAgentCallback(runner port.HookRunner,streamer port.Streamer,sid model.SessionID,onPhasefunc(model.Phase),onToolEndfunc(context.Context,string))callbacks.Handler{b:=callbacks.NewHandlerBuilder()b.OnStartFn(func(ctx context.Context,info*callbacks.RunInfo,input callbacks.CallbackInput)context.Context{switchinfo.Component{casecomponents.ComponentOfChatModel:runner.BeforeModelCall(ctx,sid,extractLastUserContent(msgs))casecomponents.ComponentOfTool:runner.BeforeToolUse(ctx,sid,tc)streamer.Emit(ctx,sid,port.StreamEvent{Type:"turn.tool_call",Payload:...})onPhase(model.PhaseToolUse)}returnctx})b.OnEndFn(func(...)context.Context{switchinfo.Component{casecomponents.ComponentOfChatModel:runner.AfterModelCall(ctx,sid,co.Message.Content)casecomponents.ComponentOfTool:runner.AfterToolUse(ctx,sid,tc,co.Response)streamer.Emit(ctx,sid,port.StreamEvent{Type:"turn.tool_result",Payload:...})onPhase(model.PhaseThinking)}returnctx})b.OnEndWithStreamOutputFn(func(ctx context.Context,info*callbacks.RunInfo,output*schema.StreamReader[...])context.Context{ifoutput!=nil{output.Close()// 必须关，否则 goroutine 泄漏}returnctx})b.OnErrorFn(func(...)context.Context{runner.OnError(ctx,sid,err)returnctx})returnb.Build()}

两条主线：

1. SSE 推流：工具调用开始时推turn.tool_call，结束时推turn.tool_result，浏览器实时看到工具执行状态。

2. HookRunner 7 阶段：BeforeModelCall、AfterModelCall、BeforeToolUse、AfterToolUse、OnError对应 hook 链里的审计 / PII 脱敏 / 合规检查阶段。

OnEndWithStreamOutput里只做output.Close()——流式 model 输出的 token 消费在 ReAct 循环里直接处理，不在 callback 里重复消费，但StreamReader 必须被 Close，否则 goroutine 泄漏。

六、tokenStreamReader：把 Eino 流变成 port.TokenReader

typetokenStreamReaderstruct{sr*schema.StreamReader[*schema.Message]}func(r*tokenStreamReader)Recv()(string,error){msg,err:=r.sr.Recv()iferr!=nil{return"",err}ifmsg!=nil{returnmsg.Content,nil}return"",nil}func(r*tokenStreamReader)Close(){r.sr.Close()}

application 层的port.TokenReader只有Recv() (string, error)和Close()——它不知道*schema.Message是什么。

tokenStreamReader把 Eino 的*schema.Message拆成纯 string，让run_turn.go里的 SSE 推送循环完全不依赖 Eino 类型。

七、ReActFactory：组装 + 缓存

ReActFactory.StreamTurn是所有适配器的组装点：

func(f*ReActFactory)StreamTurn(ctx context.Context,cfg port.AgentConfig,...)(port.TokenReader,*port.AgentInterruptInfo,error){runnable,err:=f.buildRunnable(ctx,cfg,tenantID,string(sid))cb:=NewAgentCallback(hooks,stream,sid,in.OnPhaseChange,in.OnToolEnd)opts:=[]compose.Option{compose.WithCallbacks(cb),compose.WithCheckPointID(string(sid)),}if!in.IsResume{opts=append(opts,compose.WithForceNewRun())}// resume 时预检 checkpoint 存在，fail fastifin.IsResume{if_,found,_:=f.cpStore.Get(ctx,string(sid));!found{returnnil,nil,port.ErrCheckpointMissing}}sr,err:=runnable.Stream(ctx,inputMsgs,opts...)iferr!=nil{ifinfo,ok:=compose.ExtractInterruptInfo(err);ok{returnnil,&port.AgentInterruptInfo{BeforeNodes:info.BeforeNodes},nil}returnnil,nil,err}return&tokenStreamReader{sr:sr},nil,nil}

三种返回值语义：

Runnable 缓存：图编译是昂贵操作（需要加载工具 schema、构建 graph）。buildRunnable用{tenantID, configName, llmProfile, maxTurns, temperature, toolWhitelist, ...}作 key，缓存编译好的compose.Runnable，TTL 35 分钟。

35 分钟 > 30 分钟（中断超时），保证 HITL 恢复时一定命中缓存——如果缓存 miss，就要重新调ToolsFromBrokergRPC，在恢复路径上增加额外故障面。

小结

五个适配器，一个工厂，把 Eino 的所有细节封在infrastructure/einoadapter包里：

两个贯穿所有适配器的设计决策：

1. 运行时从 context 读 sessionID——Runnable 缓存复用时不能用构建时绑定的值
2. 接口在 application 层（port）定义，实现在 infrastructure 层——这让整个 agent BC 可以不知道 Eino 是什么，换框架只改 einoadapter

下篇看ReActFactory.StreamTurn的完整上下文——ForceNewRun标志背后的 Eino 机制。

代码来源：DeepFlux platform ·server/internal/agent/infrastructure/einoadapter/