공유 상태 (Shared State)¶

CrewAI 에이전트의 상태와 앱의 UI를 양방향으로 동기화해서, 어느 한쪽에서 수정하면 그 변경이 다른 쪽에도 전달되도록 해요.

하나의 상태, 두 방향¶

공유 상태는 에이전트와 UI가 함께 읽고 쓰는 단일 상태 객체예요. 에이전트는 작업하면서 상태를 갱신하고, React 컴포넌트는 그 상태를 실시간으로 렌더링하죠. 그리고 사용자가 UI에서 같은 상태를 수정하면, 그 변경이 다시 흘러가서 다음 턴에 에이전트가 그 내용을 보게 됩니다.

전형적인 예가 레시피예요. 에이전트가 레시피를 작성하면 사용자가 재료나 조리법 하나를 수정하고, 에이전트는 수정된 버전부터 이어서 작업하죠. 어느 한쪽이 상태를 "소유"하는 게 아니라, 양쪽이 상태를 공유하는 겁니다.

공유 상태는 커스텀 상태를 가진 Flow에서 동작해요. CopilotKitState를 상속하는 AgentState를 정의하고, Flow를 Flow[AgentState]로 타입 지정하면 됩니다. Crew는 커스텀 상태를 가지지 않기 때문에 이 패턴은 Flow에서만 쓸 수 있어요.

동작 방식¶

1단계: Flow에 공유 상태 정의하기¶

CopilotKitState를 상속하면 에이전트가 CopilotKit의 메시지 파이프라인을 그대로 유지하고, 거기에 필요한 필드를 추가해요. 여기서 공유되는 필드는 recipe입니다.

# recipe_flow.py
import json
from typing import List, Optional
from pydantic import BaseModel, Field
from crewai.flow.flow import Flow, start, router, listen
from litellm import acompletion
from ag_ui_crewai.sdk import copilotkit_stream, CopilotKitState

class Ingredient(BaseModel):
    name: str
    amount: str

class Recipe(BaseModel):
    title: str
    ingredients: List[Ingredient] = Field(default_factory=list)
    instructions: List[str] = Field(default_factory=list)

class AgentState(CopilotKitState):
    recipe: Optional[Recipe] = None

2단계: 에이전트가 상태 읽고 쓰기¶

에이전트는 현재 상태를 시스템 프롬프트에 덤프해서 읽어오고, self.state.recipe에 할당해서 다시 씁니다. generate_recipe 도구는 모델이 갱신된 레시피를 구조화된 인자로 반환할 수 있게 해줘요.

GENERATE_RECIPE_TOOL = {
    "type": "function",
    "function": {
        "name": "generate_recipe",
        "description": "Generate or modify the recipe.",
        "parameters": {
            "type": "object",
            "properties": {"recipe": {"type": "object"}},
            "required": ["recipe"],
        },
    },
}

class SharedStateFlow(Flow[AgentState]):
    @start()
    @listen("route_follow_up")
    async def start_flow(self):
        pass

    @router(start_flow)
    async def chat(self):
        # The current shared state is visible to the model.
        system_prompt = f"""You help the user build a recipe.
        Current recipe: {self.state.model_dump_json(indent=2)}
        Modify it by calling generate_recipe."""

        response = await copilotkit_stream(
            await acompletion(
                model="openai/gpt-4o",
                messages=[
                    {"role": "system", "content": system_prompt},
                    *self.state.messages,
                ],
                tools=[*self.state.copilotkit.actions, GENERATE_RECIPE_TOOL],
                parallel_tool_calls=False,
                stream=True,
            )
        )
        message = response.choices[0].message
        self.state.messages.append(message)

        if message.tool_calls:
            call = message.tool_calls[0]
            if call.function.name == "generate_recipe":
                args = json.loads(call.function.arguments)
                self.state.recipe = Recipe(**args["recipe"])  # write to shared state
                self.state.messages.append({
                    "role": "tool",
                    "content": "Recipe updated.",
                    "tool_call_id": call.id,
                })
                return "route_follow_up"
        return "route_end"

    @listen("route_end")
    async def end(self):
        pass

이걸 단방향이 아니라 공유로 만들어 주는 건 두 가지예요. self.state를 프롬프트에 덤프한다는 건 에이전트가 항상 최신 레시피(UI에서 사용자가 수정한 내용 포함)에서 작업한다는 뜻이고, self.state.recipe에 할당하면 그 새 값이 스텝이 끝날 때 연결된 클라이언트로 보내지는 상태 스냅샷에 담기게 됩니다. 긴 스텝이 진행되는 동안에도 갱신을 보내려면 copilotkit_emit_state로 명시적으로 내보내면 돼요(Agentic Generative UI 참고).

3단계: Flow를 AG-UI로 서빙하기¶

FastAPI 앱에서 add_crewai_flow_fastapi_endpoint로 Flow를 노출하고, CopilotKit 런타임에 등록합니다. 전체 서버·런타임 설정은 Frontend Overview에서 확인할 수 있어요.

# server.py
from fastapi import FastAPI
from ag_ui_crewai.endpoint import add_crewai_flow_fastapi_endpoint
from recipe_flow import SharedStateFlow

app = FastAPI(title="CrewAI Agent Server")

add_crewai_flow_fastapi_endpoint(
    app=app,
    flow=SharedStateFlow(),
    path="/shared_state",
)

4단계: UI에서 상태 읽고 쓰기¶

useAgent 훅 하나로 두 방향을 모두 다룹니다. 공유 상태는 agent.state에서 읽고, agent.setState(...)로 다시 씁니다. OnStateChanged를 구독하면 에이전트가 상태를 갱신할 때마다 컴포넌트가 다시 렌더링돼요.

"use client";
import { useAgent, UseAgentUpdate } from "@copilotkit/react-core/v2";

function RecipeEditor() {
  const { agent } = useAgent({
    agentId: "shared_state",
    updates: [UseAgentUpdate.OnStateChanged],
  });

  const state = agent?.state as { recipe?: Recipe } | undefined;
  const isLoading = agent?.isRunning;

  const recipe = state?.recipe;

  // setState replaces the whole state object, so spread the current
  // state and override only the field you changed. Passing just
  // `{ recipe }` would drop messages and other runtime fields.
  const updateRecipe = (patch: Partial<Recipe>) =>
    agent?.setState({ ...(agent.state ?? {}), recipe: { ...(recipe ?? {}), ...patch } });

  return (
    <div>
      <input
        value={recipe?.title ?? ""}
        disabled={isLoading}
        onChange={(e) => updateRecipe({ title: e.target.value })}
      />
      {/* render inputs for ingredients and instructions the same way */}
    </div>
  );
}

agent.state는 공유 상태를 읽고, agent.setState(...)는 다시 써서 다음 턴에 에이전트가 그 변경을 보게 만들며, agent.isRunning은 에이전트가 현재 작업 중인지 알려줘요.

setState는 상태 객체 전체를 교체하지 병합하지 않습니다. 항상 현재 상태를 펼친({ ...agent.state, ... }) 다음 바꿀 필드만 덮어써야 해요. 그렇지 않으면 에이전트가 의존하는 대화 기록 같은 런타임 필드들이 사라집니다.

양방향 루프¶

여러 조각을 합치면, 레시피 객체 하나가 양방향으로 동기화돼요.

에이전트가 수정하면 UI가 갱신돼요. Flow가 self.state.recipe에 값을 할당하면, 그 새 값이 스텝의 상태 스냅샷에 실려 나가고 OnStateChanged가 입력 필드를 다시 렌더링합니다.
사용자가 수정하면 에이전트가 보게 돼요. UI의 변경이 agent.setState(...)를 호출하고, Flow가 self.state를 프롬프트에 덤프하기 때문에 에이전트는 다음 턴에 수정된 레시피에서 작업하게 되죠.

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