"""
Pattern 154: Human-in-the-Loop Pattern
Incorporates human feedback and decisions into agent workflows.
Implements interactive decision points with LangGraph state management.
"""
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, Any, List, Optional, Callable
from enum import Enum
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict
class ApprovalStatus(Enum):
"""Status of human approval."""
PENDING = "pending"
APPROVED = "approved"
REJECTED = "rejected"
MODIFIED = "modified"
class DecisionType(Enum):
"""Type of human decision required."""
APPROVAL = "approval"
SELECTION = "selection"
CORRECTION = "correction"
FEEDBACK = "feedback"
@dataclass
class HumanDecisionPoint:
"""Represents a point requiring human input."""
decision_id: str
decision_type: DecisionType
question: str
options: List[str] = field(default_factory=list)
context: Dict[str, Any] = field(default_factory=dict)
ai_recommendation: Optional[str] = None
human_response: Optional[str] = None
status: ApprovalStatus = ApprovalStatus.PENDING
timestamp: datetime = field(default_factory=datetime.now)
@dataclass
class WorkflowStep:
"""Step in the workflow."""
step_id: str
name: str
requires_human: bool = False
decision_type: Optional[DecisionType] = None
completed: bool = False
result: Optional[Any] = None
class HumanInLoopAgent:
"""
Agent that integrates human decisions into its workflow.
"""
def __init__(
self,
name: str,
model_name: str = "gpt-4"
):
self.name = name
self.llm = ChatOpenAI(model=model_name, temperature=0.7)
self.decision_points: List[HumanDecisionPoint] = []
self.workflow_steps: List[WorkflowStep] = []
def request_human_decision(
self,
decision_type: DecisionType,
question: str,
options: List[str] = None,
context: Dict[str, Any] = None,
provide_ai_recommendation: bool = True
) -> HumanDecisionPoint:
"""Request a decision from human."""
decision_id = f"decision-{len(self.decision_points) + 1}"
# Generate AI recommendation
ai_recommendation = None
if provide_ai_recommendation:
ai_recommendation = self._generate_recommendation(question, options, context)
decision = HumanDecisionPoint(
decision_id=decision_id,
decision_type=decision_type,
question=question,
options=options or [],
context=context or {},
ai_recommendation=ai_recommendation
)
self.decision_points.append(decision)
return decision
def _generate_recommendation(
self,
question: str,
options: List[str],
context: Dict[str, Any]
) -> str:
"""Generate AI recommendation for human."""
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {self.name}, an AI assistant. Provide a recommendation for the human."),
("human", """Question: {question}
Options:
{options}
Context:
{context}
Provide your recommendation and brief reasoning:""")
])
chain = prompt | self.llm
response = chain.invoke({
"question": question,
"options": "\n".join([f"- {opt}" for opt in options]) if options else "No specific options",
"context": "\n".join([f"{k}: {v}" for k, v in (context or {}).items()])
})
return str(response.content)
def receive_human_feedback(
self,
decision_id: str,
response: str,
status: ApprovalStatus = ApprovalStatus.APPROVED
):
"""Receive human feedback for a decision."""
for decision in self.decision_points:
if decision.decision_id == decision_id:
decision.human_response = response
decision.status = status
break
def execute_with_human_oversight(
self,
task: str,
steps: List[WorkflowStep],
human_input_func: Callable[[HumanDecisionPoint], str]
) -> Dict[str, Any]:
"""Execute workflow with human oversight."""
self.workflow_steps = steps
results = []
for step in steps:
if step.requires_human:
# Request human decision
decision = self.request_human_decision(
decision_type=step.decision_type or DecisionType.APPROVAL,
question=f"Approve step: {step.name}?",
context={"step": step.name, "task": task}
)
# Get human input (simulated)
human_response = human_input_func(decision)
# Process response
self.receive_human_feedback(
decision.decision_id,
human_response,
ApprovalStatus.APPROVED if "yes" in human_response.lower() else ApprovalStatus.REJECTED
)
if decision.status == ApprovalStatus.APPROVED:
result = self._execute_step(step, task)
results.append(result)
else:
results.append({
"step": step.name,
"status": "rejected",
"reason": human_response
})
break
else:
# Execute without human input
result = self._execute_step(step, task)
results.append(result)
return {
"task": task,
"results": results,
"decisions_count": len(self.decision_points),
"steps_completed": len([r for r in results if r.get("status") != "rejected"])
}
def _execute_step(self, step: WorkflowStep, task: str) -> Dict[str, Any]:
"""Execute a workflow step."""
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {self.name}. Execute the workflow step."),
("human", "Task: {task}\nStep: {step_name}\n\nExecute this step and provide the result:")
])
chain = prompt | self.llm
response = chain.invoke({
"task": task,
"step_name": step.name
})
step.completed = True
step.result = str(response.content)
return {
"step": step.name,
"status": "completed",
"result": step.result
}
def get_decision_summary(self) -> Dict[str, Any]:
"""Get summary of human decisions."""
if not self.decision_points:
return {"total": 0}
return {
"total": len(self.decision_points),
"approved": len([d for d in self.decision_points if d.status == ApprovalStatus.APPROVED]),
"rejected": len([d for d in self.decision_points if d.status == ApprovalStatus.REJECTED]),
"pending": len([d for d in self.decision_points if d.status == ApprovalStatus.PENDING]),
"decisions": [
{
"id": d.decision_id,
"type": d.decision_type.value,
"question": d.question,
"ai_recommendation": d.ai_recommendation,
"human_response": d.human_response,
"status": d.status.value
}
for d in self.decision_points
]
}
class HumanInLoopPattern:
"""
Complete Human-in-the-Loop pattern with LangGraph workflow.
"""
def __init__(self, agent: HumanInLoopAgent):
self.agent = agent
self.workflow = self._create_workflow()
def _create_workflow(self) -> StateGraph:
"""Create workflow with human decision points."""
class HumanLoopState(TypedDict):
task: str
current_step: int
steps: List[Dict[str, Any]]
results: List[Dict[str, Any]]
human_decisions: List[Dict[str, Any]]
workflow = StateGraph(HumanLoopState)
def execute_step(state: HumanLoopState) -> HumanLoopState:
"""Execute current step."""
if state["current_step"] str:
"""Check if workflow should continue."""
return "execute" if state["current_step"] Dict[str, Any]:
"""Run workflow with human-in-the-loop."""
return self.agent.execute_with_human_oversight(task, steps, human_input_func)
def main():
"""Example usage of Human-in-the-Loop Pattern."""
print("=" * 80)
print("Human-in-the-Loop Pattern Demo")
print("=" * 80)
# Create agent
agent = HumanInLoopAgent(name="ContentAgent")
pattern = HumanInLoopPattern(agent)
# Define workflow steps
steps = [
WorkflowStep(
step_id="step-1",
name="Generate content outline",
requires_human=False
),
WorkflowStep(
step_id="step-2",
name="Review and approve outline",
requires_human=True,
decision_type=DecisionType.APPROVAL
),
WorkflowStep(
step_id="step-3",
name="Write full content",
requires_human=False
),
WorkflowStep(
step_id="step-4",
name="Review and approve final content",
requires_human=True,
decision_type=DecisionType.APPROVAL
),
WorkflowStep(
step_id="step-5",
name="Publish content",
requires_human=False
)
]
# Simulate human input function
def simulated_human_input(decision: HumanDecisionPoint) -> str:
"""Simulate human providing input."""
print(f"\n{'=' * 80}")
print(f"🤔 Human Decision Required")
print(f"{'=' * 80}")
print(f"Question: {decision.question}")
print(f"Type: {decision.decision_type.value}")
if decision.ai_recommendation:
print(f"\n💡 AI Recommendation:")
print(f"{decision.ai_recommendation}")
# Simulate approval (in real scenario, get actual input)
import random
response = "yes, approved" if random.random() > 0.2 else "no, needs revision"
print(f"\n✅ Human Response: {response}")
return response
# Run workflow
task = "Create blog post about AI agents"
print(f"\n📝 Task: {task}")
print(f"Steps: {len(steps)}")
print("\n" + "=" * 80)
print("🔄 Executing Workflow with Human Oversight")
print("=" * 80)
result = pattern.run(task, steps, simulated_human_input)
# Display results
print("\n" + "=" * 80)
print("✅ Workflow Results")
print("=" * 80)
print(f"Task: {result['task']}")
print(f"Steps Completed: {result['steps_completed']}/{len(steps)}")
print(f"Human Decisions: {result['decisions_count']}")
print("\nStep Results:")
for i, res in enumerate(result["results"], 1):
print(f"\n{i}. {res['step']}")
print(f" Status: {res['status']}")
if res.get('result'):
print(f" Result: {str(res['result'])[:100]}...")
# Display decision summary
print("\n" + "=" * 80)
print("📊 Decision Summary")
print("=" * 80)
summary = agent.get_decision_summary()
print(f"Total Decisions: {summary['total']}")
print(f"Approved: {summary['approved']}")
print(f"Rejected: {summary['rejected']}")
print(f"Pending: {summary['pending']}")
print("\nDecision Details:")
for decision in summary["decisions"]:
print(f"\n{decision['id']} ({decision['type']})")
print(f" Question: {decision['question']}")
print(f" Status: {decision['status']}")
if decision['human_response']:
print(f" Response: {decision['human_response']}")
if __name__ == "__main__":
main()
Agentic Architecture Pattern: Human In The Loop
Pattern 154: 154 Human In The Loop Pattern – Line-by-Line Explanation
Overview
Pattern 154: Human-in-the-Loop Pattern
Incorporate human feedback and oversight at critical decision points in the agent workflow. Humans can approve, reject, or modify agent actions.
Code Structure Analysis
File Information
- Pattern Number: 154
- Pattern Name: 154 Human In The Loop Pattern
- File Path:
patterns\hybrid_ai\154_human_in_the_loop_pattern.py - Total Lines: 242
Detailed Code Explanation
Module Docstring
"""
Pattern 154: Human-in-the-Loop Pattern
Incorporate human feedback and oversight at critical decision points in the
agent workflow. Humans can approve, reject, or modify agent actions.
"""
Explanation: This module docstring describes the purpose and key features of the 154 Human In The Loop Pattern pattern.
Imports Section
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI # First 20 imports shown
Explanation: The module imports necessary dependencies:
– from typing import Dict, Any, List, Optional, Callable: Type hints for better code documentation
– from dataclasses import dataclass, field: Data class decorators for clean data structures
– from datetime import datetime: Date and time utilities
– from enum import Enum: Enumeration support for fixed sets of values
– from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration
– from langchain_openai import ChatOpenAI: LangChain framework integration
Enumerations
ApprovalStatus (Line 16)
class ApprovalStatus(Enum):
PENDING
APPROVED
REJECTED
MODIFIED
Purpose: Status of human approval.
Values:
– PENDING: Initial state, not yet processed
– APPROVED: Represents approved
– REJECTED: Represents rejected
– MODIFIED: Represents modified
Classes
ApprovalStatus (Line 16)
Description: Status of human approval.
HumanFeedback (Line 25)
Description: Human feedback on agent action.
AgentAction (Line 35)
Description: Action proposed by agent.
HumanInTheLoopPattern (Line 45)
Description: Human-in-the-Loop Pattern – Human oversight of agent actions.
Methods:
__init__ (Line 50)
def __init__(self, model_name, confidence_threshold):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– model_name: LLM model to use (e.g., “gpt-4”)
– confidence_threshold: Parameter for confidence_threshold
set_feedback_function (Line 57)
def set_feedback_function(self, func):
"""Set the function that gets human feedback (simulated or real)."""
Purpose: Set the function that gets human feedback (simulated or real).
Parameters:
– self: Instance reference
– func: Parameter for func
propose_action (Line 61)
def propose_action(self, description, proposed_action, confidence):
"""Agent proposes an action."""
Purpose: Agent proposes an action.
Parameters:
– self: Instance reference
– description: Parameter for description
– proposed_action: Parameter for proposed_action
– confidence: Parameter for confidence
get_human_feedback (Line 79)
def get_human_feedback(self, action):
"""Get human feedback on an action."""
Purpose: Get human feedback on an action.
Parameters:
– self: Instance reference
– action: Parameter for action
execute_with_approval (Line 113)
def execute_with_approval(self, action):
"""Execute action with human approval if needed."""
Purpose: Execute action with human approval if needed.
Parameters:
– self: Instance reference
– action: Parameter for action
run (Line 149)
def run(self, task):
"""Execute task with human-in-the-loop."""
Purpose: Execute task with human-in-the-loop.
Parameters:
– self: Instance reference
– task: Parameter for task
get_statistics (Line 183)
def get_statistics(self):
"""Get HITL statistics."""
Purpose: Get HITL statistics.
Parameters:
– self: Instance reference
Design Patterns Used
- Strategy Pattern: Different execution strategies for various scenarios
- State Pattern: Managing different states in the workflow
- Chain of Responsibility: Sequential processing through workflow nodes
- Observer Pattern: Tracking and monitoring pattern execution
- Factory Pattern: Creating instances with proper configuration
Usage Example
# Initialize the pattern
pattern = ApprovalStatus(name="example", model_name="gpt-4")
# Configure and execute
result = pattern.execute(data={"key": "value"})
# Check results
print(result)
Key Concepts
- LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
- Workflow Orchestration: LangGraph manages execution flow through nodes
- State Management: TypedDict defines workflow state structure
- Error Handling: Comprehensive error handling and fallback mechanisms
- Extensibility: Designed for easy customization and extension
Extension Points
This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities