Human-in-the-loop

from framework.graph import NodeSpec

# Node 1: Intake (client-facing)
intake_node = NodeSpec(
    id="intake",
    name="Research Intake",
    description="Discuss the research topic with the user, clarify scope",
    node_type="event_loop",
    client_facing=True,  # This node interacts with humans
    input_keys=["topic"],
    output_keys=["research_brief"],
    system_prompt="""
You are a research intake specialist. The user wants to research a topic.
Have a brief conversation to clarify what they need.

**STEP 1 — Read and respond:**
1. Read the topic provided
2. If it's vague, ask 1-2 clarifying questions
3. If it's clear, confirm your understanding

**STEP 2 — After the user confirms, call set_output:**
- set_output("research_brief", "A clear paragraph describing what to research")
""",
)

# Node 3: Review (client-facing)
review_node = NodeSpec(
    id="review",
    name="Review Findings",
    description="Present findings to user and decide next steps",
    node_type="event_loop",
    client_facing=True,  # Another HITL checkpoint
    input_keys=["findings", "sources", "gaps"],
    output_keys=["needs_more_research", "feedback"],
    system_prompt="""
Present the research findings to the user clearly and concisely.

**STEP 1 — Present:**
1. **Summary** (2-3 sentences of what was found)
2. **Key Findings** (bulleted, with confidence levels)
3. **Gaps** (what's still unclear)

Ask: Are they satisfied, or do they want deeper research?

**STEP 2 — After the user responds, call set_output:**
- set_output("needs_more_research", "true")  — if they want more
- set_output("needs_more_research", "false") — if satisfied
""",
)

from dataclasses import dataclass
from enum import StrEnum

class HITLInputType(StrEnum):
    """Type of input expected from human."""
    FREE_TEXT = "free_text"      # Open-ended text response
    STRUCTURED = "structured"    # Specific fields to fill
    SELECTION = "selection"      # Choose from options
    APPROVAL = "approval"        # Yes/no/modify decision
    MULTI_FIELD = "multi_field" # Multiple related inputs

@dataclass
class HITLQuestion:
    """A single question to ask the human."""
    id: str
    question: str
    input_type: HITLInputType = HITLInputType.FREE_TEXT
    options: list[str] = field(default_factory=list)  # For SELECTION
    fields: dict[str, str] = field(default_factory=dict)  # For STRUCTURED
    required: bool = True
    help_text: str = ""

@dataclass
class HITLRequest:
    """Formal request for human input at a pause node."""
    objective: str              # What we're trying to accomplish
    current_state: str          # Where we are in the process
    questions: list[HITLQuestion]
    missing_info: list[str]     # What information is needed
    instructions: str = ""
    examples: list[str] = field(default_factory=list)

@dataclass
class HITLResponse:
    """Human's response to a HITL request."""
    request_id: str
    answers: dict[str, Any]     # {question_id: answer}
    raw_input: str = ""         # Raw text if provided
    response_time_ms: int = 0

@dataclass
class ExecutionResult:
    """Result of executing a graph."""
    success: bool
    output: dict[str, Any]
    paused_at: str | None = None        # Node ID where paused for HITL
    session_state: dict[str, Any]       # State to resume from

# Example: Research agent with multiple HITL checkpoints
edges = [
    EdgeSpec(id="intake-to-research", source="intake", target="research"),
    EdgeSpec(id="research-to-review", source="research", target="review"),
    # If user wants more: loop back to research
    EdgeSpec(
        id="review-to-research-feedback",
        source="review",
        target="research",
        condition=EdgeCondition.CONDITIONAL,
        condition_expr="needs_more_research == True",
    ),
    # If user satisfied: proceed to report
    EdgeSpec(
        id="review-to-report",
        source="review",
        target="report",
        condition=EdgeCondition.CONDITIONAL,
        condition_expr="needs_more_research == False",
    ),
]

class HITLProtocol:
    @staticmethod
    def parse_response(
        raw_input: str,
        request: HITLRequest,
        use_haiku: bool = True,
    ) -> HITLResponse:
        """Parse human's raw input into structured response.
        
        Uses Haiku to intelligently extract answers for each question.
        """
        # If multiple questions asked, uses LLM to extract each answer
        # Falls back to simple parsing if LLM unavailable

nodes = [
    intake_node,    # HITL: Clarify research topic
    research_node,  # Autonomous: Search and compile
    review_node,    # HITL: Review findings, decide next steps
    report_node,    # HITL: Present final report, answer questions
]

# Graph visits client-facing nodes at strategic checkpoints:
# 1. Start: Clarify what to research
# 2. Middle: Review findings before finalizing
# 3. End: Deliver results and handle follow-ups