Human-in-the-Loop Patterns

from strands import Agent

def get_human_approval(action: str, context: dict) -> bool:
    """
    Request human approval for an action.
    
    Args:
        action: Description of the action to approve
        context: Additional context for decision
    
    Returns:
        True if approved, False if rejected
    """
    print(f"\n🔔 Human Approval Required")
    print(f"Action: {action}")
    print(f"Context: {json.dumps(context, indent=2)}")
    
    while True:
        response = input("\nApprove? (yes/no): ").lower().strip()
        if response in ['yes', 'y']:
            return True
        elif response in ['no', 'n']:
            return False
        print("Please enter 'yes' or 'no'")

# Use in agent workflow
def execute_with_approval(agent: Agent, query: str):
    # Get agent's proposed action
    result = agent(query)
    
    # Request approval
    if get_human_approval(
        action="Execute agent recommendation",
        context={"query": query, "recommendation": result}
    ):
        # Execute the action
        return execute_action(result)
    else:
        return "Action cancelled by user"

from strands import Agent
from strands.tools import tool
import json

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """
    Send an email after human approval.
    
    Args:
        to: Recipient email address
        subject: Email subject line
        body: Email body content
    """
    # Display proposed email
    print("\n" + "="*60)
    print("📧 EMAIL APPROVAL REQUEST")
    print("="*60)
    print(f"To: {to}")
    print(f"Subject: {subject}")
    print(f"\nBody:\n{body}")
    print("="*60)
    
    # Get approval
    approval = input("\nSend this email? (yes/no): ").lower().strip()
    
    if approval in ['yes', 'y']:
        # Actually send the email
        send_via_smtp(to, subject, body)
        return f"✅ Email sent to {to}"
    else:
        return "❌ Email cancelled by user"

# Agent with approval-gated tools
agent = Agent(
    model=model,
    system_prompt="You are a helpful email assistant",
    tools=[send_email]  # Tool requires human approval
)

@tool
def execute_trade(
    symbol: str,
    action: str,  # "buy" or "sell"
    quantity: int,
    price: float
) -> str:
    """
    Execute a stock trade after human approval.
    
    Args:
        symbol: Stock ticker symbol
        action: "buy" or "sell"
        quantity: Number of shares
        price: Price per share
    """
    total_value = quantity * price
    
    # Display trade details
    print(f"\n💰 TRADE APPROVAL REQUEST")
    print(f"Action: {action.upper()} {quantity} shares of {symbol}")
    print(f"Price: ${price:.2f} per share")
    print(f"Total: ${total_value:,.2f}")
    
    # Get approval with additional verification
    confirm = input(f"\nType '{symbol}' to confirm: ").upper()
    
    if confirm == symbol:
        # Execute trade
        result = broker_api.execute_trade(
            symbol=symbol,
            action=action,
            quantity=quantity,
            price=price
        )
        return f"✅ Trade executed: {action} {quantity} {symbol} @ ${price}"
    else:
        return "❌ Trade cancelled"

import streamlit as st
from agno.agent import Agent

def content_approval_workflow():
    st.title("Content Publishing Workflow")
    
    # Initialize session state
    if 'draft_content' not in st.session_state:
        st.session_state.draft_content = None
    if 'approved' not in st.session_state:
        st.session_state.approved = False
    
    # Step 1: Generate content
    topic = st.text_input("Enter topic for article")
    
    if st.button("Generate Draft") and topic:
        with st.spinner("Generating content..."):
            agent = Agent(
                model=model,
                instructions="Write a comprehensive article"
            )
            st.session_state.draft_content = agent.run(topic).content
    
    # Step 2: Review and approve
    if st.session_state.draft_content:
        st.subheader("📄 Draft Content")
        
        # Editable draft
        edited_content = st.text_area(
            "Review and edit content:",
            value=st.session_state.draft_content,
            height=400
        )
        
        col1, col2 = st.columns(2)
        
        with col1:
            if st.button("✅ Approve & Publish", type="primary"):
                # Publish content
                publish_to_cms(edited_content)
                st.success("✅ Content published successfully!")
                st.session_state.draft_content = None
        
        with col2:
            if st.button("❌ Reject"):
                st.session_state.draft_content = None
                st.warning("Draft rejected")
                st.rerun()

import streamlit as st

def multi_stage_approval(workflow_steps: list):
    """
    Multi-stage approval workflow with progress tracking.
    
    Args:
        workflow_steps: List of (step_name, agent_function) tuples
    """
    if 'current_step' not in st.session_state:
        st.session_state.current_step = 0
    if 'results' not in st.session_state:
        st.session_state.results = {}
    
    # Progress bar
    progress = st.session_state.current_step / len(workflow_steps)
    st.progress(progress)
    
    # Current step
    if st.session_state.current_step < len(workflow_steps):
        step_name, agent_func = workflow_steps[st.session_state.current_step]
        
        st.subheader(f"Step {st.session_state.current_step + 1}: {step_name}")
        
        # Execute step
        if st.button(f"Run {step_name}"):
            with st.spinner(f"Executing {step_name}..."):
                result = agent_func()
                st.session_state.results[step_name] = result
        
        # Show result and approval
        if step_name in st.session_state.results:
            st.markdown("**Result:**")
            st.write(st.session_state.results[step_name])
            
            col1, col2 = st.columns(2)
            with col1:
                if st.button("✅ Approve & Continue"):
                    st.session_state.current_step += 1
                    st.rerun()
            
            with col2:
                if st.button("🔄 Regenerate"):
                    del st.session_state.results[step_name]
                    st.rerun()
    else:
        st.success("✅ All steps approved!")
        st.balloons()

from pydantic import BaseModel
import json

class EmailDraft(BaseModel):
    to: str
    subject: str
    body: str
    cc: list[str] = []
    bcc: list[str] = []

def email_with_edits(agent: Agent, request: str) -> EmailDraft:
    """
    Generate email draft and allow human editing.
    """
    # Generate draft
    draft = agent.run(request, output_schema=EmailDraft).content
    
    print("\n📧 Generated Email Draft")
    print(json.dumps(draft.model_dump(), indent=2))
    
    # Allow editing
    print("\nEdit fields (press Enter to keep current value):")
    
    to = input(f"To [{draft.to}]: ") or draft.to
    subject = input(f"Subject [{draft.subject}]: ") or draft.subject
    
    print(f"\nBody (current):\n{draft.body}")
    print("\nEnter new body (or press Enter twice to keep current):")
    lines = []
    while True:
        line = input()
        if line == "" and (not lines or lines[-1] == ""):
            break
        lines.append(line)
    
    body = "\n".join(lines).strip() or draft.body
    
    # Create edited draft
    edited_draft = EmailDraft(
        to=to,
        subject=subject,
        body=body,
        cc=draft.cc,
        bcc=draft.bcc
    )
    
    # Final confirmation
    print("\nFinal Email:")
    print(json.dumps(edited_draft.model_dump(), indent=2))
    
    if input("\nSend? (yes/no): ").lower() == 'yes':
        send_email(edited_draft)
        return edited_draft
    else:
        raise ValueError("Email cancelled by user")

from enum import Enum
from datetime import datetime
from pydantic import BaseModel
import uuid

class ApprovalStatus(str, Enum):
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"
    EXPIRED = "expired"

class ApprovalRequest(BaseModel):
    request_id: str
    action: str
    context: dict
    status: ApprovalStatus = ApprovalStatus.PENDING
    created_at: datetime
    expires_at: datetime
    approved_by: str | None = None

class ApprovalQueue:
    """Manage async approval requests."""
    
    def __init__(self):
        self.requests = {}  # In production, use a database
    
    def create_request(
        self,
        action: str,
        context: dict,
        expires_in_hours: int = 24
    ) -> str:
        """Create a new approval request."""
        request_id = str(uuid.uuid4())
        
        request = ApprovalRequest(
            request_id=request_id,
            action=action,
            context=context,
            created_at=datetime.now(),
            expires_at=datetime.now() + timedelta(hours=expires_in_hours)
        )
        
        self.requests[request_id] = request
        
        # Notify approvers (email, Slack, etc.)
        notify_approvers(request)
        
        return request_id
    
    def get_request(self, request_id: str) -> ApprovalRequest:
        """Get approval request by ID."""
        if request_id not in self.requests:
            raise ValueError(f"Request {request_id} not found")
        
        request = self.requests[request_id]
        
        # Check expiration
        if datetime.now() > request.expires_at:
            request.status = ApprovalStatus.EXPIRED
        
        return request
    
    def approve(self, request_id: str, approver: str) -> ApprovalRequest:
        """Approve a request."""
        request = self.get_request(request_id)
        
        if request.status != ApprovalStatus.PENDING:
            raise ValueError(f"Request is {request.status}, cannot approve")
        
        request.status = ApprovalStatus.APPROVED
        request.approved_by = approver
        
        return request
    
    def reject(self, request_id: str, approver: str) -> ApprovalRequest:
        """Reject a request."""
        request = self.get_request(request_id)
        
        if request.status != ApprovalStatus.PENDING:
            raise ValueError(f"Request is {request.status}, cannot reject")
        
        request.status = ApprovalStatus.REJECTED
        request.approved_by = approver
        
        return request

# Usage in agent workflow
queue = ApprovalQueue()

async def agent_with_async_approval(query: str):
    # Generate recommendation
    result = agent.run(query)
    
    # Create approval request
    request_id = queue.create_request(
        action="Execute agent recommendation",
        context={"query": query, "result": result.content}
    )
    
    print(f"Approval request created: {request_id}")
    print("Waiting for approval...")
    
    # Poll for approval (in production, use webhooks/events)
    while True:
        request = queue.get_request(request_id)
        
        if request.status == ApprovalStatus.APPROVED:
            return execute_action(result)
        elif request.status == ApprovalStatus.REJECTED:
            return "Action rejected by approver"
        elif request.status == ApprovalStatus.EXPIRED:
            return "Approval request expired"
        
        await asyncio.sleep(5)  # Check every 5 seconds

# ✅ Good: Provide full context
def request_approval(action: str, details: dict):
    print(f"Action: {action}")
    print(f"Details: {json.dumps(details, indent=2)}")
    print(f"Risks: {assess_risks(details)}")
    print(f"Impact: {estimate_impact(details)}")
    return get_user_input()

# ❌ Bad: Vague approval request
def request_approval(action: str):
    return input(f"Approve {action}? ")  # No context

# ✅ Good: Expire old requests
request = ApprovalRequest(
    ...,
    expires_at=datetime.now() + timedelta(hours=24)
)

# ❌ Bad: No expiration (pending forever)
request = ApprovalRequest(...)  # No timeout

# ✅ Good: Allow editing before approval
edited = allow_user_edits(agent_output)
if approve(edited):
    execute(edited)

# ❌ Bad: Binary approve/reject only
if approve(agent_output):  # Can't modify
    execute(agent_output)

# ✅ Good: Log all approval decisions
class ApprovalLog:
    request_id: str
    action: str
    status: ApprovalStatus
    approver: str
    timestamp: datetime
    reason: str | None

def log_approval(request_id, status, approver, reason=None):
    log = ApprovalLog(
        request_id=request_id,
        status=status,
        approver=approver,
        timestamp=datetime.now(),
        reason=reason
    )
    save_to_database(log)

# ❌ Bad: No logging
def approve(request):
    request.status = "approved"  # No audit trail