Human-in-the-Loop AI Agent

from pydantic import BaseModel

class ProposedAction(BaseModel):
    """Action proposed by the AI agent for human review."""
    action_type: str
    description: str
    reasoning: str

class ApprovalRequest(BaseModel):
    """Request sent to human reviewer."""
    request_id: str
    proposed_action: ProposedAction
    context: str
    requested_at: str

class ApprovalDecision(BaseModel):
    """Decision received from human reviewer."""
    request_id: str
    approved: bool
    reviewer_notes: str | None = None
    decided_at: str

from temporalio import activity
from openai import AsyncOpenAI
from dataclasses import dataclass

@dataclass
class OpenAIResponsesRequest:
    model: str
    instructions: str
    input: str

@activity.defn
async def create(request: OpenAIResponsesRequest) -> str:
    # Temporal best practice: Disable retry logic in OpenAI API client library.
    client = AsyncOpenAI(max_retries=0)

    resp = await client.responses.create(
        model=request.model,
        instructions=request.instructions,
        input=request.input,
        timeout=30,
    )
    
    return resp.output_text

from temporalio import activity
from models.approval import ProposedAction
import asyncio

@activity.defn
async def execute_action(action: ProposedAction) -> str:
    """Execute the approved action.
    
    In a real system, this would call external APIs, modify databases,
    or trigger other workflows based on the action_type.
    """
    activity.logger.info(
        f"Executing action: {action.action_type}",
        extra={"action": action.model_dump()}
    )
    
    return f"Successfully executed: {action.action_type}"

from temporalio import activity
from models.approval import ApprovalRequest

@activity.defn
async def notify_approval_needed(request: ApprovalRequest) -> None:
    """Notify external systems that human approval is needed.
    
    In a real system, this would send notifications via:
    - Email
    - Slack/Teams messages
    - Push notifications to mobile apps
    - Updates to approval queue UI
    """
    # Get workflow ID from activity context
    workflow_id = activity.info().workflow_id
    
    activity.logger.info(
        f"Approval needed for request: {request.request_id}",
        extra={"request": request.model_dump()}
    )
    
    # In a real implementation, you would call notification services here
    print(f"\n{'='*60}")
    print(f"APPROVAL REQUIRED")
    print(f"{'='*60}")
    print(f"Workflow ID: {workflow_id}")
    print(f"Request ID: {request.request_id}")
    print(f"Action: {request.proposed_action.action_type}")
    print(f"Description: {request.proposed_action.description}")
    print(f"Reasoning: {request.proposed_action.reasoning}")
    print(f"\nTo approve or reject, use the send_approval script:")
    print(f"  uv run python -m send_approval {workflow_id} {request.request_id} approve 'Looks good'")
    print(f"  uv run python -m send_approval {workflow_id} {request.request_id} reject 'Too risky'")
    print(f"{'='*60}\n")

from temporalio import workflow
from datetime import timedelta
from typing import Optional
import asyncio

with workflow.unsafe.imports_passed_through():
    from models.approval import ProposedAction, ApprovalRequest, ApprovalDecision
    from activities import openai_responses, execute_action, notify_approval_needed


@workflow.defn
class HumanInTheLoopWorkflow:
    def __init__(self):
        self.current_decision: Optional[ApprovalDecision] = None
        self.pending_request_id: Optional[str] = None

    @workflow.signal
    async def approval_decision(self, decision: ApprovalDecision):
        """Signal handler for receiving approval decisions from humans."""
        # Verify this decision is for the current pending request
        if decision.request_id == self.pending_request_id:
            self.current_decision = decision
            workflow.logger.info(
                f"Received approval decision: {'approved' if decision.approved else 'rejected'}",
                extra={"decision": decision.model_dump()}
            )
        else:
            workflow.logger.warning(
                f"Received decision for wrong request ID: {decision.request_id}, expected: {self.pending_request_id}"
            )

    @workflow.run
    async def run(self, user_request: str, approval_timeout_seconds: int = 300) -> str:
        """Execute an AI agent workflow with human-in-the-loop approval.
        
        Args:
            user_request: The user's request for the AI agent
            approval_timeout_seconds: How long to wait for approval (default: 5 minutes)
            
        Returns:
            Result of the workflow execution
        """
        workflow.logger.info(f"Starting human-in-the-loop workflow for request: {user_request}")

        # Step 1: AI analyzes the request and proposes an action
        proposed_action = await self._analyze_and_propose_action(user_request)
        
        workflow.logger.info(
            f"AI proposed action: {proposed_action.action_type}",
            extra={"proposed_action": proposed_action.model_dump()}
        )

        # Step 2: Request human approval
        approval_result = await self._request_approval(
            proposed_action, 
            user_request,
            timeout_seconds=approval_timeout_seconds
        )

        # Step 3: Handle the approval result
        if approval_result == "approved":
            workflow.logger.info("Action approved, proceeding with execution")
            result = await workflow.execute_activity(
                execute_action.execute_action,
                proposed_action,
                start_to_close_timeout=timedelta(seconds=60),
            )
            return f"Action completed successfully: {result}"
        
        elif approval_result == "rejected":
            workflow.logger.info("Action rejected by human reviewer")
            reviewer_notes = self.current_decision.reviewer_notes or 'None provided'
            return f"Action rejected. Reviewer notes: {reviewer_notes}"
        
        else:  # timeout
            workflow.logger.warning("Approval request timed out")
            timeout_msg = f"Action cancelled: approval request timed out after {approval_timeout_seconds} seconds"
            return timeout_msg

    async def _analyze_and_propose_action(self, user_request: str) -> ProposedAction:
        """Use LLM to analyze request and propose an action."""
        system_instructions = """
You are an AI assistant that analyzes user requests and proposes actions.
For each request, you should:
1. Determine what action needs to be taken
2. Provide a clear description of the action
3. Explain your reasoning for why this action addresses the request

Be thorough and clear in your analysis.

Respond with a JSON string in this structure:

{
  "action_type": "A short name for the action (e.g., \\"delete_test_data\\")",
  "description": "A clear description of what the action will do",
  "reasoning": "Your explanation for why this action addresses the request"
}
"""

        result = await workflow.execute_activity(
            openai_responses.create,
            openai_responses.OpenAIResponsesRequest(
                model="gpt-4o-mini",
                instructions=system_instructions,
                input=user_request,
            ),
            start_to_close_timeout=timedelta(seconds=30),
        )

        # Parse the JSON output into our ProposedAction model
        return ProposedAction.model_validate_json(result)

    async def _request_approval(
        self, 
        proposed_action: ProposedAction, 
        context: str,
        timeout_seconds: int
    ) -> str:
        """Request human approval and wait for response.
        
        Returns:
            "approved", "rejected", or "timeout"
        """
        # Generate unique request ID using workflow's deterministic UUID
        self.current_decision = None
        self.pending_request_id = str(workflow.uuid4())
        
        # Create approval request
        approval_request = ApprovalRequest(
            request_id=self.pending_request_id,
            proposed_action=proposed_action,
            context=context,
            requested_at=workflow.now().isoformat(),
        )

        # Send notification to external systems
        await workflow.execute_activity(
            notify_approval_needed.notify_approval_needed,
            approval_request,
            start_to_close_timeout=timedelta(seconds=10),
        )

        # Wait for approval decision with timeout
        try:
            await workflow.wait_condition(
                lambda: self.current_decision is not None,
                timeout=timedelta(seconds=timeout_seconds),
            )
            
            # Decision received
            if self.current_decision.approved:
                return "approved"
            else:
                return "rejected"
                
        except asyncio.TimeoutError:
            # Timeout waiting for approval
            return "timeout"

import asyncio
import logging

from temporalio.client import Client
from temporalio.worker import Worker

from workflows.human_in_the_loop_workflow import HumanInTheLoopWorkflow
from activities.openai_responses import create
from activities.execute_action import execute_action
from activities.notify_approval_needed import notify_approval_needed
from temporalio.contrib.pydantic import pydantic_data_converter


async def main():
    # Configure logging to see workflow.logger output
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    )
    client = await Client.connect(
        "localhost:7233",
        data_converter=pydantic_data_converter,
    )

    worker = Worker(
        client,
        task_queue="human-in-the-loop-task-queue",
        workflows=[HumanInTheLoopWorkflow],
        activities=[
            create,
            execute_action,
            notify_approval_needed,
        ],
    )
    
    await worker.run()


if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import sys
import uuid

from temporalio.client import Client

from workflows.human_in_the_loop_workflow import HumanInTheLoopWorkflow
from temporalio.contrib.pydantic import pydantic_data_converter


async def main():
    client = await Client.connect(
        "localhost:7233",
        data_converter=pydantic_data_converter,
    )

    default_request = "Delete all test data from the production database"
    user_request = sys.argv[1] if len(sys.argv) > 1 else default_request
    
    # Use a unique workflow ID so you can run multiple instances
    workflow_id = f"human-in-the-loop-{uuid.uuid4()}"
    
    print(f"Starting workflow with ID: {workflow_id}")
    print(f"User request: {user_request}")
    print("\nWorkflow will pause for approval. Watch the worker output for instructions.\n")

    # Submit the workflow for execution
    result = await client.execute_workflow(
        HumanInTheLoopWorkflow.run,
        args=[user_request, 300],  # user_request and approval timeout in seconds
        id=workflow_id,
        task_queue="human-in-the-loop-task-queue",
    )
    
    print(f"\nWorkflow completed!")
    print(f"Result: {result}")


if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import sys
from datetime import datetime, timezone

from temporalio.client import Client
from temporalio.contrib.pydantic import pydantic_data_converter
from models.approval import ApprovalDecision


async def main():
    if len(sys.argv) < 4:
        print("Usage: python -m send_approval <workflow_id> <request_id> <approve|reject> [notes]")
        print("\nExample:")
        example = "  python -m send_approval human-in-the-loop-123 abc-def-ghi approve 'Looks good'"
        print(example)
        sys.exit(1)

    workflow_id = sys.argv[1]
    request_id = sys.argv[2]
    decision = sys.argv[3].lower()
    notes = sys.argv[4] if len(sys.argv) > 4 else None

    if decision not in ["approve", "reject"]:
        print("Decision must be 'approve' or 'reject'")
        sys.exit(1)

    client = await Client.connect(
        "localhost:7233",
        data_converter=pydantic_data_converter,
    )

    approval_decision = ApprovalDecision(
        request_id=request_id,
        approved=(decision == "approve"),
        reviewer_notes=notes,
        decided_at=datetime.now(timezone.utc).isoformat(),
    )

    # Get workflow handle and send signal
    handle = client.get_workflow_handle(workflow_id)
    await handle.signal("approval_decision", approval_decision)

    decision_type = 'approval' if approval_decision.approved else 'rejection'
    print(f"Sent {decision_type} signal to workflow {workflow_id}")


if __name__ == "__main__":
    asyncio.run(main())

# In start_workflow.py, change the timeout parameter:
result = await client.execute_workflow(
    HumanInTheLoopWorkflow.run,
    args=[user_request, 30],  # 30 second timeout for testing
    id=workflow_id,
    task_queue="human-in-the-loop-task-queue",
)