Human-in-the-Loop

# 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

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

"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)}
    />
  ),
});

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"}
            disabled={!waiting}
            onChange={() => toggle(i)}
          />
          {step.description}
        </label>
      ))}
      <button disabled={!waiting} onClick={() => onConfirm(choices)}>
        Confirm
      </button>
    </div>
  );
}