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 couldnt 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( queryDataType.STRING, # 必须是字符串 top_kDataType.INTEGER, # 必须是整数默认值 5 filter_metadataDataType.DICT # 可选但要是字典 ) # 输出端口明确指定产出什么 output_port OutputPort( retrieved_docsDataType.LIST_OF_DICT, # 列表每个元素是 dict query_embeddingDataType.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)\nExecution 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_queryDataType.STRING, confidence_thresholdDataType.FLOAT # 可配置的阈值 ) output_port OutputPort( route_toDataType.STRING, # rag or fallback enriched_queryDataType.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( dbnamerag_db userpostgres passwordpass hostlocalhost ) input_port InputPort( queryDataType.STRING, top_kDataType.INTEGER ) output_port OutputPort( retrieved_docsDataType.LIST_OF_DICT, query_embeddingDataType.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_factoryRealDictCursor) 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( contextDataType.LIST_OF_DICT, # RAG 检索到的文档 questionDataType.STRING, model_nameDataType.STRING # 可配置模型 ) output_port OutputPort( generated_answerDataType.STRING, llm_call_costDataType.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 fYou are a helpful assistant for a companys 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( answerDataType.STRING, retrieved_docsDataType.LIST_OF_DICT, thresholdDataType.FLOAT ) output_port OutputPort( is_confidentDataType.BOOLEAN, final_answerDataType.STRING, followup_questionDataType.STRING # 如果不自信生成追问 ) def execute(self, packet): answer packet[answer] docs packet[retrieved_docs] threshold packet.get(threshold, 0.7) # 简单的置信度启发式检查答案中是否包含明确的否定词或是否引用了检索到的文档 is_confident True followup # 如果答案里有 I dont know, not sure, insufficient information 等认为不自信 if re.search(r(i\sdon?t\sknow|not\ssure|insufficient\sinformation|unable\sto\sdetermine), answer.lower()): is_confident False followup Could you please clarify which specific aspect of the sales data youd 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 dont 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文件右侧是画布。组装步骤从组件库拖出QueryRouter、RAGRetriever、LLMGenerator、ConfidenceChecker四个组件到画布上。连线QueryRouter的route_to输出端口 →RAGRetriever的query输入端口条件当route_to ragQueryRouter的enriched_query输出端口 →RAGRetriever的query输入端口主路径RAGRetriever的retrieved_docs→LLMGenerator的contextQueryRouter的enriched_query→LLMGenerator的questionLLMGenerator的generated_answer→ConfidenceChecker的answerRAGRetriever的retrieved_docs→ConfidenceChecker的retrieved_docs配置参数为RAGRetriever设置top_k3为ConfidenceChecker设置threshold0.65设置入口点Entry Point将QueryRouter设为起点它的user_query端口将成为整个流程的 API 入口。设置出口点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(rSUM\((\w)\), rSUM(\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 / LangGraphOpenMontage关键影响核心抽象Chain/Graph/StateComponent/Port/DataPacketLangChain 的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( queryDataType.STRING ) output_port OutputPort( answerDataType.STRING, sourcesDataType.LIST_OF_STRING ) def __init__(self): super().__init__() # 这里初始化你的 LangChain 复杂链 self.chain