Alexander Whitestone
b4d362fdad
Shows how to wire crisis detection into conversation loops. Refs #673
100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
"""
|
|
Crisis Integration Example — How to wire 988 Lifeline into the conversation loop.
|
|
|
|
This example shows how to integrate crisis detection into existing agent code.
|
|
"""
|
|
|
|
# Example 1: Simple integration in conversation loop
|
|
def conversation_loop_example():
|
|
"""Example of crisis integration in a conversation loop."""
|
|
from agent.crisis_middleware import check_crisis
|
|
|
|
while True:
|
|
user_message = input("User: ")
|
|
|
|
# Check for crisis FIRST
|
|
crisis_response = check_crisis(user_message)
|
|
if crisis_response:
|
|
print(f"Agent: {crisis_response}")
|
|
continue # Skip normal processing
|
|
|
|
# Normal agent processing
|
|
response = agent.run_conversation(user_message)
|
|
print(f"Agent: {response}")
|
|
|
|
|
|
# Example 2: Using the CrisisMiddleware class
|
|
def middleware_class_example():
|
|
"""Example using the CrisisMiddleware class directly."""
|
|
from agent.crisis_middleware import CrisisMiddleware
|
|
|
|
middleware = CrisisMiddleware(enabled=True)
|
|
|
|
def process_message(user_message: str) -> str:
|
|
# Check for crisis
|
|
crisis_response = middleware.check(user_message)
|
|
if crisis_response:
|
|
return crisis_response
|
|
|
|
# Normal processing
|
|
return agent.process(user_message)
|
|
|
|
|
|
# Example 3: Using the decorator
|
|
def decorator_example():
|
|
"""Example using the @crisis_aware decorator."""
|
|
from agent.crisis_middleware import crisis_aware
|
|
|
|
class MyAgent:
|
|
@crisis_aware
|
|
def process_message(self, user_message: str) -> str:
|
|
# This method is now crisis-aware
|
|
# If crisis is detected, the decorator returns the crisis response
|
|
# Otherwise, this code runs normally
|
|
return self.normal_processing(user_message)
|
|
|
|
|
|
# Example 4: Integration with run_agent.py style
|
|
def run_agent_integration():
|
|
"""
|
|
Example of integrating crisis check into run_agent.py style code.
|
|
|
|
Add this at the beginning of the conversation processing method:
|
|
"""
|
|
# In run_agent.py, add at the start of run_conversation() or similar:
|
|
#
|
|
# from agent.crisis_middleware import check_crisis
|
|
#
|
|
# def run_conversation(self, user_message: str):
|
|
# # Crisis check — must be first
|
|
# crisis_response = check_crisis(user_message)
|
|
# if crisis_response:
|
|
# return {"final_response": crisis_response, "crisis_detected": True}
|
|
#
|
|
# # ... rest of normal processing
|
|
pass
|
|
|
|
|
|
# Example 5: Gateway integration
|
|
def gateway_integration():
|
|
"""
|
|
Example for gateway/platform integration.
|
|
|
|
The gateway can check messages before sending to the agent:
|
|
"""
|
|
# In gateway/platforms/base.py or similar:
|
|
#
|
|
# from agent.crisis_middleware import check_crisis
|
|
#
|
|
# async def handle_message(self, message: str):
|
|
# # Check for crisis before agent processing
|
|
# crisis_response = check_crisis(message)
|
|
# if crisis_response:
|
|
# await self.send_message(crisis_response)
|
|
# return
|
|
#
|
|
# # Normal agent processing
|
|
# response = await self.agent.process(message)
|
|
# await self.send_message(response)
|
|
pass
|