에이전틱 생성형 UI (Agentic Generative UI)¶

에이전트가 다단계 작업을 처리하는 동안, CrewAI Flow의 실시간 상태를 UI로 렌더링해서 에이전트가 일하는 모습이 그대로 보이게 해줘요.

에이전트의 실시간 상태 그리기¶

어떤 작업은 단 하나의 도구 호출로 끝나지 않아요. 조사 작업이라든지, 다단계 계획이라든지, 오래 걸리는 작업 같은 경우죠. 그럴 때 사용자에게 보여줄 흥미로운 건 "결과 하나"가 아니라 "과정 그 자체"예요. 에이전틱 생성형 UI는 에이전트의 상태를 그리고, 그 상태가 바뀔 때마다 다시 그려줘요.

이 패턴은 두 부분으로 나뉘어요.

Flow가 작업하면서 진행 상황을 자기 상태(state)에 기록해요.
프론트엔드는 useAgent로 그 상태를 읽어 화면에 그려주고, 상태가 흘러 들어올 때마다 다시 렌더링해요.

Flow의 상태는 여러분이 전송(transport)을 직접 연결하지 않아도 AG-UI를 통해 프론트엔드까지 도달해요. Flow의 각 스텝(메서드) 경계에서 상태 스냅샷이 자동으로 발행되고, 오래 걸리는 스텝 중에는 copilotkit_emit_state를 직접 호출해서 중간 업데이트를 밀어 넣을 수도 있어요. 상태를 상속받아 필요한 필드를 추가하고, Flow에서 그 필드를 갱신하고, React에서 읽으면 돼요.

상태 기반 렌더링에는 커스텀 상태를 가진 Flow(Flow[AgentState])가 필요해요. Crew는 채팅 중심이라 커스텀 상태를 이렇게 노출하지 않으니, Crew를 쓸 때는 도구 렌더링(Tool-Based Generative UI)을 대신 사용해요.

라이브 태스크 플래너 만들기¶

이 예시는 요청을 열 개 안팎의 스텝으로 쪼개고, 그것을 체크리스트로 UI에 스트리밍하는 플래너를 만들어요. CrewAI 서버와 CopilotKit 프론트엔드가 이미 연결되어 있다고 가정해요. 아직 아니라면 Frontend Overview 가이드부터 시작하세요.

에이전트 상태에 나만의 필드 추가하기¶

CopilotKitState를 상속받아 UI에 필요한 상태를 선언해요. CopilotKitState는 이미 대화(messages)를 들고 있어서, 렌더링하고 싶은 다른 것 — 여기서는 태스크 스텝 목록 — 만 추가하면 돼요.

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를 갱신하고 싶다면 명시적으로 발행해요(아래에서 보여드려요).

Flow에서 상태로 진행 상황 기록하기¶

커스텀 상태(Flow[AgentState])로 Flow를 타입 지정하고, 모델이 그 상태를 채우게 해요. 여기서 LLM은 generate_task_steps 도구를 호출하는데, 스트리밍된 도구 호출이 대화에 들어오고 스텝들이 상태에서 보이게 돼요.

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)

LLM 호출을 copilotkit_stream으로 감싸면 어시스턴트의 토큰과 도구 호출이 생산되는 대로 프론트엔드로 스트리밍돼요. self.state에 기록한 스텝들은 이 스텝이 끝날 때 발행되는 상태 스냅샷에 담겨 전송돼요.

오래 걸리는 스텝 중간에 진행 상황 스트리밍하기 (선택)¶

자동 스냅샷은 스텝 경계에서 발행돼요. 그런데 단일 스텝이 꽤 많은 일을 하고, 체크리스트가 진행되는 모습을 즉시 보고 싶다면 copilotkit_emit_state로 중간 상태를 직접 발행하면 돼요. 호출할 때마다 현재 상태가 즉시 프론트엔드로 전달돼요.

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)

copilotkit_emit_state는 ag_ui_crewai.sdk에서 import해요. CopilotKit SDK가 필요하니 pip install "copilotkit[crewai]"로 설치해야 해요. 이건 스텝이 충분히 길어서 경계 스냅샷을 기다리면 반응이 느려 보일 때만 사용하세요.

Flow를 AG-UI로 서빙하기¶

Flow를 다른 것과 똑같이 자기만의 경로에 등록하면 돼요.

# 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",
)

서버·런타임·프로바이더 전체 설정은 Frontend Overview 가이드를 참고하고, CopilotKit 런타임 라우트에 에이전트(여기서는 task_planner)를 등록하는 걸 잊지 마세요.

React에서 실시간 상태 읽기¶

프론트엔드에서는 useAgent가 에이전트의 실시간 상태를 주고, 상태 변경을 구독하면 Flow가 업데이트를 기록할 때마다 컴포넌트가 다시 렌더링돼요.

"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가 스텝을 스트리밍할 때 체크리스트가 채워지는 걸 보여줘요.

다음은 어디로¶

상태 읽기는 기초가 되는 부분이에요. 이 위에 바로 쌓는 두 가이드가 있어요.

Shared State — 반대 방향을 다뤄요. UI에서 에이전트의 상태를 편집하고 Flow가 그 변경을 받아들이게 해요.
Predictive State — 도구의 진행 중 인자를 상태로 스트리밍해서, 커밋되기 전에도 UI가 작업 상태를 반영하게 해줘요.

D-SKET Canvas가
정식 런칭되었습니다.