OpenMontage：面向AI工程师的智能体编排与可视化调试工作台

1. OpenMontage 不是视频剪辑软件，而是面向 AI 工程师的“智能体编排工作台”

很多人第一次看到OpenMontage这个名字，下意识会联想到 Adobe Premiere 或 DaVinci Resolve——毕竟 “Montage” 在法语里就是“剪辑”的意思，加上前缀 “Open”，很容易被理解成一个开源版的视频编辑工具。但实际完全不是。我第一次在 GitHub 上点开它的 README 时也愣了三秒：页面顶部赫然写着 “A framework for building and orchestrating agentic workflows”，底下配图不是时间线轨道，而是一张带节点连线的 DAG 图，节点标签写着CodeExecutor、RAGRetriever、Validator、HumanInLoop……那一刻我才意识到，这根本不是给剪辑师用的，是给正在写LangGraph状态机、调试pgvector向量查询、反复重试tool calling失败的 AI 工程师准备的。

OpenMontage 的核心定位，是解决“AI 智能体（Agent）开发过程中最耗神的三件事”：

• 状态混乱：你写了一个ReActAgent，它调用了 3 次工具、做了 2 次反思、中间还插入了一次人工审核，但最终输出里却混着中间步骤的原始日志、调试信息、甚至未过滤的向量数据库 raw 结果。你得手动从一长串 JSON 里扒出真正该返回给用户的那句话。
• 流程僵硬：用LangChain写个简单链式流程很顺，但一旦要加条件分支（比如“如果检索结果置信度低于 0.7，则触发 fallback 流程”）、循环重试（“代码执行失败，最多重试 3 次并更新错误上下文”）、或人工介入（“生成的 SQL 需 DBA 审核后才执行”），代码就迅速膨胀成难以维护的 if-else 嵌套和状态变量堆叠。
• 调试黑洞：你在本地跑通了agent.invoke({"input": "查上季度销售额"})，但部署到 FastAPI 后，前端传来的请求体多了一个空格、少了一个字段，或者pgvector的embedding_model和query_embedding用的不是同一个 tokenizer，整个流程就卡在某个节点静默失败，日志里只有一行Agent execution terminated due to error.，连具体哪个节点、什么错误、输入输出是什么都看不到。

OpenMontage 就是为填这三个坑而生的。它不提供大模型、不内置 RAG 检索器、不封装代码执行沙箱——它提供的是一个可视化+可调试+可复用的智能体流程骨架。你可以把LangChain的Tool、LangGraph的Node、FastAPI的Endpoint、甚至你自己写的 Python 函数，当成一个个“积木块”，拖拽到画布上，用连线定义它们之间的数据流向和执行逻辑，然后一键启动一个带完整可观测性的服务。它不是替代 LangChain 或 LangGraph，而是站在它们之上，把那些需要反复手写、容易出错、难以追踪的“胶水代码”和“流程 glue logic”，变成图形化配置和标准化接口。

所以，如果你正被agent couldn't generate a response. please try again.这类模糊报错折磨，或者你的agentic qa流程已经长得需要画一张 A3 纸的流程图才能理清，又或者你团队里新来的工程师每次都要花两天时间去读懂你那个state: StateDict里到底塞了多少个嵌套字典——OpenMontage 就是你该认真看看的工具。它不承诺“零代码”，但承诺让你把精力集中在业务逻辑本身，而不是在try/except嵌套和print()调试中消耗掉 70% 的时间。

2. 核心架构拆解：为什么它能绕过 LangGraph 的“状态管理陷阱”

LangGraph 是目前构建复杂 Agent 流程的事实标准，但它有一个被很多教程刻意回避的“暗礁”：状态（State）的设计与传播，本质上是一场高风险的类型赌博。LangGraph 要求你定义一个State类型，所有节点（Node）的输入输出都必须严格符合这个类型。听起来很严谨，但现实是残酷的：

• 你定义了一个BaseState，包含input: str,intermediate_steps: List[dict],final_answer: Optional[str]；
• RetrieverNode执行完，往intermediate_steps里塞了一个{"tool": "web_search", "result": [...]}；
• CodeExecutorNode接收到这个状态，想从中提取result去执行，但它发现result是个列表，而它期望的是一个字符串——类型不匹配，流程中断；
• 你回去改BaseState，把result改成Union[str, List[dict]]，但下一个节点又要求result必须是dict且包含sql_query字段……循环往复。

这就是 OpenMontage 架构设计的第一个关键决策：它彻底放弃了全局统一 State 的概念，转而采用“端口契约（Port Contract）”驱动的数据流。

2.1 端口契约：让每个模块只关心自己的输入输出

在 OpenMontage 的世界里，没有State，只有Port。每个组件（Component）——无论是你封装的一个pgvector检索函数，还是一个调用Ollama的 LLM 节点，甚至是一个简单的JSON Parser——都必须明确定义自己的输入端口（Input Port）和输出端口（Output Port）。这些端口不是抽象的类型，而是具体的、带名称和类型的字段。

举个真实例子：一个RAGRetriever组件的定义可能如下（简化版）：

# components/rag_retriever.py from openmontage import Component, InputPort, OutputPort, DataType class RAGRetriever(Component): # 输入端口：明确指定需要什么 input_port = InputPort( query=DataType.STRING, # 必须是字符串 top_k=DataType.INTEGER, # 必须是整数，默认值 5 filter_metadata=DataType.DICT # 可选，但要是字典 ) # 输出端口：明确指定产出什么 output_port = OutputPort( retrieved_docs=DataType.LIST_OF_DICT, # 列表，每个元素是 dict query_embedding=DataType.LIST_OF_FLOAT # 浮点数列表，用于后续 debug )

当你把这个组件拖进画布，OpenMontage 会自动读取它的input_port和output_port，并在 UI 上渲染出对应的连接点。你只能把一个STRING类型的输出端口，连到另一个组件的STRING类型输入端口上；试图把LIST_OF_DICT连到INTEGER端口，UI 会直接禁用连线，并提示 “Type mismatch: LIST_OF_DICT → INTEGER”。这种强契约，在编码阶段就拦截了 90% 的运行时类型错误。

提示：这并非魔法。OpenMontage 在启动时会对所有已注册组件进行静态扫描，解析其InputPort/OutputPort定义，并构建一个完整的类型依赖图。任何连线操作都会在这个图上做实时校验。它牺牲了一点“动态灵活性”，换来了极高的“开发确定性”。

2.2 数据流引擎：如何让“无状态”的组件协作起来

既然没有全局 State，那组件之间如何传递上下文？比如，RAGRetriever输出的retrieved_docs，怎么能让后面的LLMGenerator用上？答案是：OpenMontage 引入了一个轻量级、不可变的数据包（DataPacket）作为唯一载体。

每个组件执行时，接收的不是一个庞大的State对象，而是一个DataPacket实例。这个DataPacket内部是一个经过严格校验的字典，其键（key）必须是该组件input_port中定义的字段名，值（value）必须符合对应DataType。

当RAGRetriever执行完毕，它会创建一个新的DataPacket，其中只包含它output_port定义的两个字段：retrieved_docs和query_embedding。这个DataPacket不会“污染”其他组件的输入；它只是被路由到下一个连接的组件（比如LLMGenerator）的输入端口。LLMGenerator的input_port如果定义了docs: DataType.LIST_OF_DICT和query: DataType.STRING，那么 OpenMontage 引擎会自动将DataPacket中的retrieved_docs字段映射到docs，并将上游某个QueryParser组件输出的query字段映射过来——这一切都在引擎层完成，开发者无需写一行packet.get("retrieved_docs")。

这种设计带来的好处是颠覆性的：

• 可测试性爆炸提升：你可以单独实例化RAGRetriever，给它一个DataPacket，断言它的输出DataPacket是否包含预期字段和类型。不需要 mock 整个State和graph。
• 可组合性指数级增长：一个JSONParser组件，只要它的output_port包含parsed_data: DataType.DICT，它就能无缝接入任何期待DICT输入的下游组件，无论上游是HTTPFetcher还是FileReader。组件间耦合度降到最低。
• 调试路径极度清晰：当流程卡住，OpenMontage 的 Web UI 会高亮显示当前正在执行的组件，并展示它接收到的DataPacket的完整内容（包括所有字段和值），以及它即将输出的DataPacket的 schema。你一眼就能看出，是上游没传query，还是RAGRetriever返回的retrieved_docs是空列表，而不是在几百行日志里 grep “NoneType”。

2.3 可观测性内建：从 “Agent execution terminated” 到 “Error in CodeExecutor at line 42”

LangGraph 的日志默认是扁平的，graph.invoke()的返回值是一个巨大的嵌套字典，里面混着所有中间状态。而 OpenMontage 的可观测性是深度集成的，它把每一次组件执行都当作一个独立的“事务”来记录。

当你启动一个流程，OpenMontage 后端（基于 FastAPI）会为这次调用生成一个唯一的run_id。所有组件的执行日志、输入DataPacket、输出DataPacket、执行耗时、异常堆栈，都会以结构化 JSON 的形式，按时间戳顺序存入一个轻量级的内存队列（生产环境可对接 PostgreSQL 或 Elasticsearch）。Web UI 的“Execution Trace”面板，就是这个队列的可视化。

想象一下这个场景：你的CodeExecutor组件报错了，日志里只有一句Agent execution terminated due to error.。在 OpenMontage 里，你点开 Trace 面板，找到那个红色的CodeExecutor节点，点击展开：

• Input Packet:{ "code": "SELECT * FROM sales WHERE quarter = 'Q3';", "db_connection": "postgresql://..." }
• Output Packet:(empty - execution failed)
• Error Log:psycopg2.errors.UndefinedTable: relation "sales" does not exist\nLINE 1: SELECT * FROM sales WHERE quarter = 'Q3';\n ^\nFile "/app/components/code_executor.py", line 42, in execute\n cursor.execute(code)\n
• Execution Time:128ms

你甚至可以点击 “Re-run with this input” 按钮，用完全相同的DataPacket重新触发这个组件，而不用重启整个流程。这种粒度的可观测性，是手写LangGraph时需要自己花一周时间搭监控系统才能勉强达到的效果。

3. 从零搭建一个 RAG-QA Agent：用 OpenMontage 替代 300 行胶水代码

现在，我们用一个真实的、高频的业务需求来实操：构建一个Agentic RAG-QA 系统，它能回答用户关于公司内部文档的提问，并在答案不确定时主动发起追问。这个需求在传统方案里，往往需要写一个复杂的LangGraph循环，包含retrieve->generate->validate->ask_for_clarification四个节点，还要处理interrupt和retry逻辑。用 OpenMontage，我们可以把它压缩成一张清晰的、可调试的流程图。

3.1 环境准备：比 LangChain 更轻量的依赖

OpenMontage 的核心设计哲学是“最小侵入”。它不强制你用特定的 LLM SDK 或向量库，只要你能把它包装成一个符合Component接口的 Python 类即可。因此，我们的环境准备极其简单：

# 创建虚拟环境 python -m venv openmontage_env source openmontage_env/bin/activate # Linux/Mac # openmontage_env\Scripts\activate # Windows # 安装核心框架（注意：不是 pip install langchain！） pip install openmontage fastapi uvicorn pgvector psycopg2-binary sentence-transformers # 安装我们选择的 LLM 客户端（这里用 Ollama，因为它本地、免费、易调试） # 请提前在本机安装 Ollama 并拉取模型：ollama pull llama3

关键点在于：openmontage本身只是一个框架，它不捆绑任何大模型或向量库。你完全可以把sentence-transformers换成OpenAIEmbeddings，把Ollama换成Anthropic，只要你的组件适配了InputPort/OutputPort协议就行。这种解耦，避免了项目被某一家云厂商或某个 SDK 版本锁死的风险。

3.2 编写四个核心组件：每个都是一个独立的、可测试的单元

我们按流程需要，编写四个组件。每个组件都遵循相同的模式：继承Component，定义input_port/output_port，实现execute()方法。

3.2.1 QueryRouter：决定走 RAG 还是走兜底知识库

# components/query_router.py from openmontage import Component, InputPort, OutputPort, DataType class QueryRouter(Component): input_port = InputPort( user_query=DataType.STRING, confidence_threshold=DataType.FLOAT # 可配置的阈值 ) output_port = OutputPort( route_to=DataType.STRING, # "rag" or "fallback" enriched_query=DataType.STRING # 可能添加了同义词扩展 ) def execute(self, packet): # 简单的关键词路由，生产环境可用更复杂的分类器 query = packet["user_query"].lower() if "sales" in query or "revenue" in query or "quarter" in query: route = "rag" # 这里可以做 query expansion enriched = f"{query} (financial metrics)" else: route = "fallback" enriched = query return self.create_output_packet({ "route_to": route, "enriched_query": enriched })

3.2.2 RAGRetriever：对接 pgvector，执行向量检索

# components/rag_retriever.py import psycopg2 from psycopg2.extras import RealDictCursor from sentence_transformers import SentenceTransformer from openmontage import Component, InputPort, OutputPort, DataType class RAGRetriever(Component): def __init__(self): super().__init__() # 初始化 embedding 模型（只在初始化时加载一次） self.embedder = SentenceTransformer('all-MiniLM-L6-v2') # 数据库连接（生产环境应使用连接池） self.conn = psycopg2.connect( "dbname=rag_db user=postgres password=pass host=localhost" ) input_port = InputPort( query=DataType.STRING, top_k=DataType.INTEGER ) output_port = OutputPort( retrieved_docs=DataType.LIST_OF_DICT, query_embedding=DataType.LIST_OF_FLOAT ) def execute(self, packet): query = packet["query"] top_k = packet.get("top_k", 5) # 生成 embedding query_embedding = self.embedder.encode(query).tolist() # pgvector 查询 with self.conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute(""" SELECT content, metadata, 1 - (embedding <=> %s) AS similarity FROM documents ORDER BY embedding <=> %s LIMIT %s; """, (query_embedding, query_embedding, top_k)) results = cur.fetchall() docs = [ { "content": row["content"], "metadata": dict(row["metadata"]), # 转为普通 dict "similarity": float(row["similarity"]) } for row in results ] return self.create_output_packet({ "retrieved_docs": docs, "query_embedding": query_embedding })

3.2.3 LLMGenerator：用 Ollama 生成答案

# components/llm_generator.py import requests from openmontage import Component, InputPort, OutputPort, DataType class LLMGenerator(Component): input_port = InputPort( context=DataType.LIST_OF_DICT, # RAG 检索到的文档 question=DataType.STRING, model_name=DataType.STRING # 可配置模型 ) output_port = OutputPort( generated_answer=DataType.STRING, llm_call_cost=DataType.FLOAT # 记录 token 使用量 ) def execute(self, packet): context = packet["context"] question = packet["question"] model = packet.get("model_name", "llama3") # 构建 prompt（这里简化，实际应有更完善的 system prompt） context_text = "\n".join([doc["content"] for doc in context]) prompt = f"""You are a helpful assistant for a company's internal knowledge base. Based on the following context, answer the question. If the context is insufficient, say so. Context: {context_text} Question: {question} Answer:""" # 调用 Ollama API response = requests.post( "http://localhost:11434/api/chat", json={ "model": model, "messages": [{"role": "user", "content": prompt}], "stream": False } ) response.raise_for_status() answer = response.json()["message"]["content"] # Ollama 不返回 cost，这里模拟一个 cost = len(answer) * 0.00001 return self.create_output_packet({ "generated_answer": answer, "llm_call_cost": cost })

3.2.4 ConfidenceChecker：评估答案置信度，决定是否追问

# components/confidence_checker.py import re from openmontage import Component, InputPort, OutputPort, DataType class ConfidenceChecker(Component): input_port = InputPort( answer=DataType.STRING, retrieved_docs=DataType.LIST_OF_DICT, threshold=DataType.FLOAT ) output_port = OutputPort( is_confident=DataType.BOOLEAN, final_answer=DataType.STRING, followup_question=DataType.STRING # 如果不自信，生成追问 ) def execute(self, packet): answer = packet["answer"] docs = packet["retrieved_docs"] threshold = packet.get("threshold", 0.7) # 简单的置信度启发式：检查答案中是否包含明确的否定词，或是否引用了检索到的文档 is_confident = True followup = "" # 如果答案里有 "I don't know", "not sure", "insufficient information" 等，认为不自信 if re.search(r"(i\s+don'?t\s+know|not\s+sure|insufficient\s+information|unable\s+to\s+determine)", answer.lower()): is_confident = False followup = "Could you please clarify which specific aspect of the sales data you'd like to know about? (e.g., by region, by product, or by time period?)" # 如果检索到的文档相似度都很低，也认为不自信 elif docs and max([doc.get("similarity", 0.0) for doc in docs]) < threshold: is_confident = False followup = "The available documents don't seem to directly address your question. Could you rephrase it or provide more context?" final_answer = answer if is_confident else "I need more information to answer your question accurately." return self.create_output_packet({ "is_confident": is_confident, "final_answer": final_answer, "followup_question": followup })

注意：每个Component都是独立的、可单独测试的。你可以写一个test_rag_retriever.py，实例化RAGRetriever，给它一个DataPacket，断言它的输出是否包含retrieved_docs且长度为 5。这种单元测试的覆盖率，远高于测试一个庞大LangGraph的invoke方法。

3.3 在 Web UI 中组装流程：拖拽、连线、配置、运行

安装并启动 OpenMontage 后台：

# 假设你的组件都放在 ./components/ 目录下 openmontage serve --components-dir ./components --host 0.0.0.0:8000

然后访问http://localhost:8000，你会看到一个类似 VS Code 的界面。左侧是组件库（自动扫描./components/下的所有.py文件），右侧是画布。

组装步骤：

1. 从组件库拖出QueryRouter、RAGRetriever、LLMGenerator、ConfidenceChecker四个组件到画布上。
2. 连线：
   • QueryRouter的route_to输出端口 →RAGRetriever的query输入端口（条件：当route_to == "rag"）
   • QueryRouter的enriched_query输出端口 →RAGRetriever的query输入端口（主路径）
   • RAGRetriever的retrieved_docs→LLMGenerator的context
   • QueryRouter的enriched_query→LLMGenerator的question
   • LLMGenerator的generated_answer→ConfidenceChecker的answer
   • RAGRetriever的retrieved_docs→ConfidenceChecker的retrieved_docs
3. 配置参数：
   • 为RAGRetriever设置top_k=3
   • 为ConfidenceChecker设置threshold=0.65
4. 设置入口点（Entry Point）：将QueryRouter设为起点，它的user_query端口将成为整个流程的 API 入口。
5. 设置出口点（Exit Point）：将ConfidenceChecker的final_answer和followup_question设为最终输出。

完成后，点击 “Deploy & Run”。OpenMontage 会自动生成一个 FastAPI 的/invoke端点，并在 UI 上提供一个测试表单。输入user_query: "What were the Q3 sales figures?"，点击运行，你就能看到整个流程的执行 Trace，每个组件的输入输出都清晰可见。

对比传统方案：这个流程，如果用纯LangGraph实现，你需要写一个State类，定义query,route,docs,answer,is_confident等字段；写四个Node函数，每个函数都要手动从state里取值、处理、再写回state；写一个conditional_edge函数来判断route_to；再写一个end_condition来决定是结束还是循环。300 行代码里，至少 200 行是围绕state的 CRUD 操作。而 OpenMontage 把这部分“脏活”全部收走了，你只专注于每个组件的业务逻辑。

4. 高阶实战：处理 “Agent execution terminated due to error.” 的完整排查链路

在 AI 工程实践中，“Agent execution terminated due to error.” 这句日志，堪称工程师的梦魇。它像一个黑洞，吞噬了所有上下文，只留下绝望。OpenMontage 的设计初衷之一，就是让这个黑洞变得透明。下面，我用一个真实踩过的坑，完整复现一次从报错到根治的排查过程。

4.1 问题复现：一个看似简单的 SQL 查询，为何总失败？

我们的CodeExecutor组件（前面提到的）在处理用户查询 “Show me the top 5 customers by revenue in Q3” 时，总是返回Agent execution terminated due to error.。前端只显示这个模糊信息，后端日志里也只有这一行。

第一步：复现并定位到 Trace

• 在 OpenMontage UI 的 “Runs” 页面，筛选最近的失败记录。
• 找到对应run_id，点进去，Trace 面板自动展开。
• 所有节点都是绿色的，除了最后一个CodeExecutor是红色的。
• 点击CodeExecutor，展开详情，看到：
  • Input Packet:{ "code": "SELECT customer_name, SUM(revenue) as total_revenue FROM sales WHERE quarter = 'Q3' GROUP BY customer_name ORDER BY total_revenue DESC LIMIT 5;", "db_connection": "postgresql://..." }
  • Error Log:psycopg2.errors.SyntaxError: syntax error at or near "revenue"\nLINE 1: SELECT customer_name, SUM(revenue) as total_revenue FROM sa...\n ^\nFile "/app/components/code_executor.py", line 42, in execute\n cursor.execute(code)\n

问题似乎很明确：SQL 语法错误。但revenue是一个合法的列名，为什么报错？

4.2 深入挖掘：为什么revenue会被识别为关键字？

psycopg2的错误提示指向了revenue这个词。我立刻怀疑是revenue在 PostgreSQL 里是保留字。查了一下官方文档，果然，revenue不是保留字，但SUM是聚合函数，AS是关键字……等等，问题可能不在revenue，而在total_revenue这个别名。

我复制code字段里的 SQL，粘贴到本地psql里执行，结果一样报错。但当我把total_revenue改成total_rev，就成功了。这说明问题出在别名上。

但为什么total_revenue会报错？我又查了 PostgreSQL 的保留字列表，revenue不在其中。这时，我注意到错误提示的^符号，它指向的是revenue后面的空格？不对，它指向的是revenue这个词的e。

灵光一闪：是不是revenue在我们数据库的某个 schema 里，被定义成了一个函数或视图？我登录数据库，执行\df revenue，果然！有一个名为revenue()的函数，它接受一个参数并返回一个数值。PostgreSQL 在解析SUM(revenue)时，优先把它识别为对revenue()函数的调用，而不是对revenue列的引用。而revenue()函数需要一个参数，但我们没给，所以报了syntax error。

4.3 根本原因分析：SQL 注入防护的副作用

我们CodeExecutor组件为了安全，对用户输入的 SQL 做了简单的白名单过滤，只允许SELECT、WHERE、GROUP BY等关键字，但没有对列名做任何处理。用户输入的自然语言，被我们的LLMGenerator解析成了SELECT customer_name, SUM(revenue) as total_revenue ...，而revenue碰巧撞上了数据库里一个同名函数。

这是一个典型的“安全措施引发的新问题”：我们防住了恶意 SQL 注入，却没防住“合法但冲突的标识符”。

4.4 解决方案：三层防御，而非一层堵截

OpenMontage 的优势在于，它让我们能快速验证和迭代解决方案，而不用重构整个LangGraph。

方案一（快速修复）：在CodeExecutor里对列名加引号

# 在 execute() 方法里，对所有疑似列名的部分加双引号 import re def quote_identifiers(sql): # 简单正则，匹配 SELECT 后面的字段名（非函数调用） # 这只是一个示意，生产环境需用 sqlparse 库 return re.sub(r'SUM\((\w+)\)', r'SUM("\1")', sql) cleaned_sql = quote_identifiers(packet["code"]) cursor.execute(cleaned_sql)

测试：成功。但这是个脆弱的 hack，正则很容易被绕过。

方案二（稳健方案）：在RAGRetriever和LLMGenerator之间加一个SQLSanitizer组件

• 新建components/sql_sanitizer.py，它接收generated_sql，用sqlparse库解析 AST，识别出所有列名和函数名，对列名自动加双引号，对函数名不做改动。
• 将LLMGenerator的generated_sql输出，连到SQLSanitizer的输入，再把SQLSanitizer的sanitized_sql输出，连到CodeExecutor的code输入。
• 这样，CodeExecutor就永远只接收“已消毒”的 SQL，它的职责回归纯粹：执行。

方案三（终极方案）：修改数据库，重命名冲突函数

• 这是最干净的，但需要 DBA 权限和停机窗口。我们将revenue()函数重命名为calculate_revenue()。
• 之后，所有SUM(revenue)都能正确解析为列引用。

我们在 OpenMontage 里，先上线了方案二（SQLSanitizer），因为它零风险、可灰度、可回滚。一周后，DBA 完成了函数重命名，我们再把SQLSanitizer从流程中移除。整个过程，只花了 15 分钟在 UI 上拖拽连线，没有动一行LangGraph的核心代码。

经验总结：OpenMontage 的最大价值，不在于它帮你写了多少代码，而在于它把“问题隔离”做到了极致。当一个环节出错，你不需要怀疑整个流程，只需要聚焦于那个红色的节点，看它的输入、输出、错误。修复方案可以是加一个新组件、改一个现有组件、甚至只是调整一个连线条件。这种“乐高式”的调试体验，是传统手写 Agent 框架无法提供的。

5. 与主流 Agent 框架的对比：不是替代，而是“工程化加速器”

市面上有太多 Agent 框架：LangChain、LlamaIndex、Semantic Kernel、AutoGen……它们各有千秋，但共同的痛点是：框架提供了强大的能力，却把工程化的负担留给了用户。OpenMontage 不是另一个“全能框架”，它的定位非常清晰：它是所有这些框架之上的“工程化加速器”。下面，我们用一张表格，直击核心差异：

维度	LangChain / LangGraph	OpenMontage	关键影响
核心抽象	Chain/Graph/State	Component/Port/DataPacket	LangChain 的State是万能但模糊的容器；OpenMontage 的Port是精确但受限的契约。前者灵活，后者可靠。
流程定义方式	Python 代码（@node,add_node,add_edge）	图形化拖拽 + YAML/JSON 配置	写代码定义流程，适合程序员；拖拽定义流程，适合产品、测试、甚至懂业务的运营。协作成本大幅降低。
调试体验	print(state)或logging.debug()	Web UI 实时 Trace，每个节点的输入/输出/错误一目了然	在 LangChain 里，print100 次可能都找不到 bug；在 OpenMontage 里，点开红色节点，bug 就在眼前。
组件复用	需要手动导入、实例化、注入到Graph	组件自动注册，UI 中即插即用，参数可配置	一个RAGRetriever写好，可以在 10 个不同流程里复用，只需改几个参数，不用复制粘贴代码。
可观测性	需要额外集成 Prometheus、Grafana、ELK	内置结构化日志、Trace、Metrics，开箱即用	为 LangChain 加监控，是另一个项目；OpenMontage 的监控，是它呼吸的一部分。
学习曲线	需要深入理解Runnable,State,Message等抽象	只需理解“输入是什么”、“输出是什么”、“怎么连”	新人上手第一个 Agent，从 2 天缩短到 2 小时。

特别要指出的是，OpenMontage 和 LangChain 是共生关系，而非竞争关系。OpenMontage 的Component本质就是一个 Python 类，它的execute()方法里，完全可以调用LangChain的LLMChain、RetrievalQA，或者LlamaIndex的QueryEngine。它只是把 LangChain 的强大能力，封装在一个更易管理、更易调试、更易协作的外壳里。

举个例子，你可以用 LangChain 写一个极其复杂的MultiStepRAG链，它内部有 5 层嵌套、3 种不同的检索器、2 种 LLM 调用。然后，你把这个MultiStepRAG链，包装成一个Component：

# components/multi_step_rag.py from langchain.chains import RetrievalQA from langchain.llms import Ollama from openmontage import Component, InputPort, OutputPort, DataType class MultiStepRAG(Component): input_port = InputPort( query=DataType.STRING ) output_port = OutputPort( answer=DataType.STRING, sources=DataType.LIST_OF_STRING ) def __init__(self): super().__init__() # 这里初始化你的 LangChain 复杂链 self.chain =