基于 Agno 的 CopilotKit 子代理编排实战：监督者委派模式与实时委派日志-迪斯科星球

基于 Agno 的 CopilotKit 子代理编排实战：监督者委派模式与实时委派日志

【免费下载链接】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 仓库中 Sub-Agents 演示为核心，深入剖析一套可复用的**多智能体委派（Multi-agent Delegation）**实现方案：一个监督者（Supervisor）LLM 通过工具调用，将任务分派给三个专业子代理（研究、写作、评审），并把每次委派实时推送到前端形成"委派日志"。读完本文，你将掌握：如何用 AgnoAgent定义"子代理即工具"的完整链路、如何借 AG-UI 协议的StateSnapshotEvent把后端共享状态回传到前端、以及如何用useAgent/useRenderTool渲染实时委派状态。本文以仓库内真实源码为唯一依据，所有代码片段均来自当前仓库。

演示总览：监督者 + 三个专业子代理

该演示位于showcase/integrations/agno/src/app/demos/subagents/，其核心思路一句话概括：监督者不亲自干活，而是把任务拆解后委派给三个各司其职的子代理，并将每笔委派记录写入共享状态。

三个子代理各自是独立的 AgnoAgent实例（见 subagents.py）：

research_agent：收集事实，输出 3–5 条要点列表；
writing_agent：根据简报与事实，产出一段润色好的草稿；
critique_agent：对草稿给出 2–3 条可执行的评审意见。

每个子代理都有独立的系统提示词，互不共享内存与工具，监督者只能看到子代理的最终文本输出。子代理统一使用gpt-4o-mini模型（_SUB_MODEL_ID = "gpt-4o-mini"），并设置 120 秒超时。

监督者本身也是一个 AgnoAgent（subagents.py）：

agent = Agent( model=OpenAIChat(id=_SUB_MODEL_ID, timeout=120), tools=[research_agent, writing_agent, critique_agent], description="Supervisor agent coordinating research / writing / critique sub-agents.", instructions=_SUPERVISOR_INSTRUCTION, tool_call_limit=10, )

其系统提示词（_SUPERVISOR_INSTRUCTION）明确要求：对大多数非平凡请求，按research → write → critique的顺序依次委派，把相关事实/草稿通过每个工具的task参数传递；如果某个子代理失败，如实向用户说明失败原因而非编造结果，并自行决定是否重试。

子代理即工具：委派工具的实现细节

演示的关键模式是"子代理以工具的形式暴露给监督者"。在 subagents.py 中，三个委派工具research_agent(task)、writing_agent(task)、critique_agent(task)都收敛到同一个_delegate流程：

def _delegate(run_context, *, sub_agent_name, sub_agent, task): entry_id = _append_delegation( run_context, sub_agent=sub_agent_name, task=task, status="running", result="", ) try: result = _invoke_sub_agent(sub_agent, task) except Exception as exc: # 只把异常类名暴露给上层，完整堆栈留在服务端日志 _update_delegation(run_context, entry_id=entry_id, status="failed", result=message) return {"status": "failed", "error": message} _update_delegation(run_context, entry_id=entry_id, status="completed", result=result) return {"status": "completed", "result": result}

委派条目写入共享状态

_append_delegation（subagents.py）把每次委派追加到run_context.session_state["delegations"]：

entry = { "id": entry_id, # uuid4 "sub_agent": sub_agent, # research_agent / writing_agent / critique_agent "task": task, "status": status, # running / completed / failed "result": result, } run_context.session_state["delegations"] = delegations

_update_delegation（subagents.py）则按entry_id定位并更新条目的状态与结果。如果条目已丢失（被其他环节替换了delegations数组），只记录一条logger.warning而绝不追加合成条目——这一保守行为与仓库中 google-adk 参考实现保持一致。值得注意的是，委派失败的错误信息只暴露异常类名，避免把携带 URL、请求 ID 或部分凭据的提供方错误串泄露到前端。

一次委派的完整生命周期

从状态槽视角看，一次委派经历三个阶段：

running：_append_delegation先插入一条status="running"的条目；
子代理执行：_invoke_sub_agent同步调用sub_agent.run(input=task)，取回result.content；
completed / failed：_update_delegation把最终结果写回同一条记录。

工具返回的字典{"status": "completed", "result": ...}同时成为监督者的工具结果（ToolMessage），供其下一步决策（如把研究结果作为写作任务的输入）。

说明：该演示的同名 README 沿用了跨框架共享的表述（如 LangGraph 的Command与create_agent）。在当前 Agno 实现中，这一模式由run_context.session_state的delegations槽位等价实现，其"追加共享状态 + 以工具消息回传结果"的语义完全一致。

共享状态如何到达前端：AG-UI 与 StateSnapshotEvent

Agno 自带的 AGUI 路由器默认不会向后端客户端发射StateSnapshotEvent，这会导致依赖useAgent({ updates: [OnStateChanged] })的前端收不到共享状态变化。为此，仓库在 agent_server.py 中实现了一个状态感知的 AGUI 处理器_run_agent_with_state_snapshot：

复刻官方agno.os.interfaces.agui.router.run_agent的流式逻辑；
抑制内层流的RUN_STARTED/RUN_FINISHED，在流结束后、发出自己的RunFinishedEvent之前，发射一条携带最终session_state的StateSnapshotEvent；
快照优先通过agent.aget_session_state(session_id=thread_id)（或同步的get_session_state）读取，确保合并了会话数据库中落盘的任何额外状态；读取失败则回退到内存快照。

随后通过_attach_state_aware_route将该处理器挂载到/subagents/agui路由（agent_server.py）。这是"双向共享状态"契约的关键：/subagents/agui与共享状态读写演示/shared-state-rw/agui使用同一套路由器。

在前端侧，Next.js 运行时通过 copilotkit/route.ts 把subagents代理名映射到该后端端点：

function createSubagentsAgent() { return new HttpAgent({ url: `${AGENT_URL}/subagents/agui` }); } // ... agents["subagents"] = createSubagentsAgent();

后端默认跑在AGENT_URL || "http://localhost:8000"，与agent_server.py中/subagents前缀的 AGUI 端点对应。

前端实时委派日志的实现

1. Provider 与 useAgent 订阅

page.tsx 中，CopilotKit提供者指定agent="subagents"，页面组件用useAgent订阅状态变化与运行状态变化：

<CopilotKit runtimeUrl="/api/copilotkit" agent="subagents"> <DemoContent /> </CopilotKit> const { agent } = useAgent({ agentId: "subagents", updates: [UseAgentUpdate.OnStateChanged, UseAgentUpdate.OnRunStatusChanged], });

前端据此读取agent.state.delegations（委派列表）与agent.isRunning（监督者是否在运行），两者共同驱动左侧委派日志。

2. 委派日志组件

delegation-log.tsx 的DelegationLog组件负责左侧面板渲染：

头部显示标题、"Supervisor running" 状态徽章（isRunning为真时出现）以及{delegations.length} calls计数；
中部始终渲染三个子代理的指示徽章（subagent-indicator-<role>），无论是否已委派过——便于用户和 e2e 套件一眼看清存在哪些子代理、哪些已被触发（data-fired）；
列表中每个delegation-entry展示序号、角色徽章、状态（completed）、任务描述与最终结果；空列表时提示"Ask the supervisor to complete a task..."。

Delegation的 TypeScript 类型与后端写入结构一一对应：

export interface Delegation { id: string; sub_agent: SubAgentName; // research_agent | writing_agent | critique_agent task: string; status: "completed"; result: string; }

3. 当前活跃子代理推断

active-subagent.ts 的inferActiveSubAgent采用防御式结构探测：遍历消息流，找出最近一条尚未收到ToolMessage回应的监督者工具调用（对research/writing/critique三个工具名精确匹配），并尽力从流式的部分 JSON 工具参数中解析出task字段（先严格JSON.parse，失败则用正则嗅探"task": "..."，最终回退占位文案）。若消息流为空，则回退到委派列表中最新一条非completed条目作为软信号。

该结果驱动聊天面板顶部的 supervisor-activity-banner.tsx 粘性横幅——即使内联卡片滚出视口，用户也能看到"某子代理正在运行哪个任务"。

4. 聊天流内的内联活动卡片

除左侧日志外，page.tsx 还用三个useRenderTool为每个子代理工具注册内联渲染器：

useRenderTool( { name: "research_agent", parameters: z.object({ task: z.string() }), render: ({ parameters, status, result }) => ( <SubAgentActivityCard subAgent="research_agent" task={parameters?.task} status={status as SubAgentToolStatus} result={typeof result === "string" ? result : undefined} /> ), }, [], );

subagent-activity-card.tsx 中的卡片状态沿inProgress → executing → complete推进，对应徽章文案starting → running → done；complete前显示"正在工作…"占位，完成后才挂载data-testid="subagent-result"的结果区块。三个子代理分别有独立的 emoji、角色文案与配色（研究 🔎 紫、写作 ✍️ 绿、评审 🧐 橙）。

5. 建议提示词

suggestions.ts 通过useConfigureSuggestions提供三个常驻建议 chip，每个都明确要求按 research → write → critique 顺序执行：

Write a blog post："Produce a short blog post about the benefits of cold exposure training. Research first, then write, then critique."
Explain a topic："Explain how large language models handle tool calling. Research, write a paragraph, then critique."
Summarize a topic："Summarize the current state of reusable rockets in 1 polished paragraph, with research and critique."

如何运行与体验

启动方式见 agno 集成包的 package.json 中的dev脚本：

npm run dev

该命令同时启动两部分：next dev --turbopack（Next.js 前端，含/api/copilotkit运行时路由）和PYTHONPATH=. python -m uvicorn agent_server:app --host 0.0.0.0 --port 8000 --reload（Agno 后端服务）。前端通过AGENT_URL（默认http://localhost:8000）代理到后端的/subagents/agui端点，依赖OPENAI_API_KEY环境变量提供模型能力。

体验路径：打开子代理演示页 → 点击任一建议 chip 或自行输入提示词 → 观察聊天流中的内联活动卡片依次出现"研究 → 写作 → 评审"，同时左侧委派日志逐条增长、计数更新、指示徽章点亮。用户也可以给监督者下达任意任务，例如让它在撰写段落前先做研究、再对成稿给出评审意见。

测试验证：端到端守护核心行为

仓库为这一演示配套了完整的 Playwright 端到端测试 tests/e2e/subagents.spec.ts，覆盖了四条关键行为：

页面加载完整性：输入框、3 个建议 pill、3 个常驻子代理指示徽章均可见；
三条提示词链路：每个 pill 点击后，三个角色卡片（subagent-card-<role>）都达到complete，且每个卡片的subagent-result非空、不含"Hi there! I'm your showcase assistant"等助手样板文本——该断言专门防止回归中 Writer/Critic 卡片泄漏聊天欢迎语的问题；
Summarize 回归：专门标注该 pill 历史上曾因delegations状态键未声明 reducer 而返回INVALID_CONCURRENT_GRAPH_UPDATE的 400 错误，测试确认当前实现可稳定到达全部三个卡片done状态；
Critic 恰好运行一次：点击后断言 critic 卡片数量为 1 且状态保持complete，再停留 5 秒复查数量与状态不变，防止"评审循环"（supervisor 反复重新进入 critic）回归。

测试通过data-testid系列选择器（copilot-suggestion、subagent-indicator-*、subagent-card-*、subagent-status、subagent-result）与前端组件解耦，前后端职责清晰。

小结

纵观整条链路，本演示给出了一套可直接复用的多智能体委派范式的完整参考实现：

后端（Agno）：子代理即Agent，监督者通过普通 Python 函数工具委派；委派记录写入session_state["delegations"]，状态经历 running → completed/failed；
协议层（AG-UI）：自定义路由器在每次 run 结束时发射StateSnapshotEvent，补齐 Agno 默认 AGUI 不推送状态的缺口；
前端（CopilotKit v2）：useAgent订阅OnStateChanged/OnRunStatusChanged驱动委派日志，useRenderTool在聊天流内渲染每笔委派的活动卡片。

如果你需要在其他 Agno 集成中复用此模式，需要三处配套改动：定义子代理与委派工具、挂载状态感知的 AGUI 路由、在 CopilotKit 运行时中把代理名映射到/xxx/agui端点。三者缺一，实时委派日志都无法工作。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

企业官网建设流程全解析