CopilotKit 与 Langroid 双向共享状态实战：Shared State (Read + Write) 演示深度解析-平芜编程栈

CopilotKit 与 Langroid 双向共享状态实战：Shared State (Read + Write) 演示深度解析

【免费下载链接】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 仓库中 Langroid 集成示例Shared State (Read + Write)为线索，完整拆解「UI 与 Agent 双向读写同一状态对象」的实现模式：前端通过agent.setState把用户偏好写入共享状态，后端中间件每轮读取并注入系统提示词；Agent 通过set_notes工具把笔记写回状态，前端订阅状态变更后实时重渲染。读完本文，你将掌握 CopilotKituseAgent+ AG-UISTATE_SNAPSHOT事件驱动的双向共享状态实现原理、Langroid 侧的手写 SSE 管线细节，以及从 Next.js 路由到 FastAPI 端点的完整接线方式。

演示要解决的核心问题：状态只能单向流动怎么办？

在很多 Agent 应用中，状态流动是单向的：用户输入进 Agent，Agent 回复文本。但真实产品往往需要「双向共享状态」——UI 与 Agent 都持有并修改同一份数据。本演示展示的就是这种完整闭环：

UI → Agent（写入）：侧边栏表单（姓名 name、语气 tone、语言 language、兴趣 interests）通过agent.setState(...)写入state.preferences。后端中间件每轮对话读取该字段，注入系统提示词，让回复实时跟随表单变化。
Agent → UI（写入）：Agent 通过set_notes工具写入state.notes，侧边栏的笔记卡片在 Agent 每次更新时自动重渲染。
完整往返（Round-trip）：在侧边栏修改偏好后，Agent 的下一轮回复会明显体现出来——语气、语言、直呼你的名字。

该演示位于 showcase/integrations/langroid/src/app/demos/shared-state-read-write/README.md，是 LangGraph-Python 规范版本（showcase/integrations/langgraph-python/src/agents/shared_state_read_write.py）在 Langroid 集成中的移植，并在 showcase/integrations/langroid/PARITY_NOTES.md 中被列为 Wave 1 已移植的演示之一。

如何上手体验

按 showcase/integrations/langroid/package.json 中的dev脚本，同时启动 Next.js 前端与 FastAPI Agent 服务：

npm run dev # 等价于：concurrently "next dev --turbopack" "PYTHONPATH=. python -m uvicorn agent_server:app --host 0.0.0.0 --port 8000 --reload"

前端默认通过AGENT_URL（默认http://localhost:8000）连接后端，LLM 调用需要OPENAI_API_KEY（模型名由LANGROID_MODEL环境变量指定，默认gpt-4.1）。

先在侧边栏编辑偏好，然后依次尝试这些建议提问（由 suggestions.ts 通过useConfigureSuggestions预置）：

"Say hi and introduce yourself."
"Remember that I prefer morning meetings and that I don't eat dairy."
"Suggest a weekend plan based on my interests."

观察 Agent 的回复如何随偏好调整，以及当你让它「记住」某些事情时，侧边栏笔记卡片如何实时出现新笔记。

前端：UI 订阅与写入共享状态

用 useAgent 订阅状态变更

页面组件 page.tsx 是双向状态的中枢。它通过@copilotkit/react-core/v2提供的useAgent订阅 Agent 的每一次状态变更：

const { agent } = useAgent({ agentId: "shared-state-read-write", updates: [UseAgentUpdate.OnStateChanged], });

关键点在于updates: [UseAgentUpdate.OnStateChanged]：它把页面订阅到 Agent 的每一次状态变更事件，因此 Agent 通过set_notes工具改写的state.notes会触发组件重渲染，侧边栏笔记即时刷新。页面随后把agent.state拆成两个切片：

const agentState = agent.state as RWAgentState | undefined; const preferences = agentState?.preferences ?? INITIAL_PREFERENCES; const notes = agentState?.notes ?? [];

共享状态的 TypeScript 形状在 preferences-card.tsx 中定义：

export interface Preferences { name: string; tone: "formal" | "casual" | "playful"; language: string; interests: string[]; }

页面层把两个切片合并成完整的RWAgentState（{ preferences, notes }），这是前端与后端约定一致的唯一状态形状。

UI 写入：agent.setState 是唯一入口

前端写状态全部收敛到agent.setState。初次挂载时先播种一份默认偏好，保证 Agent 在第一轮就能读到内容：

useEffect(() => { if (!agentState?.preferences) { agent.setState({ preferences: INITIAL_PREFERENCES, notes: [], } as RWAgentState); } }, []);

表单每次编辑都直接写入 Agent 状态：

const handlePreferencesChange = (next: Preferences) => { agent.setState({ preferences: next, notes, // 保留 Agent 已写入的笔记 } as RWAgentState); };

注意这里每次写入都携带完整的状态对象（preferences+notes），避免覆盖掉 Agent 刚写入的笔记。「Clear」按钮清空笔记同样是写操作：agent.setState({ preferences, notes: [] })。

设计上有个细节值得借鉴：preferences-card.tsx 是一个完全不知道 Agent 存在的受控表单——每次编辑只触发onChange，由上层页面组件统一路由进agent.setState。表单（name输入框、tone/language下拉、interests兴趣徽章切换）与 Agent 状态接线被刻意分层，便于复用与测试。

UI 读取：NotesCard 只渲染、不写状态

notes-card.tsx 是「只读」侧的样板：它只接收父组件传入的notes数组并渲染列表，从不直接触碰 Agent 状态。因为父页面订阅了OnStateChanged，Agent 每调用一次set_notes写回新快照，笔记列表就会自动更新。唯一的写回点是 Clear 按钮，以onClearprop 暴露给父层，演示了「同一字段两个方向都能写」的能力。

后端：Langroid 侧的手写共享状态管线

Langroid 并没有原生的共享状态通道，因此在 showcase/integrations/langroid/src/agents/shared_state_read_write.py 中，直接基于 AG-UI 的STATE_SNAPSHOT事件手工实现了整套读写管线：每当 Agent 变更状态，就发出一个新的快照事件。

状态规范化：信任边界上的防御

AG-UI 把RunAgentInput.state类型化为Any，因此入站状态可能是None、列表甚至任意畸形结构。_normalize_state做了白名单式防御：非 dict 一律视为"无状态"，preferences必须是 dict，notes只保留字符串元素：

def _normalize_state(raw: Any) -> dict[str, Any]: if not isinstance(raw, dict): return {"preferences": {}, "notes": []} prefs = raw.get("preferences") if isinstance(raw.get("preferences"), dict) else {} notes_raw = raw.get("notes") notes = ( [n for n in notes_raw if isinstance(n, str)] if isinstance(notes_raw, list) else [] ) return {"preferences": prefs, "notes": notes}

注释中明确说明：前端 bug 不应导致整轮对话 500。

偏好注入：把 UI 写入变成系统提示词

build_preferences_system_message把 UI 供应的偏好渲染成系统消息字符串。一个值得注意的健壮性设计：tone 被限制在_VALID_TONES = frozenset({"formal", "casual", "playful"})闭集内，未知值静默丢弃（与 agent-config 演示的姿态一致——前端 bug 不应搞挂一轮对话）。若没有任何可用字段则返回None，调用方可以干净地跳过注入：

if not prefs: return None lines: list[str] = ["The user has shared these preferences with you:"] # - Name: ... # - Preferred tone: ...（仅当在合法闭集内） # - Preferred language: ... # - Interests: a, b, c lines.append( "Tailor every response to these preferences. Address the user " "by name when appropriate." )

在handle_run中，系统提示词由固定基底_SYSTEM_PROMPT与动态偏好消息拼接而成：

system_message = _SYSTEM_PROMPT if prefs_msg is not None: system_message = f"{_SYSTEM_PROMPT}\n\n{prefs_msg}"

这与 LangGraph 规范版的PreferencesInjectorMiddleware.wrap_model_call（读取request.state["preferences"]后request.override(messages=[prefs_message, *request.messages])）实现的是同一语义：每轮都在最前面注入偏好。

set_notes 工具：以 OpenAI 函数规范定义写回通道

Agent 写回状态的工具以 OpenAI 格式工具规范定义。规范中反复强调一个关键约定：必须传完整列表而非增量补丁（"This REPLACES the array"），每条笔记短于 120 字符。系统提示词_SYSTEM_PROMPT同样要求 "NEVER pass a partial diff — always the complete list"。这是因为共享状态是幂等替换语义，Agent 必须有能力复述全量内容，UI 才能无条件信任快照。

事件流：一次完整回合的 SSE 时序

handle_run返回StreamingResponse（text/event-stream），在 agent_server.py 中注册为POST /shared-state-read-write。一次带工具调用的回合按以下顺序发射事件：

RUN_STARTED：携带thread_id与随机生成的run_id；
STATE_SNAPSHOT：把入站状态原样回显为初始快照——既给 UI 订阅提供一个已知良好的基线，也让全新会话在 Agent 写入前就能看到空的notes数组；
调用 OpenAI：_call_openai(oai_messages, [_SET_NOTES_TOOL_SPEC])；
若响应包含set_notes调用（_extract_set_notes_args解析），依次发射TOOL_CALL_START→TOOL_CALL_ARGS（delta为 JSON 化的{"notes": [...]}）→TOOL_CALL_END，然后再次发射STATE_SNAPSHOT（携带更新后的state["notes"]），UI 据此重渲染；
构建带role: "tool"结果的续问消息数组，不带工具再调用一次 OpenAI，让模型输出自然语言确认（避免模型在确认回合重复调用工具）；
文本确认以TEXT_MESSAGE_START→TEXT_MESSAGE_CONTENT→TEXT_MESSAGE_END流式下发；
以RUN_FINISHED收尾。若_call_openai抛错，则发RUN_ERROR后紧跟RUN_FINISHED，错误消息只暴露异常类名，不泄漏内部细节。

整个事件序列完整覆盖 AG-UI 协议核心事件，与 showcase/integrations/langroid/src/agent_server.py 注释中 "RUN_STARTED / STATE_SNAPSHOT / TEXT_* / TOOL_CALL_* / RUN_FINISHED" 的说明完全一致。

为什么直接用 OpenAI 客户端而不是 Langroid 的 Agent 抽象

该后端刻意绕开 Langroid 的 agent 抽象，直接用openai.AsyncOpenAI()调用 Chat Completions（shared_state_read_write.py 顶部文档字符串说明了原因）：为了让 aimock 能够按完整消息历史做 fixture 匹配——包括对后续轮次中role: "tool"消息做hasToolResult匹配。_agui_messages_to_openai负责把 AG-UI 消息转为 OpenAI 格式并保留结构化字段（tool_calls、tool_call_id），这样工具调用后的确认回合在消息形状上可被稳定录制与回放。

运行时接线：Next.js 路由 → FastAPI 端点

前端页面中<CopilotKit runtimeUrl="/api/copilotkit" agent="shared-state-read-write">的请求最终落到专用路由 showcase/integrations/langroid/src/app/api/copilotkit-shared-state-read-write/route.ts。该路由的注释解释了为什么需要专用 runtime：统一的 Langroid Agent 端点（POST /）既不消费RunAgentInput.state，也不发射STATE_SNAPSHOT事件，而本演示两者都需要。

因此路由用一个指向AGENT_URL + "/shared-state-read-write"的HttpAgent实例（来自@ag-ui/client）包装，通过createCopilotRuntimeHandler以single-route模式暴露，并把"shared-state-read-write"同时注册为具名 agent 与 default。请求链路为：浏览器 → Next.js/api/copilotkit-shared-state-read-write→ FastAPIPOST /shared-state-read-write→handle_run的 SSE 流。

关键要点与可复用模式

双向状态 = 明确的所有权划分：preferences归 UI 所有、Agent 只读；notes归 Agent 所有、UI 只读（UI 的 Clear 是例外写回）。两端共用同一状态形状（RWAgentState↔ Python dict），是避免状态漂移的基础。
订阅驱动重渲染：useAgent({ updates: [UseAgentUpdate.OnStateChanged] })把 Agent 的每次状态变更都变成前端可观察的事件，这是 agent → UI 方向零额外轮询的关键。
写操作全量替换：无论是agent.setState还是set_notes工具，都携带完整状态而非 diff，配合STATE_SNAPSHOT的幂等快照语义，让 UI 永远可以信任最新快照。
无原生通道也能实现：当框架（如 Langroid）缺少共享状态原语时，基于 AG-UISTATE_SNAPSHOT事件手工构建读写管线是完全可行的路径，本演示即是可复制的参考实现。

如果要在此基础上扩展，可以继续研读同仓库中 LangGraph 规范版（showcase/integrations/langgraph-python/src/agents/shared_state_read_write.py），它用AgentMiddleware与Command(update=...)表达同样的双向语义，可作为对比 Langroid 手写管线与框架原生能力差异的绝佳样本。

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