1. 从Eino到Mermaid为什么我们需要这个转换如果你最近在折腾AI应用开发尤其是基于LangChain、LangGraph这类框架构建智能体Agent或者复杂的工作流那你大概率见过或者用过Eino。它不是一个独立的工具而是LangGraph官方提供的一个可视化调试工具能把你用代码定义的复杂流程变成一个可以交互的图形界面。这玩意儿在调试一个多步骤、带循环、有状态的工作流时简直是救命稻草——你可以清晰地看到数据流向了哪个节点当前状态是什么哪里卡住了。但Eino有个不大不小的痛点它的图是“一次性”的。你只能在开发环境里跑起来看截图质量一般也没法方便地嵌入到你的技术文档、设计稿或者项目README里。想象一下你花了三天三夜调通了一个精妙的Agent协作流程想写篇博客记录一下或者给团队新人做培训结果发现最核心的流程图你只能贴一张模糊的截图或者费力地用文字描述“先这样再那样”。这体验太差了。这时候Mermaid的优势就体现出来了。Mermaid是一种基于文本的图表生成语言你用简单的代码就能描述流程图、时序图、类图等等。它最大的好处是“文本即图表”可以轻松地用Markdown嵌入被Git版本管理并且能通过各种渲染引擎比如GitHub、GitLab、Notion、Typora生成清晰、一致的矢量图。如果把Eino的编排图转换成Mermaid代码就意味着我们能把动态的、调试用的视图固化为静态的、可文档化的资产。所以这个转换的核心价值就出来了打通从动态调试到静态文档的壁垒。它不是为了替代Eino而是作为Eino的完美补充。让你在享受Eino交互式调试便利的同时也能轻松产出专业级的技术文档图表。接下来我会带你深入Eino的内部机制看看我们如何能从中“挖”出足够的信息来构建一份Mermaid能理解的蓝图。2. 解构Eino它的图数据从何而来要完成转换我们首先得成为“拆弹专家”搞清楚Eino这幅可视化地图的“测绘数据”源头在哪里。你不能指望Eino直接给你一个“导出为Mermaid”的按钮至少目前没有我们需要自己动手丰衣足食。Eino可视化的是LangGraph的StateGraph对象。当你运行一个LangGraph应用并启用Eino时框架会在内存中构建一个图结构这个结构包含了我们定义的所有关键信息。我们的目标就是拦截或读取这个结构。根据LangGraph的公开API和常见用法我梳理出几个可行的数据抓取入口并分析它们的优劣。2.1 方案一直接解析Graph对象最直接但需深入框架这是最理想的方案直接从源头获取数据。一个构建好的StateGraph对象其内部包含了nodes和edges的定义。from langgraph.graph import StateGraph # 假设这是你定义的图 workflow StateGraph(YourStateClass) # 添加节点和边... # workflow.add_node(...) # workflow.add_conditional_edges(...) # 编译图 compiled_graph workflow.compile()理论上compiled_graph对象或者workflow对象内部应该有一个完整的图表示。我们可以尝试通过Python的反射inspection来获取这些信息。例如workflow.__dict__或compiled_graph.__dict__可能会暴露一些内部属性。我在实际项目中尝试过发现不同版本下内部属性名可能变化但通常能找到类似_nodes、_edges这样的结构。优点数据最准确、最完整能拿到节点名、边条件等原始信息。缺点依赖LangGraph内部未公开的API稳定性差。版本升级可能导致代码失效。属于“黑魔法”不推荐在生产文档流程中强依赖。2.2 方案二拦截Eino的WebSocket数据逆向工程思路Eino作为一个本地Web服务它的前端图表是通过WebSocket从后端你的LangGraph应用实时获取数据来渲染的。如果我们能监听这个WebSocket通信就能拿到最接近Eino视图的原始数据。你可以使用浏览器开发者工具F12切换到Network网络标签页过滤WSWebSocket连接当Eino界面刷新时观察收发的数据报文。这些报文通常是JSON格式里面很可能包含了节点、边、以及当前执行状态的信息。优点数据格式与Eino视图完全对应包含了执行时的动态信息如高亮当前节点。缺点技术门槛较高需要解析可能复杂的JSON结构数据可能夹杂了大量用于前端渲染的冗余信息需要清洗同样存在因Eino升级而格式变动的风险。2.3 方案三自定义图编译过程推荐方案这是我认为在平衡可靠性、可控性和复杂度之后的最佳实践。思路是在定义图的过程中我们自己额外维护一份图的元数据。LangGraph在添加节点和边时我们完全知晓这些信息。我们可以在一个自定义的“图构建器”里把这些信息同时记录到一个独立的数据结构中。这个数据结构就是我们生成Mermaid代码的原料。from typing import Dict, List, Any from langgraph.graph import StateGraph, END class MermaidGraphRecorder: def __init__(self): self.nodes: List[str] [] self.edges: List[Dict[str, Any]] [] # 例如{from: node_a, to: node_b, label: condition} def add_node(self, name: str): if name not in self.nodes: self.nodes.append(name) def add_edge(self, from_node: str, to_node: str, condition: str None): self.edges.append({ from: from_node, to: to_node, label: condition }) # 在你的图构建代码中使用 recorder MermaidGraphRecorder() workflow StateGraph(YourStateClass) def node_a(state): # ... 业务逻辑 return state workflow.add_node(NodeA, node_a) recorder.add_node(NodeA) def node_b(state): # ... 业务逻辑 return state workflow.add_node(NodeB, node_b) recorder.add_node(NodeB) # 添加边 workflow.add_edge(NodeA, NodeB) recorder.add_edge(NodeA, NodeB) # 条件边 def router(state): if state[value] 10: return NodeC else: return NodeD workflow.add_conditional_edges( NodeB, router, {NodeC: NodeC, NodeD: NodeD} ) # 这里需要根据router的返回值记录多条边 recorder.add_edge(NodeB, NodeC, value 10) recorder.add_edge(NodeB, NodeD, value 10)这个方案将转换逻辑与图定义逻辑紧密耦合虽然增加了少量样板代码但换来了绝对的可靠性和灵活性。我们完全掌控了需要记录什么信息以及信息的格式。接下来我们就可以基于recorder对象中存储的nodes和edges来生成Mermaid代码了。3. 构建Mermaid生成器从数据到图表代码拿到了结构化的图数据节点列表和边列表下一步就是将它们翻译成Mermaid语法。Mermaid的流程图flowchart语法非常直观我们主要利用其flowchart TD自上而下或flowchart LR从左到右的布局。3.1 基础转换节点与普通边首先我们需要将每个节点定义为一个Mermaid节点。Mermaid支持多种节点形状对于工作流矩形是最常用的。def generate_mermaid_code(recorder: MermaidGraphRecorder) - str: lines [] lines.append(mermaid) lines.append(flowchart TD) # 采用自上而下的布局更符合工作流阅读习惯 # 1. 定义所有节点 for node in recorder.nodes: # 使用方括号定义矩形节点节点ID和显示文本暂时相同 # 如果节点名包含特殊字符或空格需要处理 node_id node.replace( , _).replace(-, _) lines.append(f {node_id}[{node}]) # 2. 定义所有边 for edge in recorder.edges: from_id edge[from].replace( , _).replace(-, _) to_id edge[to].replace( , _).replace(-, _) label edge.get(label) if label: # 带标签的边 lines.append(f {from_id} --|{label}| {to_id}) else: # 普通边 lines.append(f {from_id} -- {to_id}) lines.append() return \n.join(lines)这样就能生成一个最基本的流程图。但Eino图中的一些高级特性我们还需要进一步处理。3.2 处理特殊节点开始与结束在LangGraph中有一个特殊的END节点。在Mermaid中我们通常用(( ))表示圆形开始/结束或者用[ ]加上特殊样式。为了更贴近Eino的视觉习惯我们可以将名为END的节点渲染为不同的形状。# 在生成节点定义的循环中修改 for node in recorder.nodes: node_id node.replace( , _).replace(-, _) if node END: # 使用双圆环表示结束节点 lines.append(f {node_id}(({node}))) elif node __start__: # 假设我们有一个开始节点LangGraph内部可能有 lines.append(f {node_id}[{node}]:::start) else: lines.append(f {node_id}[{node}])注意LangGraph不一定有显式的__start__节点。图的入口是你调用compiled_graph.invoke()时指定的第一个节点。在Mermaid中我们可以通过样式或注释来标记入口节点或者不特殊处理让读者从没有入边的节点识别起点。3.3 处理条件边与分支这是转换中最有趣也最具挑战的部分。Eino图中条件边add_conditional_edges通常会从一个节点引出多个分支每个分支上有一个条件标签。上面的基础代码已经通过--|label|支持了带标签的边。但这里有个细节Mermaid的边标签默认放在线条旁边。对于从同一个节点出发的多个条件边Mermaid会自动处理布局但有时线条会交叉影响可读性。我们可以利用Mermaid的subgraph子图功能来对条件分支进行视觉上的分组但这会增加代码复杂度。对于大多数情况清晰的标签已经足够。一个更重要的点是条件路由函数router的映射。在add_conditional_edges中我们传入一个字典来映射router函数的返回值到下一个节点名。在我们的MermaidGraphRecorder.add_edge调用中我们必须模拟这个逻辑为每一种可能的返回值创建一条边。这要求我们在记录边时必须知晓路由函数的所有可能输出。这通常意味着我们需要在业务代码层面“硬编码”这些可能性或者通过解析路由函数的源码/注解来获取这又回到了方案一的复杂领域。因此在自定义记录器方案中手动维护边的映射是最务实的选择。3.4 样式美化与自定义Mermaid支持CSS类定义可以让图表更美观。我们可以在图表定义前加入样式定义。def generate_mermaid_code_with_style(recorder: MermaidGraphRecorder) - str: lines [] lines.append(mermaid) lines.append(flowchart TD) # 定义样式类 lines.append( classDef default fill:#f9f,stroke:#333,stroke-width:2px,color:#000;) lines.append( classDef start fill:#6f6,stroke:#333,stroke-width:3px;) lines.append( classDef end fill:#f66,stroke:#333,stroke-width:3px;) lines.append( classDef condition fill:#9cf,stroke:#333,stroke-width:2px;) # ... 添加节点和边 ... # 应用样式类 lines.append( class __start__ start;) lines.append( class END end;) # 可以为特定节点应用condition样式 # lines.append(f class {some_node_id} condition;) lines.append() return \n.join(lines)这样开始和结束节点就有了醒目的颜色。你可以根据你的文档主题自定义这些颜色和样式。4. 实战集成打造自动化文档工作流理论讲完了我们来点实际的。如何将这套转换机制无缝集成到你的AI应用项目中实现“代码即文档”我分享一个我正在用的、基于FastAPI和LangGraph的项目结构。4.1 项目结构设计your_agent_project/ ├── app/ │ ├── graphs/ │ │ ├── __init__.py │ │ ├── base.py # 包含MermaidGraphRecorder和基础图类 │ │ └── customer_support.py # 具体的业务工作流图 │ ├── api/ │ │ └── endpoints.py # FastAPI 路由 │ └── main.py ├── docs/ │ └── workflows/ │ └── customer_support.md # 自动生成的文档 ├── scripts/ │ └── generate_workflow_docs.py # 文档生成脚本 └── requirements.txt4.2 实现可复用的基础图类在app/graphs/base.py中我们创建增强版的图构建器。# app/graphs/base.py from typing import Dict, List, Any, Optional from langgraph.graph import StateGraph class MermaidGraphRecorder: # ... 同上文定义 ... class DocumentedStateGraph(StateGraph): 继承自StateGraph自动记录图结构以生成Mermaid。 def __init__(self, state_schema, name: str Unnamed Workflow): super().__init__(state_schema) self.recorder MermaidGraphRecorder() self.graph_name name def add_node(self, name: str, node): super().add_node(name, node) self.recorder.add_node(name) return self def add_edge(self, start_key: str, end_key: str): super().add_edge(start_key, end_key) self.recorder.add_edge(start_key, end_key) return self def add_conditional_edges( self, start_key: str, condition, path_map: Dict[str, str], path_key: Optional[str] None, ): super().add_conditional_edges(start_key, condition, path_map, path_key) # 记录所有可能的分支边 for condition_value, next_node in path_map.items(): self.recorder.add_edge(start_key, next_node, str(condition_value)) return self def get_mermaid_code(self) - str: 返回当前图的Mermaid代码。 return generate_mermaid_code_with_style(self.recorder) def export_mermaid_to_file(self, filepath: str): 将Mermaid代码导出到文件。 code self.get_mermaid_code() with open(filepath, w, encodingutf-8) as f: f.write(code) print(fMermaid diagram exported to: {filepath})4.3 定义业务工作流在app/graphs/customer_support.py中使用我们的新类来定义图。# app/graphs/customer_support.py from app.graphs.base import DocumentedStateGraph from typing import TypedDict, Annotated from langgraph.graph import END import operator class SupportState(TypedDict): user_query: str classification: str knowledge_answer: Optional[str] llm_answer: Optional[str] final_answer: str def classify_query(state: SupportState): # 模拟一个分类器 query state[user_query].lower() if 退款 in query or 退货 in query: state[classification] policy elif 怎么用 in query or 步骤 in query: state[classification] tutorial else: state[classification] general return state def retrieve_policy(state: SupportState): # 模拟检索知识库 state[knowledge_answer] 根据政策第7条购买后30天内可无理由退货。 return state def generate_with_llm(state: SupportState): # 模拟LLM生成 state[llm_answer] f关于{state[user_query]}这是一个通用问题建议联系人工客服。 return state def format_final_answer(state: SupportState): if state[classification] policy: state[final_answer] state[knowledge_answer] else: state[final_answer] state[llm_answer] return state def route_by_classification(state: SupportState): # 路由函数 classification state[classification] if classification policy: return retrieve_policy elif classification tutorial: return generate_tutorial # 假设有另一个节点 else: return generate_with_llm # 构建图 def create_support_workflow(): workflow DocumentedStateGraph(SupportState, nameCustomer Support Workflow) workflow.add_node(classify_query, classify_query) workflow.add_node(retrieve_policy, retrieve_policy) workflow.add_node(generate_with_llm, generate_with_llm) workflow.add_node(format_final_answer, format_final_answer) workflow.set_entry_point(classify_query) workflow.add_conditional_edges( classify_query, route_by_classification, { policy: retrieve_policy, general: generate_with_llm, tutorial: generate_with_llm # 简化处理都导向LLM } ) workflow.add_edge(retrieve_policy, format_final_answer) workflow.add_edge(generate_with_llm, format_final_answer) workflow.add_edge(format_final_answer, END) return workflow.compile(), workflow # 获取图和记录器 compiled_graph, documented_graph create_support_workflow()4.4 自动化文档生成脚本创建一个脚本scripts/generate_workflow_docs.py在每次代码更新后运行自动更新文档。# scripts/generate_workflow_docs.py import sys import os sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ..))) from app.graphs.customer_support import documented_graph def generate_markdown_doc(graph, output_path: str): mermaid_code graph.get_mermaid_code() markdown_content f# 客户支持工作流图 本文档由自动化脚本生成对应 app/graphs/customer_support.py 中定义的工作流。 ## 流程图 {mermaid_code} ## 节点说明 - **classify_query**: 对用户查询进行分类。 - **retrieve_policy**: 针对政策类问题从知识库检索答案。 - **generate_with_llm**: 针对一般或教程类问题使用大语言模型生成答案。 - **format_final_answer**: 整合答案并格式化输出。 - **END**: 工作流结束节点。 ## 边与条件 - classify_query - retrieve_policy: 当分类结果为 policy。 - classify_query - generate_with_llm: 当分类结果为 general 或 tutorial。 - retrieve_policy - format_final_answer: 无条件。 - generate_with_llm - format_final_answer: 无条件。 - format_final_answer - END: 无条件。 --- *最后更新于: {datetime.datetime.now().strftime(%Y-%m-%d %H:%M:%S)}* os.makedirs(os.path.dirname(output_path), exist_okTrue) with open(output_path, w, encodingutf-8) as f: f.write(markdown_content) print(fDocumentation generated at: {output_path}) if __name__ __main__: output_path ./docs/workflows/customer_support.md generate_markdown_doc(documented_graph, output_path)将这个脚本加入你的CI/CD流程例如GitHub Actions的on: push或者配置一个pre-commit钩子就能确保你的项目文档永远与代码定义的工作流同步。5. 高级技巧与避坑指南在实际操作中你可能会遇到一些预料之外的情况。这里分享几个我踩过坑后总结的经验。5.1 处理复杂节点名与ID冲突Mermaid的节点ID必须是一个简单的字符串不能包含空格、连字符在某些上下文中会引发解析问题。我们的转换函数里用了简单的替换replace( , _)但这可能不够。问题如果节点名本身带有下划线或者经过替换后产生重复ID例如“My-Node”和“My_Node”都会变成“My_Node”就会导致Mermaid图表错误。解决方案使用更稳健的ID生成方法比如基于节点名生成一个唯一哈希或者维护一个ID到显示名的映射字典。import hashlib def safe_node_id(node_name: str) - str: 生成一个安全的、唯一的节点ID。 # 或者使用 uuid但为了可读性这里用前缀哈希 prefix node_ hash_obj hashlib.md5(node_name.encode()) short_hash hash_obj.hexdigest()[:8] return f{prefix}{short_hash} # 在记录节点时同时保存映射关系 self.node_id_map[node_name] safe_node_id(node_name) # 生成Mermaid时用 self.node_id_map[node] 作为ID用 node 作为显示标签。5.2 可视化“状态”与“数据流”Eino的一个强大之处是能展示状态State的变化。我们的基本转换只处理了“控制流”节点和边。如果你也想在Mermaid中暗示数据流可以通过更丰富的标签来实现。例如在边的标签上不仅写明条件还可以注明该步骤主要更新了哪个状态字段。# 在add_edge时可以记录更多信息 def add_data_edge(self, from_node: str, to_node: str, condition: str None, updates: List[str] None): self.edges.append({ from: from_node, to: to_node, label: condition, updates: updates or [] }) # 生成边时 label_parts [] if edge.get(label): label_parts.append(edge[label]) if edge.get(updates): label_parts.append(f更新: {, .join(edge[updates])}) edge_label br/.join(label_parts) # 用HTML换行符Mermaid支持基础HTML if edge_label: lines.append(f {from_id} --|\{edge_label}\| {to_id})这样边上的标签就会显示为“条件更新: 字段A, 字段B”更接近Eino的调试视图。5.3 集成到FastAPI并在线预览你甚至可以在你的FastAPI应用中增加一个端点实时查看工作流的Mermaid图。# app/api/endpoints.py from fastapi import APIRouter from fastapi.responses import HTMLResponse from app.graphs.customer_support import documented_graph router APIRouter() router.get(/workflow/diagram, response_classHTMLResponse) async def get_workflow_diagram(): mermaid_code documented_graph.get_mermaid_code() # 移除代码块的标记因为我们要嵌入到HTML中 mermaid_code_clean mermaid_code.replace(mermaid\n, ).replace(\n, ) html_content f !DOCTYPE html html head script srchttps://cdn.jsdelivr.net/npm/mermaid10/dist/mermaid.min.js/script scriptmermaid.initialize({{startOnLoad:true}});/script style body {{ font-family: sans-serif; margin: 20px; }} .mermaid {{ border: 1px solid #ccc; padding: 20px; background: white; }} /style /head body h1Customer Support Workflow/h1 div classmermaid {mermaid_code_clean} /div psmallDiagram generated from live graph definition./small/p /body /html return HTMLResponse(contenthtml_content)访问/workflow/diagram你就能看到一个实时渲染的、可交互Mermaid支持点击的工作流图。这对于团队协作和演示非常有用。5.4 最大的坑动态图与静态图的差异请务必记住我们生成的是静态的结构图而Eino显示的是动态的执行跟踪图。这是本质区别。静态图展示了所有可能的节点和路径。就像一张地图标出了所有的道路和地点。动态图Eino展示了一次特定执行所走过的路径。就像地图上高亮显示了你这次旅行的具体路线。我们的转换工具生成的是“地图”。它无法复现某一次具体运行的“路线高亮”。如果你需要记录某次特定执行的路径用于事后分析你需要记录LangGraph的调用日志然后根据日志来生成一个高亮了特定路径的Mermaid图例如通过给经过的节点和边添加不同的CSS类。这涉及到更复杂的状态追踪但原理是相通的在invoke过程中记录下被访问的节点序列然后在生成Mermaid代码时为这些节点和边附加特殊的样式。实现这个功能会大大增加复杂性但对于调试复杂的、非确定性的工作流比如包含LLM调用的非常有价值。你可以考虑将其作为一个进阶功能在基础的静态图生成稳定后再进行迭代。