系列「企业级 AI Agent 实现拆解」E27 篇。上一篇讲了 流程可视化：把 Eino 编排图变成 Mermaid 图表。这篇讲交互式调试——在不修改业务代码的情况下，从任意节点注入输入、逐节点观察输出。

读完这篇你会知道

• devops.Init(ctx) 怎么在进程内起一个调试 HTTP 服务
• 全局编译回调如何自动捕获所有已编译的图
• 五个核心 API 端点：从拓扑查询到流式执行
• 从任意节点开始运行：BuildDevGraph 的子图重建逻辑
• NodeDebugState：每个节点的输入、输出、耗时、Token 消耗
• any 类型输入的特殊 JSON 格式
• AppendType 注册自定义类型

---

一、问题：图跑不对，怎么定位

Agent 调用链出问题时，最常见的痛点是：不知道数据在哪个节点开始走偏。

日志会告诉你最终结果，但不会告诉你每一跳的中间状态。想复现一个特定分支，就得修改入口参数，或者在代码里加断点、改 fmt.Println——改完再改回来，麻烦且容易引入错误。

Eino Dev 的思路：进程旁挂一个 HTTP 服务，浏览器/插件直接连上来，任意选一个节点作为起点，填入 mock 数据，按节点顺序看执行结果。

---

二、三步接入

只需在 main.go 里加两行：

import "github.com/cloudwego/eino-ext/devops"

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

    // ① 启动调试服务（默认 127.0.0.1:52538）
    if err := devops.Init(ctx); err != nil {
        log.Fatalf("devops init failed: %v", err)
    }

    // ② 正常注册图（编译时自动被捕获）
    graph.RegisterSimpleGraph(ctx)
    graph.RegisterSimpleStateGraph(ctx)
    chain.RegisterSimpleChain(ctx)

    // ③ 阻塞直到退出
    // ...
}

Init 返回时（约 2 秒超时），HTTP 服务已经在后台运行。之后的所有 g.Compile()、chain.Compile()、wf.Compile() 调用都会被自动拦截，图结构存入内存。

自定义监听地址：

devops.Init(ctx,
    devops.WithDevServerIP("0.0.0.0"),   // 允许外部访问（注意安全）
    devops.WithDevServerPort("9999"),
)

---

三、底层机制：全局编译回调

devops.Init 内部调用的关键一行：

// devops/internal/apihandler/debug.go
compose.InitGraphCompileCallbacks([]compose.GraphCompileCallback{
    service.NewGlobalDevGraphCompileCallback(),
})

这是 GraphCompileCallback——和上一篇 MermaidGenerator 实现的是同一个接口。区别是这里把图信息存进内存，而不是写文件。

每个 g.Compile() 执行后，回调的 OnFinish(ctx, *GraphInfo) 被触发，GraphInfo 里的节点、边、分支、输入输出类型全部记录下来。

---

四、五个核心 API

服务根路径是 /eino/devops，调试接口在 /eino/devops/debug/v1：

方法	路径	作用
GET	/eino/devops/ping	健康检查
GET	/eino/devops/debug/v1/input_types	列出已注册的 Go 类型（供输入表单用）
GET	/eino/devops/debug/v1/graphs	列出所有已编译的图
GET	/eino/devops/debug/v1/graphs/{id}/canvas	获取图的拓扑结构（节点、边、子图）
POST	/eino/devops/debug/v1/graphs/{id}/threads	创建调试会话（thread）
POST	/eino/devops/debug/v1/graphs/{id}/threads/{tid}/stream	从指定节点开始调试执行（SSE 流式返回）

浏览器插件或 VS Code 扩展连接到 127.0.0.1:52538 后，先调 /graphs 列出图列表，再通过 /canvas 拿到拓扑渲染成可点击的节点图，最后在界面上填输入、点执行。

---

五、从任意节点开始

调试时不需要每次都从 START 开始。选中图中任意一个节点作为 from_node，StreamDebugRun 接口的请求体格式：

{
  "from_node": "node_2",
  "input": { "content": "hello" },
  "log_id": "debug-001"
}

内部调用 BuildDevGraph(gi, fromNode)，动态重建一张从 node_2 开始、到 END 结束的子图，再用 mock 输入跑它。

// devops/internal/model/container.go（节选）
func BuildDevGraph(gi *GraphInfo, fromNode string) (g *Graph, err error) {
    // BFS 从 fromNode 开始，只把它后面的节点和边加进来
    queue := []string{fromNode}
    // ...
    // 把 fromNode 接在 START 上，让子图能独立跑
    if fromNode != compose.START {
        g.AddEdge(compose.START, fromNode)
    }
    return g, nil
}

这样你能隔离出问题节点，把上游节点的预期输出直接 mock 进来，不用跑完整链路。

---

六、NodeDebugState：每节点的完整快照

执行时每个节点完成后推送一条 SSE 事件，payload 就是 NodeDebugState：

type NodeDebugState struct {
    NodeKey   string  // 节点 key（对应图中的 "node_2"）
    Input     string  // 节点实际接收到的输入（JSON 字符串）
    Output    string  // 节点产生的输出（JSON 字符串）
    Error     string  // 如果有错误，错误信息明文
    ErrorType ErrorType  // NodeError / SystemError

    Metrics NodeDebugMetrics
}

type NodeDebugMetrics struct {
    PromptTokens     int64  // 如果节点是 ChatModel，输入 token 数
    CompletionTokens int64  // 输出 token 数
    InvokeTimeMS     int64  // 节点总耗时（毫秒）
    CompletionTimeMS int64  // 流式完成耗时（毫秒）
}

前端把每个节点的 Input/Output 展开显示，能直接看到「node_1 输出了什么，node_2 收到了什么」——数据在哪一跳变形，一眼就能找到。

---

七、any 类型输入的特殊格式

Go 的 any（interface{}）在 JSON 里无法携带类型信息，反序列化时默认变成 map[string]interface{}。调试时如果节点接收的是 map[string]any，前端表单无法知道每个值的具体类型。

解决方案：用带类型标注的 JSON 格式：

{
  "name": {
    "_value": "alice",
    "_eino_go_type": "string"
  },
  "score": {
    "_value": "99",
    "_eino_go_type": "int"
  }
}

_eino_go_type 告诉运行时应该把 _value 反序列化成哪种 Go 类型。框架自动处理，不需要修改节点代码。

---

八、AppendType 注册自定义类型

默认只注册了框架内置类型（*schema.Message、map[string]interface{} 等）。如果你的节点输入是自定义结构体，需要注册：

type MyRequest struct {
    Query  string `json:"query"`
    Limit  int    `json:"limit"`
}

devops.Init(ctx,
    devops.AppendType(&MyRequest{}),
)

注册后，前端 /input_types 会列出 *MyRequest，调试表单能自动生成对应的字段输入框。

---

九、StateGraph 的调试

带状态图（compose.WithGenLocalState）的每个节点可以有 StatePreHandler 和 StatePostHandler。调试时这些 handler 正常执行，state 的变化会体现在 NodeDebugState.Output 里，可以看到每个节点执行后 state 的 snapshot。

sg.AddLambdaNode("node_1", compose.InvokableLambda(fn),
    compose.WithStatePreHandler(func(ctx context.Context, input string, state *nodeState) (string, error) {
        state.Messages = append(state.Messages, input)
        return input, nil
    }),
)

---

小结

Eino Dev 的设计思路是零侵入：业务代码不动，devops.Init 往全局编译回调里塞一个 listener，图结构自动捕获。

调试服务提供五个 HTTP 端点，核心是 /stream — 给定起始节点 + mock 输入，内部用 BuildDevGraph 裁剪出子图，按正常流程跑，每个节点完成后 SSE 推一条 NodeDebugState，输入、输出、Token 数、耗时一目了然。

any 类型用 _eino_go_type 标注，自定义类型用 AppendType 注册，两个机制覆盖了大部分实际项目的输入类型需求。

下篇继续。

---

代码来源：cloudwego/eino-ext · cloudwego/eino-examples