CopilotKit × CrewAI Sub-Agents 实战Supervisor 委派多代理并实现实时委派日志【免费下载链接】CopilotKitThe Frontend Stack for Agents Generative UI. React, Angular, Mobile, Slack, and more. Makers of the AG-UI Protocol项目地址: https://gitcode.com/GitHub_Trending/co/CopilotKit导读本文围绕 CopilotKit 仓库中crewai-conversational-flows集成示例的 Sub-Agents 演示演示说明深入讲解一个 supervisor LLM 将工作委派给多个专业化子代理、并把每次委派实时流式渲染到前端的完整实现。读完本文你将掌握如何用 CrewAI 原生Crew构建子代理、如何把它们以 OpenAI 兼容工具的形式暴露给 supervisor、如何通过共享 agent state 驱动前端实时委派日志以及这条链路在 CopilotKit 前后端中的每一环代码落点。Demo 概览Supervisor 调度三个专业化子代理Sub-Agents 演示的核心是一条多代理委派链路一个 supervisor LLM 编排三个专门的子代理每个子代理被暴露为可调用的工具而每一次委派都通过共享 agent state 实时流进 UI。三个专业化子代理分别是research_agent负责收集事实产出 35 条要点writing_agent负责把简报与事实草拟成一段精炼文字critique_agent负责审阅草稿并给出 23 条可执行的改进意见。每个子代理在源码中都是真实的代理实体。CrewAI 版本里它们是三个单代理的Crew子代理构建源码LangGraph 参照版本中它们则是三个完整的create_agent(...)LangGraph 参照实现。Supervisor 通过工具调用来调度它们。每次调用tool包装的委派工具时wrapper 运行对应的子代理并向共享的delegations状态槽追加一条记录。前端左侧面板渲染这份delegations随着 supervisor 把工作分发给子代理而不断增长形成实时委派日志。交互方式点击建议提示词suggestion chip或自行输入提示词即可体验。仓库内置了三条演示提示词suggestions.tsProduce a short blog post about the benefits of cold exposure training. Research first, then write, then critique.Explain how large language models handle tool calling. Research, write a paragraph, then critique.Summarize the current state of reusable rockets in 1 polished paragraph, with research and critique.运行后观察左侧日志随 supervisor 依次执行 research → write → critique 而逐条填充。架构决策为什么用 Flow 工具调用而不是单个 hierarchical Crew这是整个演示最值得借鉴的设计决策。CrewAI 本身提供 hierarchical / sequential 的Process模式来内部编排子代理但这里刻意没有采用。源码模块注释给出了完整理由subagents.py 架构说明CrewAI 的 hierarchical / sequentialProcess模式在内部编排子代理并通过 AG-UI 桥只向上层暴露最终 crew 输出——每一个中间子任务 / 委派对客户端都是不透明的。也就是说如果用单个 supervisor Crew前端最多只能看到最终结果无法看到 research 完成、writing 进行中、critique 已完成这类中间过程。而本演示的硬性要求是每次委派都要向 state 追加一条Delegation记录且 UI 要渲染实时委派日志这强制要求每次委派都可见。因此最干净的架构是每个子代理都是一个真正的 CrewAI Crew原生 CrewAI 原语保证真实性supervisor 是一个由 litellm 驱动的 LLM把三个 crew 暴露为工具supervisor 的外层 wrapper Flow 在每次委派后发出状态快照STATE_SNAPSHOT。这个形态与langgraph-python参照实现同构——只是把每个子图替换为真正的Crew(agents[...], tasks[...])。对应关系在源码注释中明确写出subagents.py 顶部注释。后端实现CrewAI 版本后端核心文件是 src/agents/subagents.py完整实现了状态模型、子代理 crew、委派工具 schema 与 supervisor 循环。共享状态模型Delegation 与 AgentState委派日志的每一条记录由DelegationPydantic 模型定义SubAgentName Literal[research_agent, writing_agent, critique_agent] class Delegation(BaseModel): id: str sub_agent: SubAgentName task: str status: Literal[running, completed, failed] result: str class AgentState(CopilotKitState): delegations: List[Delegation] Field(default_factorylist)要点status支持running / completed / failed三态这是前端实现先显示 running、后切换为 completed的关键Delegation与 LangGraph 参照实现一一对应源码注释明确Mirrors the LangGraph reference 1:1这样前端类型可以跨运行时原样共享AgentState继承自CopilotKitState来自ag_ui_crewai包delegations字段通过 AG-UI 桥同步到前端这正是useAgent能读到它的基础。值得对照的是 LangGraph 版本的状态定义LangGraph AgentState那里delegations使用了Annotated[list[Delegation], operator.add]reducer。LangGraph 的多值并发更新需要用 reducer 累加否则会抛出INVALID_CONCURRENT_GRAPH_UPDATE。CrewAI Flow 版本没有这个约束直接以普通列表追加即可两版实现的差异恰好反映了两种运行时对共享状态语义的不同处理。子代理 Crew每个都是真实的 Crew每个子代理由_build_*_crew()工厂构建以research_agent为例def _build_research_crew() - Crew: researcher Agent( llmgpt-5.4, roleResearcher, goalProduce a concise bulleted list of 3-5 key facts on the topic., backstory( You are a research sub-agent. You gather and distil information into short, structured bullets. No preamble. ), verboseFalse, allow_delegationFalse, ) research_task Task( description( Topic: {task}\n\n Produce a concise bulleted list of 3-5 key facts about the topic. Each bullet ≤ 1 short sentence. No preamble or closing remarks. ), expected_output3-5 short bullets, one per line, prefixed with - ., agentresearcher, ) return Crew( agents[researcher], tasks[research_task], processProcess.sequential, verboseFalse, chat_llm_LLM, )写作 crew 与批判 crew 结构相同但 role/goal/backstory 与 Task 描述各不相同writing/critique 构建源码。三个子代理之间不共享内存或工具——supervisor 只能看到 crew 的最终原始输出通过Crew.kickoff(...)返回。代码还做了一项工程优化Crew 采用懒加载单例_get_research_crew()等只在首次被调用时构建一次。源码注释解释了两个动机Crew 构建成本高跨请求复用懒加载让导入开销小且被 aimock 模拟的测试不会在模块加载时触发任何 Crew 机制懒加载单例源码。委派工具OpenAI 兼容的 function schemasupervisor 通过 litellm 调用 LLM因此三个 crew 以纯 OpenAI 兼容工具 schema暴露而非 CrewAI 原生的 tool 对象def _delegation_tool(name: SubAgentName, description: str) - dict: return { type: function, function: { name: name, description: description, parameters: { type: object, properties: { task: { type: string, description: ( The full task / brief to hand off to the sub-agent. Include any facts or draft text the sub-agent will need. ), } }, required: [task], }, }, }每个工具都带一段精心撰写的 description指导 supervisor 何时调用、把什么放进task参数research_agentDelegate a research task to the research sub-agent. Use for gathering facts, background, definitions, or statistics.writing_agentPass relevant facts from prior research insidetask.critique_agentPass the draft insidetask.三个工具聚合为DELEGATION_TOOLS并派生DELEGATION_TOOL_NAMES集合用于运行时识别工具定义源码。Supervisor 系统提示词与委派循环Supervisor 的系统提示词明确告诉模型委派顺序SUPERVISOR_SYSTEM_PROMPT对于大多数非平凡请求按 research → write → critique 的顺序委派通过task参数传递事实/草稿自身消息保持简短并在结束后给出简明摘要。委派主循环supervise()start()装饰器标记为 Flow 入口是整条链路的心脏supervise 源码其核心逻辑为回合开始先发状态快照await copilotkit_emit_state(self.state)把当前状态含历史委派推给前端构造消息与工具列表system prompt 历史消息工具列表 前端注册的 actions 三个委派工具调用 LLMcopilotkit_stream(await acompletion(modelfopenai/{_LLM}, ..., toolstools, parallel_tool_callsFalse, streamTrue))无工具调用则结束说明 supervisor 已产出最终回答遍历所有工具调用防御性迭代防止并行调用被静默丢弃对每个委派工具解析task参数缺失时返回 tool-error 消息让模型下一轮自愈先追加一条statusrunning的Delegation并发一次状态快照UI 立刻显示进行中通过_kickoff_crew真正运行 crew把该条目替换为completed失败则为failed并追加ToolMessage回写消息历史再发一次状态快照遇到前端工具调用则退出循环由 AG-UI 客户端负责前端工具的回程。两个值得注意的工程细节1同步 crew 放进异步线程。_kickoff_crew用asyncio.to_thread(crew.kickoff, inputs{task: task})把同步的Crew.kickoff挪到工作线程避免阻塞 supervisor 的事件循环保证流式输出不被卡住源码。2委派轮数硬上限。_MAX_DELEGATION_ROUNDS 6一次完整的 research → write → critique 期望消耗 3 轮加一次总结6 轮是留足余量又不至于让 LLM 无限循环的折中源码注释。3失败信息脱敏。子 crew 抛异常时结果只记录异常类名exc.__class__.__name__并提示see server logs for details注释明确说明repr(exc)可能泄漏 URL、请求 ID 或部分凭证源码。前端实现useAgent 订阅状态 useRenderTool 内联渲染前端页面位于 src/app/demos/subagents/page.tsx整条数据通路可以拆成三层。1. Provider 绑定agentsubagentsCopilotKit runtimeUrl/api/copilotkit agentsubagents DemoContent / /CopilotKitagentsubagents与后端的subagentsFlow 端点一一对应见下文的链路说明。2. useAgent 订阅状态更新const { agent } useAgent({ agentId: subagents, updates: [UseAgentUpdate.OnStateChanged, UseAgentUpdate.OnRunStatusChanged], });OnStateChanged让前端在每次 STATE_SNAPSHOT 到来时拿到最新的agent.state.delegationsOnRunStatusChanged让前端感知 supervisor 是否还在运行。页面据此推导const delegations agentState?.delegations ?? []; const isRunning agent.isRunning; const activeSubAgent isRunning ? inferActiveSubAgent(delegations, agent.messages) : null;inferActiveSubAgentactive-subagent.ts是一个防御性实现它从消息流中反向查找最近一条还没有收到 ToolMessage 回复的子代理工具调用以此判断当前正在执行的子代理解析流式参数时先尝试严格 JSON.parse失败则用正则嗅探task: ...。这样即使 v2 消息形状跨运行时略有变化也能稳定工作。3. 实时委派日志与内联活动卡片DemoLayout把页面分为两栏demo-layout.tsx左侧是DelegationLog右侧是CopilotChat聊天面板与SupervisorActivityBanner。DelegationLogdelegation-log.tsx是核心展示组件头部显示Sub-agent delegations标题、Supervisor running脉冲徽章data-testidsupervisor-running和委派计数data-testiddelegation-count三个子代理角色以常驻指示 chip展示即使 supervisor 还没委派也能看到角色存在已触发的角色高亮、未触发的降透明每条委派条目渲染序号、角色徽章research / writing ✍️ / critique 各有专属配色、completed状态、Task 原文与结果正文。SubAgentActivityCardsubagent-activity-card.tsx通过useRenderTool把子代理工具内联渲染进聊天流——用户不必看侧栏就能在消息流里看到Researcher is gathering facts…这样的进行中状态useRenderTool( { name: research_agent, parameters: z.object({ task: z.string() }), render: ({ parameters, status, result }) ( SubAgentActivityCard subAgentresearch_agent task{parameters?.task} status{status as SubAgentToolStatus} result{typeof result string ? result : undefined} / ), }, [], );工具卡片的status会按inProgress → executing → complete三个阶段推进对应 badge 文案starting → running → done。聊天面板顶部还有SupervisorActivityBannersupervisor-activity-banner.tsx——一条吸顶横幅即使活动卡片滚出视口也始终显示当前正在运行的子代理 任务摘要。端到端链路从 Next.js 路由到 FastAPI Flow 端点整个请求链路清晰地串联了 CopilotKit 前端与 CrewAI 后端前端CopilotKitprovider 指向runtimeUrl/api/copilotkitsrc/app/api/copilotkit-subagents/route.ts 中HttpAgent被创建为${AGENT_URL}/conversational_flows/subagents其中AGENT_URL默认http://localhost:8000并通过new CopilotRuntime({ agents })注册subagentsagent后端 src/agent_server.py 遍历CONVERSATIONAL_FLOW_TYPES用add_crewai_flow_fastapi_endpoint(app, flow_type(), f/conversational_flows/{feature}, conversationalTrue, ...)为每个 Flow 挂载 FastAPI 端点CONVERSATIONAL_FLOW_TYPES中的subagents映射到_conversational_type(SubagentsFlow)conversational_flows.py且conversationalTrue意味着使用 CrewAI 的对话式会话路由Flow 在 AG-UI 桥的驱动下进入supervise()开始委派循环并持续copilotkit_emit_state。本地运行命令见 package.jsonconcurrently next dev --turbopack PYTHONPATH. python -m uvicorn agent_server:app --host 0.0.0.0 --port 8000 --reload前提条件包括后端OPENAI_API_KEY已设置FastAPI agent server 挂载了/subagentsFlow 端点。可验证依据e2e 测试与 QA 清单仓库为该演示提供了端到端测试 tests/e2e/subagents.spec.ts其中两个用例特别能说明实现的行为契约Summarize a topic pill produces 3 subagent cards (regression: delegations reducer)该 pill 历史上会因delegations状态键并发更新返回 HTTP 400INVALID_CONCURRENT_GRAPH_UPDATE。测试确认三个卡片全部到达done状态验证了状态累加机制的正确性Critic runs exactly once per pill click and stays done (no loop)点击后 critic 卡片数量恒为 1、状态恒为complete并保持 5 秒后复查。这防止 supervisor LLM 对同一草稿反复调用critique_agent造成卡片堆叠。与之呼应LangGraph 参照实现专门用_MAX_CRITIQUE_ITERATIONS 1硬性截断重复批判LangGraph 说明。此外qa/subagents.md 提供了一份人工 QA 清单覆盖空消息 no-op、歧义消息Hi 可直接回复而不委派或最多触发一次子代理、以及每条委派从running过渡到completed/failed、supervisor 结束后不得卡在running等验收标准。小结Sub-Agents 演示给出的是一条可复用的多代理可视化模式用 Flow 掌控编排循环把每个子代理实现为真实代理实体并伪装成工具用共享 agent state 承载委派记录再借助 CopilotKit 的useAgentuseRenderTool把过程实时、内联地呈现在前端。无论你的后端是 CrewAI Flow 还是 LangGraph这套supervisor 委派 共享状态日志 前端实时渲染的结构都可以原样迁移——这正是本演示作为仓库中跨运行时langgraph-python、mastra 等对齐样本的价值所在。【免费下载链接】CopilotKitThe Frontend Stack for Agents Generative UI. React, Angular, Mobile, Slack, and more. Makers of the AG-UI Protocol项目地址: https://gitcode.com/GitHub_Trending/co/CopilotKit创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考