如何在 LangGraph Agent 中用 subgraphs 构建多节点 CopilotKit 工作流

如何在 LangGraph Agent 中用 subgraphs 构建多节点 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当你把 LangGraph 应用从单节点 agent 扩展成多个节点、甚至多个嵌套 agent 时前端要能实时看到子图subgraph内部的执行事件而不是只拿到最终结果。本文面向已经在用 LangGraph 接 CopilotKit 的开发者说明 subgraphs 在 CopilotKit 中的工作方式agent 侧无需额外配置前端订阅 agent 状态即可拿到子图节点的实时流式输出。适用前提是 agent 通过 AG-UILangGraph 的ag-ui-langgraph端点或 LangSmith 部署暴露给 CopilotKit Runtime参考路径为 LangGraph 集成 Quickstart。Subgraph 是什么什么场景用官方文档 Subgraphs 的定义A subgraph is simply a graph used as a node inside another graph.即一个图作为另一个图中的节点。文档给出了三类典型用途把图结构拆分成更小的、可复用的单元构建多 agent 系统multi-agent system复用一组节点node sets到多个图中让不同团队独立开发图的不同部分。也就是说如果你的图只是START → 单个 LLM 节点 → END这种结构用不到 subgraph当图中出现某个节点内部是一整条含多个节点、可能还有分支的子图时就是 subgraph 场景。Agent 侧subgraph 不需要额外配置CopilotKit 对流式输出的支持直接覆盖子图文档原话是CopilotKit supports streaming directly from subgraphs, so nested graphs can emit events in real time just like a single graph. Using this feature requires no extra steps on the agent side.即你在 LangGraph 中正常用StateGraph组织子图、把编译后的图传给LangGraphAGUIAgent不需要为子图单独加任何 CopilotKit 相关代码。agent 侧的代码仍按 Quickstart 的 FastAPI 路径组织见 quickstart.mdximport os from dotenv import load_dotenv from ag_ui_langgraph import add_langgraph_fastapi_endpoint from copilotkit import LangGraphAGUIAgent from fastapi import FastAPI from langgraph.graph import END, START, MessagesState, StateGraph from langchain_core.messages import SystemMessage from langchain_openai import ChatOpenAI from langgraph.checkpoint.memory import MemorySaver import uvicorn load_dotenv() async def mock_llm(state: MessagesState): model ChatOpenAI(modelgpt-4.1-mini) system_message SystemMessage(contentYou are a helpful assistant.) response await model.ainvoke( [ system_message, *state[messages], ] ) return {messages: response} graph StateGraph(MessagesState) graph.add_node(mock_llm) graph.add_edge(START, mock_llm) graph.add_edge(mock_llm, END) graph graph.compile( checkpointerMemorySaver() ) app FastAPI() add_langgraph_fastapi_endpoint( appapp, agentLangGraphAGUIAgent( namesample_agent, descriptionAn example agent to use as a starting point for your own agent., graphgraph, ), path/, ) def main(): Run the uvicorn server. uvicorn.run( main:app, host0.0.0.0, port8123, reloadTrue, ) if __name__ __main__: main()这段是 Quickstart 中可直接运行的最小 agent。要把它变成多节点工作流只需用 LangGraph 自身的StateGraphAPI 把graph内部节点换成你的子图结构例如某个节点的值是另一张已编译的 graphLangGraphAGUIAgent与端点注册方式不变。FastAPI 路径下compile(checkpointerMemorySaver())是必需的原因见 Quickstart 中 Why this tab compiles with a checkpointer 说明ag-ui-langgraph通过graph.aget_state(...)读取线程状态没有 checkpointer 会抛ValueError: No checkpointer setMemorySaver只存在于进程内存生产环境可换成持久化 saver如 Postgres。走 LangSmith 部署时则相反compile()不带checkpointer由 LangSmith 平台在运行时注入。启动 agent在 agent 目录下uv run main.py服务运行在main.py中uvicorn.run(...)指定的 8123 端口。Runtime 与前端订阅 agent 状态即可Runtime 侧按 Quickstart 配置即可FastAPI 部署路径下路由如下app/api/copilotkit/route.tsimport { CopilotKitIntelligence, CopilotRuntime, createCopilotRuntimeHandler, } from copilotkit/runtime/v2; import { LangGraphHttpAgent } from copilotkit/runtime/langgraph; const runtime new CopilotRuntime({ agents: { sample_agent: new LangGraphHttpAgent({ url: process.env.LANGGRAPH_DEPLOYMENT_URL || http://localhost:8123, }), intelligence: new CopilotKitIntelligence({ apiKey: process.env.CPK_INTELLIGENCE_API_KEY!, }), identifyUser: (request) ({ id: request.headers.get(x-user-id) ?? anonymous, name: request.headers.get(x-user-name) ?? Anonymous, }), }, }); const handler createCopilotRuntimeHandler({ runtime, basePath: /api/copilotkit, mode: single-route, }); export const POST handler;其中http://localhost:8123是文档给出的本地回退地址与 agent 侧 uvicorn 端口对应如果不用 CopilotKit Intelligence 平台可删除intelligence与identifyUser两项runtime 会回退到 SSE 模式加内存 runner聊天仍可用文档原文的说明。前端是 subgraphs 特性唯一需要关注的地方。文档要求subscribe to the agent state in your frontendReact 端写法来自 subgraphs.mdxconst { agent } useAgent({ agentId: sample_agent, }); // Access agent state as usual - subgraph streaming is handled automatically const state agent.state;Angular 端import { injectAgentStore } from copilotkit/angular; export class AgentStatusComponent { readonly store injectAgentStore(sample_agent); // Read this.store().state() in the template or another computed signal. // Subgraph streaming updates it automatically. }文档的关键结论subgraph 节点会像普通节点一样正常流式输出The subgraph nodes will stream as usual并且可以在子图中直接使用interrupt()。验证子图流式输出是否生效文档没有给出独立的subgraph 流式成功检查命令验证方式是端到端跑通后观察前端状态启动 agentuv run main.py确认服务在 8123 端口LangSmith 部署则用npx langchain/langgraph-cli dev --port 8123 --no-browserQuickstart 专门提醒不带--port 8123时langgraph dev默认监听 2024端口不匹配会导致 runtime 连接失败且没有额外诊断信息。启动前端npm run dev或 pnpm/yarn/bun 对应命令。在聊天界面发起一轮对话触发你构建的多节点流程如果前端通过useAgent或 Angular 的injectAgentStore读取的state随子图节点的执行实时更新说明子图事件已在流中。连接类问题按 Quickstart 的 Troubleshooting 章节排查常见项包括保持localhost不要换成 IP 字面量0.0.0.0是 bind-all 地址不能作为客户端 URL 目标agent 目录要有langgraph.json且其中引用了.envOpenAI API key 正确配置runtime 的LANGGRAPH_DEPLOYMENT_URL或回退值端口与 agent 实际监听端口一致runtime endpoint 路径与 provider 的runtimeUrl一致。相关能力与边界子图内可以直接调用interrupt()Human-in-the-Loop文档明确支持用法见 human-in-the-loop 章节。多 agent 编排的另一种做法是把多个独立 agent 交给 CopilotKit 路由Router Mode或用agentprop 锁定单 agentAgent Lock Mode见 Multi-Agent Flows。那是多 agent而非单图内嵌子图两者解决不同问题subgraphs 是 LangGraph 图内部的结构封装Router Mode 是 CopilotKit 层面对多个 agent 的调度。官方在线示例Feature Viewer 中的 subgraphs demo在文档页内以 iframe 形式嵌入可在文档站直接查看运行效果与对应代码。子图流式本身没有需要你在 agent 侧填写的配置项如果你按上述路径接好后前端仍只看到最终结果而非节点级更新优先核对 agent 是否按 Quickstart 的方式通过 AG-UI 端点暴露、端口是否一致而不是在子图代码里找问题。【免费下载链接】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），仅供参考