"""
Pattern 231: Conversational Agent
Purpose: Engage in natural multi-turn dialogue with context awareness and personality.
Maintains conversation history, understands context, and provides coherent responses.
"""
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import deque
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
class ConversationMode(Enum):
"""Modes of conversation."""
CASUAL = "casual" # Friendly, informal chat
PROFESSIONAL = "professional" # Formal, business-like
SUPPORTIVE = "supportive" # Empathetic, helpful
EDUCATIONAL = "educational" # Explanatory, teaching
@dataclass
class Message:
"""A single message in conversation."""
role: str # "user" or "assistant"
content: str
timestamp: datetime = field(default_factory=datetime.now)
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class ConversationTurn:
"""A complete turn in the conversation."""
turn_number: int
user_message: str
agent_response: str
context_used: List[str]
intent: Optional[str] = None
sentiment: Optional[str] = None
timestamp: datetime = field(default_factory=datetime.now)
@dataclass
class ConversationContext:
"""Context information for conversation."""
user_name: Optional[str] = None
user_preferences: Dict[str, Any] = field(default_factory=dict)
topics_discussed: List[str] = field(default_factory=list)
current_topic: Optional[str] = None
conversation_goal: Optional[str] = None
class ConversationalAgent:
"""
Conversational Agent - Natural multi-turn dialogue with context awareness.
Engages in coherent conversations by maintaining history, understanding
context, adapting tone, and providing relevant responses.
"""
def __init__(self,
name: str,
personality: str = "helpful and friendly",
mode: ConversationMode = ConversationMode.CASUAL,
max_history: int = 20):
self.name = name
self.personality = personality
self.mode = mode
self.max_history = max_history
self.llm = ChatOpenAI(model="gpt-4", temperature=0.8)
# Conversation state
self.conversation_history: deque = deque(maxlen=max_history)
self.turns: List[ConversationTurn] = []
self.context = ConversationContext()
self.current_turn = 0
# Statistics
self.statistics: Dict[str, Any] = {
"total_turns": 0,
"total_messages": 0,
"topics_covered": 0,
"avg_response_length": 0.0
}
def converse(self, user_input: str) -> str:
"""Process user input and generate response."""
self.current_turn += 1
# Add user message to history
user_msg = Message(role="user", content=user_input)
self.conversation_history.append(user_msg)
# Analyze input
intent = self._detect_intent(user_input)
sentiment = self._detect_sentiment(user_input)
# Extract context
context_items = self._extract_context(user_input)
# Generate response with context
response = self._generate_response(user_input, context_items)
# Add assistant message to history
agent_msg = Message(role="assistant", content=response)
self.conversation_history.append(agent_msg)
# Record turn
turn = ConversationTurn(
turn_number=self.current_turn,
user_message=user_input,
agent_response=response,
context_used=context_items,
intent=intent,
sentiment=sentiment
)
self.turns.append(turn)
# Update statistics
self.statistics["total_turns"] = self.current_turn
self.statistics["total_messages"] = len(self.conversation_history)
self._update_statistics()
return response
def _detect_intent(self, user_input: str) -> str:
"""Detect user intent from input."""
user_lower = user_input.lower()
if any(word in user_lower for word in ["?", "what", "how", "why", "when", "where"]):
return "question"
elif any(word in user_lower for word in ["please", "can you", "could you", "would you"]):
return "request"
elif any(word in user_lower for word in ["thank", "thanks", "appreciate"]):
return "gratitude"
elif any(word in user_lower for word in ["hello", "hi", "hey", "greetings"]):
return "greeting"
elif any(word in user_lower for word in ["bye", "goodbye", "see you"]):
return "farewell"
else:
return "statement"
def _detect_sentiment(self, user_input: str) -> str:
"""Detect sentiment of user input."""
user_lower = user_input.lower()
positive_words = ["good", "great", "excellent", "happy", "love", "wonderful"]
negative_words = ["bad", "terrible", "hate", "sad", "angry", "frustrated"]
positive_count = sum(1 for word in positive_words if word in user_lower)
negative_count = sum(1 for word in negative_words if word in user_lower)
if positive_count > negative_count:
return "positive"
elif negative_count > positive_count:
return "negative"
else:
return "neutral"
def _extract_context(self, user_input: str) -> List[str]:
"""Extract relevant context from conversation history."""
context_items = []
# Add recent conversation context
recent_messages = list(self.conversation_history)[-4:] # Last 2 turns
if len(recent_messages) > 1:
context_items.append(f"Recent context: {recent_messages[-2].content[:50]}...")
# Add current topic if set
if self.context.current_topic:
context_items.append(f"Current topic: {self.context.current_topic}")
# Add user preferences if relevant
if self.context.user_preferences:
context_items.append(f"User preferences: {list(self.context.user_preferences.keys())[:2]}")
return context_items
def _generate_response(self, user_input: str, context_items: List[str]) -> str:
"""Generate contextually appropriate response."""
# Build conversation history for LLM
messages = []
# System message with personality and mode
system_prompt = f"""You are {self.name}, a {self.personality} conversational agent.
Conversation mode: {self.mode.value}.
Maintain context awareness and provide coherent, relevant responses.
Be natural, engaging, and adapt to the user's tone."""
messages.append(SystemMessage(content=system_prompt))
# Add recent conversation history
for msg in list(self.conversation_history)[-6:]: # Last 3 turns
if msg.role == "user":
messages.append(HumanMessage(content=msg.content))
else:
messages.append(AIMessage(content=msg.content))
# Add current user message if not already in history
if not self.conversation_history or self.conversation_history[-1].content != user_input:
messages.append(HumanMessage(content=user_input))
# Generate response
response = self.llm.invoke(messages)
return response.content if hasattr(response, 'content') else str(response)
def set_context(self, key: str, value: Any):
"""Set context information."""
if key == "user_name":
self.context.user_name = value
elif key == "topic":
self.context.current_topic = value
if value not in self.context.topics_discussed:
self.context.topics_discussed.append(value)
elif key == "goal":
self.context.conversation_goal = value
else:
self.context.user_preferences[key] = value
print(f" Context updated: {key} = {value}")
def switch_mode(self, mode: ConversationMode):
"""Switch conversation mode."""
old_mode = self.mode
self.mode = mode
print(f" Switched from {old_mode.value} to {mode.value} mode")
def get_conversation_summary(self) -> Dict[str, Any]:
"""Get summary of conversation."""
if not self.turns:
return {"total_turns": 0}
intents = {}
sentiments = {}
for turn in self.turns:
intents[turn.intent] = intents.get(turn.intent, 0) + 1
sentiments[turn.sentiment] = sentiments.get(turn.sentiment, 0) + 1
return {
"total_turns": len(self.turns),
"topics_discussed": self.context.topics_discussed,
"intent_distribution": intents,
"sentiment_distribution": sentiments,
"current_topic": self.context.current_topic,
"conversation_mode": self.mode.value
}
def _update_statistics(self):
"""Update conversation statistics."""
if self.turns:
total_length = sum(len(turn.agent_response) for turn in self.turns)
self.statistics["avg_response_length"] = total_length / len(self.turns)
self.statistics["topics_covered"] = len(self.context.topics_discussed)
def get_statistics(self) -> Dict[str, Any]:
"""Get comprehensive statistics."""
return {
"name": self.name,
"personality": self.personality,
"mode": self.mode.value,
**self.statistics
}
def reset_conversation(self):
"""Reset conversation state."""
self.conversation_history.clear()
self.turns.clear()
self.context = ConversationContext()
self.current_turn = 0
print(f" Conversation reset for {self.name}")
class ConversationalPattern:
"""
Conversational Pattern - Manage natural dialogue agents.
Coordinates conversational agents for multi-turn interactions.
"""
def __init__(self):
self.agents: Dict[str, ConversationalAgent] = {}
def register_agent(self, agent: ConversationalAgent):
"""Register conversational agent."""
self.agents[agent.name] = agent
print(f" Registered agent: {agent.name}")
def run(self, agent_name: str, conversation: List[str]) -> Dict[str, Any]:
"""Run conversation with agent."""
if agent_name not in self.agents:
return {"error": f"Agent {agent_name} not found"}
agent = self.agents[agent_name]
responses = []
print(f"\n=== Conversation with {agent_name} ===\n")
for user_input in conversation:
print(f"User: {user_input}")
response = agent.converse(user_input)
print(f"{agent_name}: {response}\n")
responses.append(response)
summary = agent.get_conversation_summary()
return {
"agent": agent_name,
"turns": len(conversation),
"responses": responses,
"summary": summary
}
# Example usage
if __name__ == "__main__":
print("=== Conversational Agent Pattern Demo ===\n")
# Create pattern
pattern = ConversationalPattern()
# Create conversational agents with different personalities
print("1. Creating conversational agents...")
friendly_agent = ConversationalAgent(
name="FriendlyBot",
personality="warm, empathetic, and encouraging",
mode=ConversationMode.SUPPORTIVE
)
pattern.register_agent(friendly_agent)
professional_agent = ConversationalAgent(
name="ProfessionalBot",
personality="professional, concise, and efficient",
mode=ConversationMode.PROFESSIONAL
)
pattern.register_agent(professional_agent)
# Set context
print("\n2. Setting context...")
friendly_agent.set_context("user_name", "Alex")
friendly_agent.set_context("topic", "productivity")
# Run conversations
print("\n3. Running conversations...")
friendly_conversation = [
"Hi! I'm feeling a bit overwhelmed with work.",
"I have so many tasks and don't know where to start.",
"That's helpful! What about prioritizing them?",
"Thanks for the advice!"
]
result1 = pattern.run("FriendlyBot", friendly_conversation)
professional_conversation = [
"I need help organizing my project timeline.",
"The deadline is in 3 weeks.",
"What's the best approach for risk management?"
]
result2 = pattern.run("ProfessionalBot", professional_conversation)
# Print summaries
print("\n4. Conversation Summaries:")
for agent_name, agent in pattern.agents.items():
summary = agent.get_conversation_summary()
print(f"\n {agent_name}:")
print(f" Total turns: {summary['total_turns']}")
print(f" Topics: {summary.get('topics_discussed', [])}")
print(f" Intents: {summary.get('intent_distribution', {})}")
print(f" Sentiments: {summary.get('sentiment_distribution', {})}")
# Statistics
print("\n5. Agent Statistics:")
for agent_name, agent in pattern.agents.items():
stats = agent.get_statistics()
print(f"\n {agent_name}:")
for key, value in stats.items():
if isinstance(value, float):
print(f" {key}: {value:.2f}")
else:
print(f" {key}: {value}")
print("\nā Conversational agents engaged in natural dialogue!")
Agentic Architecture Pattern: Conversational Agent
Pattern 231: 231 Conversational Agent – Line-by-Line Explanation
Overview
Pattern 231: Conversational Agent
Purpose: Engage in natural multi-turn dialogue with context awareness and personality. Maintains conversation history, understands context, and provides coherent responses.
Code Structure Analysis
File Information
- Pattern Number: 231
- Pattern Name: 231 Conversational Agent
- File Path:
patterns\domain_specific\231_conversational_agent.py - Total Lines: 389
Detailed Code Explanation
Module Docstring
"""
Pattern 231: Conversational Agent
Purpose: Engage in natural multi-turn dialogue with context awareness and personality.
Maintains conversation history, understands context, and provides coherent responses.
"""
Explanation: This module docstring describes the purpose and key features of the 231 Conversational Agent pattern.
Imports Section
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import deque
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage # First 20 imports shown
Explanation: The module imports necessary dependencies:
– from typing import Dict, Any, List, Optional: 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
– from langchain_core.messages import HumanMessage, AIMessage, SystemMessage: LangChain framework integration
Enumerations
ConversationMode (Line 18)
class ConversationMode(Enum):
CASUAL
PROFESSIONAL
SUPPORTIVE
EDUCATIONAL
Purpose: Modes of conversation.
Values:
– CASUAL: Represents casual
– PROFESSIONAL: Represents professional
– SUPPORTIVE: Represents supportive
– EDUCATIONAL: Represents educational
Classes
ConversationMode (Line 18)
Description: Modes of conversation.
Message (Line 27)
Description: A single message in conversation.
ConversationTurn (Line 36)
Description: A complete turn in the conversation.
ConversationContext (Line 48)
Description: Context information for conversation.
ConversationalAgent (Line 57)
Description: Conversational Agent – Natural multi-turn dialogue with context awareness.
Engages in coherent conversations by maintaining history, understanding context, adapting tone, and providing relevant responses.
Methods:
__init__ (Line 65)
def __init__(self, name, personality, mode, max_history):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– name: Identifier or label
– personality: Parameter for personality
– mode: Parameter for mode
– max_history: Parameter for max_history
converse (Line 90)
def converse(self, user_input):
"""Process user input and generate response."""
Purpose: Process user input and generate response.
Parameters:
– self: Instance reference
– user_input: Parameter for user_input
_detect_intent (Line 130)
def _detect_intent(self, user_input):
"""Detect user intent from input."""
Purpose: Detect user intent from input.
Parameters:
– self: Instance reference
– user_input: Parameter for user_input
_detect_sentiment (Line 147)
def _detect_sentiment(self, user_input):
"""Detect sentiment of user input."""
Purpose: Detect sentiment of user input.
Parameters:
– self: Instance reference
– user_input: Parameter for user_input
_extract_context (Line 164)
def _extract_context(self, user_input):
"""Extract relevant context from conversation history."""
Purpose: Extract relevant context from conversation history.
Parameters:
– self: Instance reference
– user_input: Parameter for user_input
_generate_response (Line 183)
def _generate_response(self, user_input, context_items):
"""Generate contextually appropriate response."""
Purpose: Generate contextually appropriate response.
Parameters:
– self: Instance reference
– user_input: Parameter for user_input
– context_items: Parameter for context_items
set_context (Line 211)
def set_context(self, key, value):
"""Set context information."""
Purpose: Set context information.
Parameters:
– self: Instance reference
– key: Parameter for key
– value: Parameter for value
switch_mode (Line 226)
def switch_mode(self, mode):
"""Switch conversation mode."""
Purpose: Switch conversation mode.
Parameters:
– self: Instance reference
– mode: Parameter for mode
get_conversation_summary (Line 232)
def get_conversation_summary(self):
"""Get summary of conversation."""
Purpose: Get summary of conversation.
Parameters:
– self: Instance reference
_update_statistics (Line 252)
def _update_statistics(self):
"""Update conversation statistics."""
Purpose: Update conversation statistics.
Parameters:
– self: Instance reference
get_statistics (Line 259)
def get_statistics(self):
"""Get comprehensive statistics."""
Purpose: Get comprehensive statistics.
Parameters:
– self: Instance reference
reset_conversation (Line 268)
def reset_conversation(self):
"""Reset conversation state."""
Purpose: Reset conversation state.
Parameters:
– self: Instance reference
ConversationalPattern (Line 277)
Description: Conversational Pattern – Manage natural dialogue agents.
Coordinates conversational agents for multi-turn interactions.
Methods:
__init__ (Line 284)
def __init__(self):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
register_agent (Line 287)
def register_agent(self, agent):
"""Register conversational agent."""
Purpose: Register conversational agent.
Parameters:
– self: Instance reference
– agent: Parameter for agent
run (Line 292)
def run(self, agent_name, conversation):
"""Run conversation with agent."""
Purpose: Run conversation with agent.
Parameters:
– self: Instance reference
– agent_name: Parameter for agent_name
– conversation: Parameter for conversation
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 = ConversationMode(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
Leave a Reply