主题
智能体化生成式 UI
将你的 CrewAI Flow 的实时状态渲染为 UI，随智能体执行多步任务而更新。
渲染智能体的实时状态
有些工作无法塞进单次工具调用。一项研究任务、一个多步计划、一个长时间运行的工作：值得展示给用户的不是某一个结果，而是进度。智能体化生成式 UI 渲染智能体的状态，并在该状态每次变化时重新渲染。
这个模式分为两部分：
- 你的 Flow 在工作过程中把进度写入自己的状态。
- 你的前端用
useAgent读取该状态并渲染它，随状态流式传入而重新渲染。
Flow 的状态通过 AG-UI 到达前端，你无需搭建任何传输层。在 Flow 的每个步骤（方法）边界都会自动发出一次状态快照，你还可以在长时间运行的步骤中通过显式调用 copilotkit_emit_state 推送中间更新。你通过继承状态来添加自己的字段，在 Flow 中更新它们，并在 React 中读取它们。
NOTE
状态驱动的渲染需要一个带自定义状态的 Flow（Flow[AgentState]）。Crew 是面向对话的，不会以这种方式暴露自定义状态，因此使用 Crew 时请改用工具渲染。
构建一个实时任务规划器
本示例构建一个规划器，它把请求拆成大约十个步骤，并以清单的形式流式输出到 UI。它假定你已经搭好 CrewAI 服务器和 CopilotKit 前端。如果还没有，请从前端概览开始。
Add your own fields to the agent state
通过继承 CopilotKitState 来声明 UI 所需的状态。CopilotKitState 已经携带了对话（messages）；你再添加任何想要渲染的其他内容，这里是任务步骤列表。
python
from typing import List, Literal
from pydantic import BaseModel, Field
from ag_ui_crewai.sdk import CopilotKitState
class TaskStep(BaseModel):
description: str
status: Literal["enabled", "disabled"]
class AgentState(CopilotKitState):
steps: List[TaskStep] = Field(default_factory=list)AgentState 上的所有内容都会包含在前端收到的状态快照中。每个步骤边界都会自动发出一次快照，因此只需写入 self.state，UI 就能在步骤之间获取到更新。若要在长时间运行的步骤进行中更新 UI，请显式发出（见下文）。
Write progress into state from the Flow
用自定义状态为你的 Flow 标注类型（Flow[AgentState]），让模型去填充它。这里 LLM 调用 generate_task_steps 工具；流式输出的工具调用落入对话中，这些步骤就在状态中可见了。
python
from crewai.flow.flow import Flow, start
from litellm import acompletion
from ag_ui_crewai.sdk import copilotkit_stream
GENERATE_TASK_STEPS_TOOL = {
"type": "function",
"function": {
"name": "generate_task_steps",
"description": "Break a task into about 10 short imperative steps.",
"parameters": {
"type": "object",
"properties": {
"steps": {
"type": "array",
"items": {
"type": "object",
"properties": {
"description": {"type": "string"},
"status": {"type": "string", "enum": ["enabled"]},
},
"required": ["description", "status"],
},
},
},
"required": ["steps"],
},
},
}
class TaskPlannerFlow(Flow[AgentState]):
@start()
async def chat(self):
response = await copilotkit_stream(
await acompletion(
model="openai/gpt-4o",
messages=[
{"role": "system", "content": "Plan the task the user asks for."},
*self.state.messages,
],
tools=[GENERATE_TASK_STEPS_TOOL],
parallel_tool_calls=False,
stream=True,
)
)
message = response.choices[0].message
self.state.messages.append(message)用 copilotkit_stream 包裹 LLM 调用，就能在助手产生的 token 和工具调用生成时即流式输出到前端。你写入 self.state 的 steps 会在本步骤结束时发出的状态快照中一并发送。
Stream progress during a long step (optional)
自动快照在步骤边界触发。如果单个步骤做了大量工作，而你希望清单在执行过程中逐步填充，可以用 copilotkit_emit_state 自行发出中间状态。每次调用都会立即把当前状态推送到前端。
python
from ag_ui_crewai.sdk import copilotkit_emit_state
class TaskPlannerFlow(Flow[AgentState]):
@start()
async def execute(self):
for step in self.state.steps:
step.status = "disabled" # mark done as you go
await copilotkit_emit_state(self.state) # push update now
await do_work(step)从 ag_ui_crewai.sdk 导入 copilotkit_emit_state。它需要 CopilotKit SDK（pip install "copilotkit[crewai]"）。只有当某个步骤长到等待其边界快照会显得毫无响应时，才需要用它。
Serve the Flow over AG-UI
像注册其他任何 Flow 一样，在它自己的路径上注册它：
python
# server.py
from fastapi import FastAPI
from ag_ui_crewai.endpoint import add_crewai_flow_fastapi_endpoint
from my_agents.task_planner import TaskPlannerFlow
app = FastAPI(title="CrewAI Agent Server")
add_crewai_flow_fastapi_endpoint(
app=app,
flow=TaskPlannerFlow(),
path="/task_planner",
)完整的服务器、运行时和 provider 配置见前端概览，并记得在 CopilotKit 运行时路由中注册该智能体（这里是 task_planner）。
Read the live state in React
在前端，useAgent 会给你智能体的实时状态。订阅状态变化，这样每次 Flow 写入更新时你的组件都会重新渲染。
tsx
"use client";
import { useAgent, UseAgentUpdate } from "@copilotkit/react-core/v2";
function TaskPlan() {
const { agent } = useAgent({
agentId: "task_planner",
updates: [UseAgentUpdate.OnStateChanged],
});
const steps = agent?.state?.steps ?? [];
return (
<ul>
{steps.map((s, i) => (
<li key={i}>{s.description}</li>
))}
</ul>
);
}useAgent 返回 { agent }。有几点需要注意：
agent.state是实时的 Flow 状态。它的形状与你添加到AgentState的字段一致，因此agent.state.steps就是你的任务步骤列表。agent.isRunning告诉你智能体何时正在积极工作，可用于显示加载指示器或禁用输入。updates: [UseAgentUpdate.OnStateChanged]会在状态变化时重新渲染组件，因此清单会随 Flow 流式输出步骤而逐步填充。
接下来往哪里走
读取状态是基础。有两篇指南直接建立在其之上：