A demonstration of how to integrate human feedback and decision-making into automated DSPy ReAct agent workflows. This system allows AI agents to pause execution and request human input when they need clarification, approval, or additional context.
It might be simplest to start with the blog post.
The included demo features a pizza ordering agent that:
- Takes natural language requests like "I want two large pizzas"
- Asks clarifying questions about toppings, sizes, and special instructions
- Builds structured order data through human interaction
- Works in both console and web interfaces
- Seamless Integration: Drop-in DSPy Tool that agents can use naturally
- Multiple Interfaces: Console (terminal) and web (browser) implementations
- Real-time Communication: Server-Sent Events (SSE) for live web updates
- Concurrent Support: Multiple browser tabs can connect simultaneously
- Type Safety: Full TypeScript-style typing with TypedDict
- Production Ready: Clean architecture with proper error handling
human_in_the_loop.py: Core infrastructure using asyncio for coordinationpizza_agent.py: DSPy signatures and types for the democonsole_app.py: Terminal-based interfaceweb_app.py: FastAPI web server with SSE streamingmain.py: Usage instructions and entry point
The system uses a Requester Pattern where different transport mechanisms (console, web) implement the same async interface:
# Create a human-input tool for any transport
tool = human_in_the_loop(requester_function)
# Use in DSPy ReAct agent
agent = dspy.ReAct(
signature=OrderPizza,
tools=[tool],
max_iters=6
)- Agent asks question - Creates
HumanInputRequest - Requester handles transport - Console input() or web queue
- Human provides response - Through terminal or browser
- Response delivered back - Agent continues execution
- Python 3.11+
- uv for dependency management
- OpenRouter API key (or other LLM provider)
git clone <repository-url>
cd dspy-react-hitl
uv syncSet your LLM provider credentials. The demo uses OpenRouter with Gemini 2.5 Flash:
export OPENROUTER_API_KEY="your-api-key-here"Or modify the LM configuration in the demo files:
lm = dspy.LM('openrouter/google/gemini-2.5-flash')
# or use other providers like:
# lm = dspy.LM('openai/gpt-4')
# lm = dspy.LM('anthropic/claude-3-sonnet-20240229')Interactive terminal-based interface:
uv run python console_app.pyExample interaction:
DSPy Human-in-the-Loop Pizza Agent (Console Version)
What is your order?
> I want a large pizza
Agent is thinking about: 'I want a large pizza'
What toppings would you like on your large pizza?
> pepperoni and mushrooms
Your order:
1. large pizza with pepperoni, mushrooms
Browser-based interface with real-time updates:
uv run python web_app.pyThen open http://localhost:8000 in your browser.
Features:
- Pizza favicon
- Real-time question/response flow
- Multiple browser tabs supported
- Mobile-friendly responsive design
- Activity logging
import dspy
from human_in_the_loop import human_in_the_loop, console_requester
# Create the tool
human_input = human_in_the_loop(console_requester)
# Use in any DSPy agent
agent = dspy.ReAct(
signature=YourSignature,
tools=[human_input, other_tools...],
max_iters=10
)
# Agent can now ask humans questions
result = await agent.aforward(your_input="...")Implement your own transport mechanism:
async def slack_requester(request: HumanInputRequest):
# Send question to Slack
await slack_client.send_message(request.question)
# Set up webhook handler to call:
# request.set_response(user_response)
async def email_requester(request: HumanInputRequest):
# Send email with question
# Set up email parser to call:
# request.set_response(user_response)Define your own agent signatures:
class CustomerSupport(dspy.Signature):
"""Agent that handles customer inquiries with human escalation"""
customer_message = dspy.InputField()
response: str = dspy.OutputField()
escalated: bool = dspy.OutputField()
agent = dspy.ReAct(
signature=CustomerSupport,
tools=[human_in_the_loop(your_requester)],
max_iters=8
)The system includes both unit-testable components and integration demos:
# Test console version
echo "medium pizza" | uv run python console_app.py
# Test web version
uv run python web_app.py
# Then visit http://localhost:8000- Current: In-memory queues, single-server
- Production: Consider Redis pub/sub, database persistence, load balancing
- Authentication: Add user authentication for web interface
- Rate Limiting: Prevent abuse of human input requests
- Input Validation: Sanitize all human responses
- Metrics: Track response times, abandonment rates
- Logging: Log all human interactions for audit
- Alerting: Monitor for stuck requests
The system uses asyncio.Future for coordination between agent execution and human response:
class HumanInputRequest:
def __init__(self, question: str):
self._response_future = asyncio.Future()
async def response(self) -> str:
return await self._response_future # Blocks until response
def set_response(self, response: str):
self._response_future.set_result(response) # Unblocks waiting agent- FastAPI: Modern async web framework
- Server-Sent Events: Real-time updates without WebSocket complexity
- Broadcast Pattern: Each browser tab gets its own event queue
- Graceful Cleanup: Proper handling of client disconnections
- Timeouts: Removed to allow unlimited human response time
- Network Issues: Automatic SSE reconnection
- Validation: Type-safe request/response handling
- Graceful Degradation: Fallback behaviors for edge cases
- Fork the repository
- Create a feature branch (
git checkout -b feature/amazing-feature) - Commit your changes (
git commit -m 'Add amazing feature') - Push to the branch (
git push origin feature/amazing-feature) - Open a Pull Request
This project is licensed under the MIT License - see the LICENSE file for details.