휴먼 인 더 루프 (Human in the Loop)

휴먼 인 더 루프 (Human in the Loop)¶

사용자를 루프 안에 넣기¶

모든 단계를 사람의 확인 없이 자동으로 처리하면 안 되는 일이 있어요. 휴먼 인 더 루프(Human-in-the-Loop)는 에이전트 실행 중간에 멈춰서 프론트엔드에 인터랙티브 컴포넌트를 띄우고 사용자의 선택을 기다려요. 사용자가 결정을 내리면 에이전트는 그 선택을 가지고 실행을 이어가요.

동작 원리는 프론트엔드가 등록하는 도구(tool) 하나로 설명돼요. 모델이 그 도구를 호출하는 순간, 그 지점에서 실행이 멈추고 사용자가 응답할 때까지 대기해요. 어떤 것도 자동으로 진행되지 않아요 — respond()가 제어권을 돌려주기 전까지 에이전트는 그 자리에 멈춰 있어요.

아래 예시에서는 에이전트가 작업 단계 목록을 제안해요. 사용자는 각 단계를 켜거나 끈 다음 확인하고, 에이전트는 사용자가 승인한 내용 그대로 존중하며 계속 진행해요.

이 패턴은 Flows에서 동작해요. respond() 이후 Flow의 채팅 루프가 다시 진입하는 방식에 의존하죠 — 반환된 값이 도구 결과(tool result)로 돌아오고, 에이전트의 다음 턴이 그 값을 받아서 동작해요.

만들기¶

1. 프론트엔드 액션을 모델의 도구에 바인딩하기¶

Flow에서 프론트엔드가 등록한 액션을 모델의 도구 목록에 *self.state.copilotkit.actions로 추가해요. 이 액션들이 바로 프론트엔드가 useHumanInTheLoop로 등록한 도구예요. 이렇게 바인딩하면 모델이 그 도구를 호출할 수 있고, 그 도구 호출 지점에서 사용자가 응답할 때까지 실행이 멈춰요.

# human_in_the_loop_flow.py
from crewai.flow.flow import Flow, start, router, listen
from litellm import acompletion
from ag_ui_crewai.sdk import copilotkit_stream, CopilotKitState

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

    @router(start_flow)
    async def chat(self):
        system_prompt = (
            "You perform tasks for the user. When asked to do a task, call the "
            "tool the frontend provides so the user can approve or adjust the steps "
            "before you continue."
        )

        response = await copilotkit_stream(
            await acompletion(
                model="openai/gpt-4o",
                messages=[
                    {"role": "system", "content": system_prompt},
                    *self.state.messages,
                ],
                tools=[*self.state.copilotkit.actions],  # tools registered by the frontend
                parallel_tool_calls=False,
                stream=True,
            )
        )

        message = response.choices[0].message
        self.state.messages.append(message)
        return "route_end"

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

CopilotKitState는 프론트엔드가 등록한 액션을 self.state.copilotkit.actions에 담고 있어요. 모델이 그중 하나를 호출하면 그 지점에서 실행이 멈추죠. 사용자가 응답하면 반환된 값이 도구 결과로 self.state.messages에 들어오고, Flow가 chat으로 다시 돌아가서 모델이 그 결정을 바탕으로 동작해요.

2. Flow를 AG-UI로 서빙하기¶

다른 모든 에이전트와 마찬가지로, FastAPI 서버에서 add_crewai_flow_fastapi_endpoint로 Flow를 노출해요. 전체 서버·런타임·프로바이더 설정은 Frontend Overview 문서를 참고하세요.

# server.py
from fastapi import FastAPI
from ag_ui_crewai.endpoint import add_crewai_flow_fastapi_endpoint
from my_agents.human_in_the_loop_flow import HumanInTheLoopFlow

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

add_crewai_flow_fastapi_endpoint(
    app=app,
    flow=HumanInTheLoopFlow(),
    path="/human_in_the_loop",
)

3. 프론트엔드에 인터랙티브 도구 등록하기¶

useHumanInTheLoop는 에이전트가 멈추는 데 쓰는 도구를 등록하고, 인터랙티브 UI를 그릴 render 함수를 반환해요. 에이전트가 그 도구를 호출하면 컴포넌트가 나타나고, 사용자가 행동하면 respond()를 호출해서 에이전트를 다시 진행시켜요.

"use client";
import { useHumanInTheLoop } from "@copilotkit/react-core/v2";
import { z } from "zod";

useHumanInTheLoop({
  agentId: "human_in_the_loop",
  name: "generate_task_steps",
  parameters: z.object({
    steps: z.array(
      z.object({
        description: z.string(),
        status: z.enum(["enabled", "disabled", "executing"]),
      })
    ),
  }),
  render: ({ args, respond, status }) => (
    <StepReview
      steps={args.steps ?? []}
      // `status === "executing"` means the agent is waiting for the user
      waiting={status === "executing"}
      onConfirm={(chosen) => respond?.(chosen)}
    />
  ),
});

render 함수가 받는 값 세 가지를 짚어볼게요.

args — 모델이 만들어낸 도구 인자예요 (여기서는 제안된 steps). 모델이 생성하는 대로 스트리밍돼요.
status — 도구 호출의 생명주기예요. "executing"인 동안 에이전트는 멈춰서 사용자를 기다리고 있어요.
respond(value) — 사용자의 결정과 함께 에이전트를 재개해요. 에이전트의 다음 턴이 반환된 값을 보고 그에 맞춰 동작해요.

4. 사용자가 결정하고 응답하기¶

컴포넌트는 args.steps를 읽어서 사용자가 각 단계를 토글하게 하고, 최종 선택과 함께 respond()를 호출해요. 그 값이 에이전트가 이어서 사용하는 내용이에요.

function StepReview({ steps, waiting, onConfirm }) {
  const [choices, setChoices] = useState(steps);

  const toggle = (i) =>
    setChoices((prev) =>
      prev.map((s, idx) =>
        idx === i
          ? { ...s, status: s.status === "enabled" ? "disabled" : "enabled" }
          : s
      )
    );

  return (
    <div>
      {choices.map((step, i) => (
        <label key={i}>
          <input
            type="checkbox"
            checked={step.status === "enabled"}
            onChange={() => toggle(i)}
            disabled={!waiting}
          />
          {step.description}
        </label>
      ))}
      <button disabled={!waiting} onClick={() => onConfirm(choices)}>
        Confirm
      </button>
    </div>
  );
}

사용자가 Confirm을 누르면 respond()가 실행되고, 실행이 재개되면서 Flow의 chat 단계가 사용자의 선택을 메시지 히스토리에 담아 다시 돌아가요.

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