报错修复(Debug->Fix):自动检测错误连线、异常参数、缺失依赖,输出修复建议与可执行方案,支持直接对当前画布进行 Debug 工作流改写(Rewrite):直接描述不满意之处与新诉求,Copilot 自动重排工作流结构、替换/新增关键节点,结合你的环境推荐最优参数范围 可度量的改进(Measurable):用 GenLab 将“玄学调参”变为对比可视化与可复现的实验。 ComfyUI-Copilot V2.0是基于多智能体协作框架(Multi-Agent)构建,并通过MCP接入本地服务与ComfyUI官方生态的Tools。
⭐ 星标历史 精细化可控生成 作为 ComfyUI 的 UI 插件,Copilot 在“高质量”“可控生成”维度更强调“从结果可见、从过程可控、在工程上可复现”。依托多智能体与结构化工具链,Copilot 将“可控性”落在三层:意图→结构→参数,并通过 GenLab 提供对比与复现能力。
高质量与可控性:可落地案例
通过“工作流改写 + 模型/LoRA 推荐 + 参数映射”确保同一角色在多镜头、一致风格下保持五官与妆面一致;支持“锁定构图/角色,仅替换风格”的一键策略。 构图/姿态/光照的可控(Structure Control)
对接常见控制类节点(如姿态/边缘/深度等)并提供“节点智能补全 + 下游子图推荐”,让构图受控同时维持细节质量;支持“锁定布局,开放质感细节”的软硬约束组合。 面向电商/物料场景,提供“高分辨率重建链路 + 去伪影参数模板”,减少边缘锯齿与纹理拉伸;支持多方案A/B对比(主观评分 + SSIM/LPIPS 等客观指标)并一键回溯最优快照。 可复现与过程可追溯(Reproducibility)
全流程快照(工作流结构、模型版本、关键参数、种子)与实验记录,保证团队协作下结果可复现、可审计;失败自动回退与参数范围保护,降低“玄学调参”引起的崩溃概率。 可复用所有端到端模型的能力,下面是电商场景的案例展示
我希望将如下的淘宝女装图,模特换成黑人,并添加营销文案
nano banana可以保证衣服主体不变,但文字添加上效果很差
qwen image可以添加营销文案,但主体保持做不好
ComfyUI可以组合所有端到端模型的优势,以及所有图片可控组件,例如Controlnet,来生成更可控的生产图
电商原图背景杂乱
nano banana把人物改为黑人,改变站的朝向,添加背景
qwen image edit + controlnet添加营销文案
与基模方案的差异化优势(对比 Nano Banana / Qwen Image Edit)
一句话总结:Copilot 的差异化在于“以工程化的结构化可控,支撑高质量生成与团队级复现”。基模方案更偏“单次编辑/快速出图”,而 Copilot 更适合“可控、可比、可复用”的生产场景。
架构以及设计思路 Canvas 插件形态选择: 基于生态流量获取的考虑: 我们决定采用 ComfyUI 插件形式而非独立应用开发。这一选择不仅能够直接触达 ComfyUI 的核心用户群体,还能充分利用 ComfyUI 的生态流量,降低用户获取成本。
ComfyUI 生态中的插件实现方式: ComfyUI 生态中常见的插件实现方式主要有两种:
Custom Nodes: 通过自定义节点扩展功能,适合技术型用户。这种方式灵活性高,但需要用户具备一定的技术背景。
UI Plugins: 提供可视化界面,更适合小白用户。这种方式能够显著降低使用门槛,提升交互体验。
用户群体分析: 经过对目标用户群体的深入分析,我们发现大多数用户更倾向于简单易用的工具。因此,我们最终选择开发 UI Plugins,以最大化降低使用门槛,同时提升交互体验。
核心架构设计: V1.0 的局限性:ComfyUI-Copilot V1.0 在插件侧仅实现了前端交互能力,所有逻辑都放在后端。随着产品功能的扩展,这一架构逐渐暴露出以下问题:
V2.0 的改进:为了解决这些问题,V2.0 演变为以下架构:
插件端:在 Canvas 和 Copilot-Local 中安置 MCP-Client,支持调用远程工具(SSE)和本地服务(stdio)。
远程调用:包括工作流生成器(SD/Flux/API Node)、节点推荐、模型推荐、节点智能补全、图像评估等。
本地服务:包括工作流运行、一键安装节点、一键安装模型、参数修改/参数本地映射等。
服务端:开放 MCP-Server,提供 SSE Tools 的方式,支持工作流生成、节点推荐、模型推荐等功能。
本地模型支持:部分用户可以直接调用本地安装的开源模型(如 Llama3),满足个性化需求。
Copilot-Local | MultiAgent 问题背景:如果仅依靠单一的 Agent-Tools 能力来实现复杂功能(如工作流修改和 Debug),会导致 Agent 挂载过多 Tools,性能下降且难以维护。
解决方案:采用分层架构,将复杂场景拆分为多个子任务,由不同的 Agent 处理,每个 Agent 仅绑定必要的 Tools。
Master Agent:负责整体协调和决策,与用户交互,并与 RewriteAgent 协作。通过 Handoff 机制传递上下文信息,确保任务无缝衔接。
Debug Agent:负责工作流的 Debug,识别错误类型并调用 LinkAgent、ParameterAgent 和 WorkflowBugfixAgent 进行修复。
Rewrite Agent:负责工作流的改写,根据用户需求调用 RAG 体系,召回相关经验和节点信息,生成优化后的工作流。
Copilot-Remote | RAG 服务端能力:MCP-Server 提供 SSE Tools,支持插件端的 MCP-Client 调用。核心功能包括:
技术栈选择 OpenAI Agents (Python):
a. 原生支持多智能体协作:完美契合 MasterAgent 协调多个子 Agent 的设计需求。
b. 标准化 Tool 注册/发现机制:便于分层管理不同 Agent 的专属工具集,同时支持便捷配置 MCP。
c. 内置 Context 管理:支持复杂的工作流调试会话保持,并通过 Handoff 机制实现上下文信息的无缝传递。
MasterAgent MCP Client & MCP Server 在 ComfyUI-Copilot V2.0 中,MasterAgent 通过 MCP(Multi-Agent Control Protocol)的 SSE(Server-Sent Events)方式挂载了 4 个核心工具(Tools),以实现与后端服务的高效交互。以下是具体的实现代码:
async with MCPServerSse( params= { "url": BACKEND_BASE_URL + "/mcp-server/mcp", "timeout": 300.0, }, cache_tools_list=True, client_session_timeout_seconds=300.0 ) as server: triage_agent = Agent( name="ComfyUI-Copilot", instructions="...", mcp_servers=[server], handoffs=[rewrite_agent], )
技术细节补充: - MCP-SSE 的作用:SSE 是一种轻量级的实时通信协议,适用于单向数据推送场景。在这里,它用于MasterAgent 与后端服务之间的高效通信,确保工具列表的实时更新和任务状态的同步。 - triage_agent 的设计:作为 MasterAgent 的核心组件,triage_agent 负责协调任务分发,并通过 handoffs 机制将特定任务(如工作流改写)传递给 rewrite_agent。 为了兼容现有的 FASTAPI 体系,同时支持 MCP 和传统 API 调用,我们采用了以下架构设计:
async with MCPServerSse( params= { "url": BACKEND_BASE_URL + "/mcp-server/mcp", "timeout": 300.0, }, cache_tools_list=True, client_session_timeout_seconds=300.0 ) as server: triage_agent = Agent( name="ComfyUI-Copilot", instructions="...", mcp_servers=[server], handoffs=[rewrite_agent], )
FastMCP 的作用:它是一个轻量级框架,用于将 MCP 协议集成到 FASTAPI 中,支持 SSE 和传统 HTTP 请求的混合模式。
工具注册:通过 @mcp.tool() 装饰器,可以将函数注册为 MCP 工具,供 MasterAgent 调用。例如,recall_workflow 工具用于召回工作流数据。
RAG & BenchMark体系 为了从海量信息中精准匹配用户需求,我们设计了一套完整的 RAG(Retrieval-Augmented Generation)体系,并辅以 Benchmark 评估机制。
离线处理环节承担了较重的计算任务,所有复杂耗时的运算都预先完成并存储入库。在线服务保持轻量化设计,确保快速响应以提升用户体验。 离线处理主要分为四个步骤:
数据清洗:对不同类型数据进行分类处理,重点解决文档、图文和多语言内容的结构化问题。例如对节点文档进行标题分段处理,Lora信息主要存储在图片中,通过多模态模型解析整合图文信息,对多语言内容进行统一翻译
信息结构化:将节点文档转化为结构化数据;提取Lora底模作为过滤标签;构建节点关联知识图谱
向量化处理:通过embedding技术生成向量数据,与结构化数据共同入库
为实现类似Cursor代码补全的工作流补全功能,系统会在用户选中节点后推荐多个下游子图。通过将复杂工作流解构为若干简化子图连接的方式,帮助大模型更好地理解工作流结构。运用图算法提取高频出现的子图模式,将其固化为可复用的公共组件。
在线流程充分利用离线处理生成的结构化数据:
用户输入往往存在表述不完整或模糊的情况,需经Agent进行语义改写和补全,再通过元数据过滤消除干扰项
采用双路径召回策略兼顾召回广度和结果精度:向量召回覆盖语义相似案例,关键词匹配确保结果相关性,两者加权融合排序
对召回结果进行相关性评估,部分场景引入业务指标(如在节点召回中结合GitHub stars权重)进行最终排序
Debug Agent 设计思路 参考Cursor的Debug流程,运行脚本后获得错误信息,会尝试进行一次改写,之后会基于改写的代码再运行一次测试脚本。经过多次错误反馈和修改,最终完全解决报错问题。这一流程的核心在于错误捕获、智能分析、迭代修复和验证闭环。
错误捕获阶段:通过工作流验证工具自动运行当前工作流,捕获结构化错误日志。这一步的关键是确保错误信息的完整性和可解析性,为后续分析提供基础。
智能分析阶段:错误分类器将错误路由到对应领域的Agent(连接/参数/结构)。例如:
连接错误(connection_error)交给Link Agent处理。
参数异常(value_not_in_list)交给Parameter Agent处理。
结构问题(node_compatibility)交给Workflow Bugfix Agent处理。
迭代修复阶段:各Agent使用专用工具进行修复,每次修改后自动保存工作流快照。例如:
Parameter Agent负责调整参数值或建议模型下载。
Workflow Bugfix Agent负责处理节点兼容性问题或移除无效节点。
验证闭环:修复后自动触发二次验证,形成调试闭环(最多6次迭代)。如果验证通过,则输出成功;否则,继续分析并修复。
Multi Agent结构 debug_agent = Agent( name="Debug Agent", instructions="You determine which agent to use based on the user's homework question", tools=[run_workflow, analyze_error_type, save_current_workflow], ) parameter_agent = Agent( name="Parameter Agent",
handoff_description="", tools=[find_matching_parameter_value, get_model_files, suggest_model_download, update_workflow_parameter, get_current_workflow], handoffs=[debug_agent], ) link_agent = Agent( name="Link Agent", handoff_description="", tools=[analyze_missing_connections, apply_connection_fixes, get_current_workflow, get_node_info], handoffs=[debug_agent], ) workflow_bugfix_default_agent = Agent( name="Workflow Bugfix Agent", handoff_description="", tools=[get_current_workflow, get_node_info, update_workflow], handoffs=[debug_agent], ) debug_agent.handoffs = [link_agent, workflow_bugfix_default_agent, parameter_agent]
每个Agent专注于特定任务,并通过上下文路由协议实现高效协作。以下是关键设计点:
连接错误(connection_error) ➔ Link Agent
参数异常(value_not_in_list) ➔ Parameter Agent
结构问题(node_compatibility) ➔ Workflow Bugfix Agent
基础工具(如get_current_workflow)跨Agent共享,避免重复实现。
专用工具(如apply_connection_fixes)按需挂载,确保功能聚焦。
工具输出自动合并到全局上下文(如Link Agent的连接修复记录)。
结构化数据通过tool返回给前端,减轻LLM负担。
处理结果回传,结构化数据通过tool返回给前端。整个MultiAgent体系里的所有tool的返回,都可以通过Event的方式统一处理。这个实现是让LLM专注于核心功能的重点,至此MultiAgent体系可以通过Tool返回各种复杂格式的JSON数据,而不会给LLM带来压力。
debug_agent = Agent( name="Debug Agent", instructions="You determine which agent to use based on the user's homework question", tools=[run_workflow, analyze_error_type, save_current_workflow], ) parameter_agent = Agent( name="Parameter Agent", handoff_description="", tools=[find_matching_parameter_value, get_model_files, suggest_model_download, update_workflow_parameter, get_current_workflow], handoffs=[debug_agent], ) link_agent = Agent( name="Link Agent", handoff_description="", tools=[analyze_missing_connections, apply_connection_fixes, get_current_workflow, get_node_info], handoffs=[debug_agent], ) workflow_bugfix_default_agent = Agent( name="Workflow Bugfix Agent", handoff_description="", tools=[get_current_workflow, get_node_info, update_workflow], handoffs=[debug_agent], )
debug_agent.handoffs = [link_agent, workflow_bugfix_default_agent, parameter_agent]
实现这一整个MultiAgent的结构是很快的,但很快我们遇到了非常多问题。MultiAgent是非常难以Debug的,而且我们使用了OpenAI Agents的框架,有许多技术细节需要处理,对我们来说都是呈现出黑盒的形态。接下来的工作才是最费时间的,那就是,让MultiAgent更稳定,更可控,更聪明。
聪明思路:上下文控制,只给必要信息 让MultiAgent里的单个LLM专注于具体的某个任务,若不做任何配置,handoff的时候默认传递完整的上下文。
filter:当发生Agent交接时,默认传递完整上下文。但可以通过input_filter限制传递的信息量,例如移除所有工具调用历史,确保新Agent只关注必要信息。 from agents import Agent, handoff from agents.extensions import handoff_filters agent = Agent(name="FAQ agent") handoff_obj = handoff( agent=agent, input_filter=handoff_filters.remove_all_tools, )
InputData:在某些情况下,当大模型进行交接时,你希望它能够提供一些数据,且仅提供这些数据,不要传递冗长的完整上下文。例如,假设要交接给“升级代理”,你可能希望提供一个理由,以便进行记录。
from pydantic import BaseModel from agents import Agent, handoff, RunContextWrapper class EscalationData(BaseModel): reason: str async def on_handoff(ctx: RunContextWrapper[None], input_data: EscalationData): print(f"Escalation agent called with reason: {input_data.reason}") agent = Agent(name="Escalation agent") handoff_obj = handoff( agent=agent, on_handoff=on_handoff, input_type=EscalationData, )
聪明思路:“人工智能”,增强确定性 有多少人工就有多少智能。分离出来确定性的工作和必须让LLM做决策的,确定性的工作通过编码实现,最小化LLM产出难度和长度,只让LLM做决策。
Param Agent是在参数异常的时候,把参数修改为正常的值,这种情况下如果也让LLM返回一个庞大的工作流,就修改了里面的几个参数字段,类似的场景会凭空给LLM增加负担,容易产生幻觉。
与之类似,工作流连线的场景,如果完全让LLM来实现,会变得非常被动,输入一个庞大的工作流,连好几根线以后,再输出几个庞大的工作流。下面以Link Agent为例:
# Tool枚举出所有有待连线的节点和对应的参数 @function_tool async def analyze_missing_connections() -> str: """ 分析工作流中缺失的连接,枚举所有可能的连接选项和所需的新节点。 返回格式说明: - missing_connections: 缺失连接的详细列表,包含节点ID、输入名称、需要的数据类型等(仅包含required输入) - possible_connections: 现有节点可以提供的连接选项 - universal_inputs: 可以接受任意输出类型的通用输入端口 - optional_unconnected_inputs: 未连接的可选输入列表,包含节点ID、输入名称、配置信息等 - required_new_nodes: 需要创建的新节点类型列表 - connection_summary: 连接分析的统计摘要(包含optional输入的统计) """ try: # 从context获取session_id,通过session_id从db获取workflow session_id = get_session_id() workflow_data = get_workflow_data(session_id) if not workflow_data: return json.dumps({"error": "No workflow data found for this session"}) # 获取当前节点信息 object_info = await get_object_info() ------------------------------------------------------------------------ # 引导LLM返回要添加的节点和需要连接的线 # Tool对工作流进行修改,并通过Event反馈给到前端 @function_tool def apply_connection_fixes(fixes_json: str) -> str: """批量应用连接修复,fixes_json应为包含修复指令的JSON字符串""" ... return json.dumps([{ "type": "workflow_update", "data": { "workflow_data": workflow_data, "changes": { "applied_fixes": applied_fixes, "failed_fixes": failed_fixes } } }])
智能工具管理妙招:让AI不再手忙脚乱 当系统里塞满30多个功能工具时,就像让新手厨师同时操作十个灶台,很容易手忙脚乱。
V1版本中,多个子Agent挂载大量通用工具,导致职责交叉、决策混乱
如何分配角色以充分利用不同代理的专长,并拆解任务到各代理,是提高协作效率的关键。
我们在开发ComfyUI-Copilot V2.0时,总结出两招管理秘诀:
第一招:工具分类打包术
每个工具就像厨房的调料瓶,名称要一看就懂(比如"参数校验器"、"连线小助手")
输入输出要规范,就像调料瓶的开口大小要统一(强制类型标注和固定返回格式)
发现黄金搭档:像"洗菜→切菜"这样的固定流程,就打包成预制菜
参考Link Agent的连线修复功能,把常见操作流程固化
第二招:AI团队分工法
基层员工(L1):专精3-5个工具,像Param Agent只管参数问题
小组长(L2):协调多个工具,像Link Agent负责整个工作流连线
给每个工具贴"特征标签"(比如"参数处理"、"图形连接")
Tracing助手 - Langsmith 在MultiAgent场景下,debug变得非常痛苦,报错都报的层次很深,极难排障,这种情况下,需要接入Tracing来协助排障和调试,OpenAI Agents推荐使用Langfuse,但目前还是Langsmith最好用,集成Langsmith的方法如下:
import os os.environ['LANGCHAIN_TRACING_V2'] = "true" os.environ['LANGCHAIN_API_KEY'] = "xxx" import asyncio from agents import Agent, Runner, set_trace_processors, set_tracing_disabled, set_default_openai_api from langsmith.wrappers import OpenAIAgentsTracingProcessor set_tracing_disabled(False) set_default_openai_api("chat_completions") async def main(): agent = Agent( name="Captain Obvious", instructions="You are Captain Obvious...", model="gpt-4.1-2025-04-14-GlobalStandard", ) question = "hello" result = await Runner.run(agent, question) print(result.final_output) if __name__ == "__main__": set_trace_processors([OpenAIAgentsTracingProcessor()]) asyncio.run(main())
RewriteAgent
聪明思路:从Prompt Engineering到Context Engineering Context Engineering 是一门系统性学科,专注于设计、构建并维护一个动态系统,该系统负责在 Agent 执行任务的每一步,为其智能地组装出最优的上下文组合,以确保任务能够被可靠、高效地完成。
长上下文带来成本与协同压力,更易暴露四类上下文失效:污染、干扰、混淆、冲突。它们常彼此耦合,并直接损害推理稳定性与跨代理传递。Context Engineering可以通过对上下文进行智能管理与压缩,仅将高价值的结论与信息注入上下文(相当一部分 token 是没有价值的分析),从而规避上述风险。
数据流向示例:
# 上下文驱动架构示例 import asyncio from dataclasses import dataclass from agents import Agent, RunContextWrapper, Runner, function_tool @dataclass class UserInfo: name: str uid: int @function_tool async def fetch_user_age(wrapper: RunContextWrapper[UserInfo]) -> str: """Fetch the age of the user. Call this function to get user's age information.""" return f"The user {wrapper.context.name} is 47 years old" async def main(): user_info = UserInfo(name="John", uid=123) agent = Agent[UserInfo]( name="Assistant", handoff=[agentB], ) agentB = Agent( name="B Agent", tools=[fetch_user_age], ) result = await Runner.run( starting_agent=agent, input="What is the age of the user?",
context=user_info, ) print(result.final_output) # The user John is 47 years old. if __name__ == "__main__": asyncio.run(main())
聪明思路:五指不沾阳春水 任何的Agent的设计,挂载多个tools,都会给LLM带来更多要求和上下文信息量。针对特别复杂的任务,模型能力有限的情况下,我们应当返璞归真,让LLM什么别的都不做,所有的信息采集都提前做好,以结构化的数据格式存到context里,然后让LLM拿着已经准备好的所有上下文,专心的处理这个复杂的任务。
# RewriteAgent把所有的信息采集都做好,存储到context里 @function_tool async def get_node_info(node_class: str) -> str: """节点元数据采集器""" try: object_info = await get_object_info() if node_class in object_info: node_data = json.dumps(object_info[node_class], ensure_ascii=False) # 上下文持久化存储 get_rewrite_context().node_infos[node_class] = node_data return node_data except Exception as e: return json.dumps({"error": f"元数据获取失败: {str(e)}"}) # 上下文驱动的工作流生成器 def build_llm_context(rewrite_context) -> str: """结构化上下文构建器""" return f""" ## 核心要素 * 业务意图: {rewrite_context.rewrite_intent} * 当前状态: {rewrite_context.current_workflow} * 环境数据: {json.dumps(rewrite_context.node_infos, ensure_ascii=False)} * 领域知识: {rewrite_context.rewrite_expert or '基础规则'} """ # 精简LLM交互接口 def generate_workflow(context_str: str) -> dict: """上下文驱动的工作流生成""" return client.chat.completions.create( model=WORKFLOW_MODEL_NAME, messages=[{ "role": "system", "content": "你是一个工作流生成专家,请根据以下结构化上下文生成方案:" }, { "role": "user", "content": context_str }], response_format=RewriteResponse # 强类型响应约束 )
快来试用ComfyUI-Copilot吧 Github:https://github.com/AIDC-AI/ComfyUI-Copilot
感谢你看到这里,添加小助手 AIGC_Tech 加入官方 AIGC读者交流群,下方扫码加入 AIGC Studio 星球,获取前沿AI应用、AIGC实践教程、大厂面试经验、AI学习路线以及IT类入门到精通学习资料等,欢迎一起交流学习💗~