系列「企业级 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 CheckPointStoreAgentCallback把 HookRunner Streamer 变成 callbacks.HandlertokenStreamReader把 Eino StreamReader 变成 application 层的 TokenReaderReActFactory把这五个组装起来加 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 ChatModeltypechatModelAdapterstruct{client port.LLMClient// DeepFlux 内部 LLM 调用接口profilestring// LLM 配置名如 deepseek-v3tenantIDstring// 计费 / 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);oksid!{sessionIDsid}// ...}Runnable 会被缓存复用——同一份编译好的图可能服务多个 session。a.sessionID是构建时绑定的如果用它就会把 session A 的 LLM 调用算到 session B 头上。所以运行时优先从 context 读context 里没有才 fallback 到构建时的值。Generate是流的消费者不直接调 LLM 同步接口而是调Stream然后收集 chunks 合并。DeepFlux 的 LLMClient 只有流式实现这样两个接口共用同一条路径。三、toolBrokerAdapter工具调用路由 名字净化ToolsFromBroker把ToolBroker.Schemas()返回的清单变成[]tool.InvokableToolfuncToolsFromBroker(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];okorig!s.Name{returnnil,fmt.Errorf(tool name collision: %q and %q both → %q,orig,s.Name,sanitized)}seen[sanitized]s.Name toolsappend(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_searchEino 的 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.HandlerNewAgentCallback用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()}两条主线SSE 推流工具调用开始时推turn.tool_call结束时推turn.tool_result浏览器实时看到工具执行状态。HookRunner 7 阶段BeforeModelCall、AfterModelCall、BeforeToolUse、AfterToolUse、OnError对应 hook 链里的审计 / PII 脱敏 / 合规检查阶段。OnEndWithStreamOutput里只做output.Close()——流式 model 输出的 token 消费在 ReAct 循环里直接处理不在 callback 里重复消费但StreamReader 必须被 Close否则 goroutine 泄漏。六、tokenStreamReader把 Eino 流变成 port.TokenReadertypetokenStreamReaderstruct{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{optsappend(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}returntokenStreamReader{sr:sr},nil,nil}三种返回值语义返回值含义(TokenReader, nil, nil)正常执行读流拿 token(nil, *AgentInterruptInfo, nil)HITL 中断等待人工决策(nil, nil, error)执行失败Runnable 缓存图编译是昂贵操作需要加载工具 schema、构建 graph。buildRunnable用{tenantID, configName, llmProfile, maxTurns, temperature, toolWhitelist, ...}作 key缓存编译好的compose.RunnableTTL 35 分钟。35 分钟 30 分钟中断超时保证 HITL 恢复时一定命中缓存——如果缓存 miss就要重新调ToolsFromBrokergRPC在恢复路径上增加额外故障面。小结五个适配器一个工厂把 Eino 的所有细节封在infrastructure/einoadapter包里适配器从到chatModelAdapterport.LLMClient自研einomodel.ToolCallingChatModeltoolBrokerAdapterport.ToolBroker自研tool.InvokableTool× NSessionCheckpointStoredomain.SessionRepoPostgrescompose.CheckPointStoreAgentCallbackport.HookRunnerport.Streamercallbacks.HandlertokenStreamReaderschema.StreamReader[*schema.Message]port.TokenReader两个贯穿所有适配器的设计决策运行时从 context 读 sessionID——Runnable 缓存复用时不能用构建时绑定的值接口在 application 层port定义实现在 infrastructure 层——这让整个 agent BC 可以不知道 Eino 是什么换框架只改 einoadapter下篇看ReActFactory.StreamTurn的完整上下文——ForceNewRun标志背后的 Eino 机制。代码来源DeepFlux platform ·server/internal/agent/infrastructure/einoadapter/