"""
Basic Chat Bot Example
간단한 챗봇 - 대화 기록 관리 기능 포함
This example shows:
- BaseNode implementation
- NodeContext and WorkflowContext usage
- ChatHistoryNode for memory management
- InMemoryChatHistory for session storage
- Message handling (user/assistant)
"""
import asyncio
from datetime import datetime
from nadoo_flow import (
BaseNode, NodeResult, NodeContext, WorkflowContext,
ChatHistoryNode, SessionHistoryManager,
InMemoryChatHistory, Message
)
class SimpleChatBot(BaseNode, ChatHistoryNode):
"""Simple chatbot with memory"""
def __init__(self, name="ChatBot"):
BaseNode.__init__(self, node_id="chatbot")
# Setup session-based history manager
history_manager = SessionHistoryManager(
history_factory=lambda session_id: InMemoryChatHistory(),
window_size=10 # Keep only last 10 messages
)
ChatHistoryNode.__init__(
self,
history_manager=history_manager,
session_key="session_id" # Key to extract session ID from context
)
self.name = name
async def execute(self, node_context: NodeContext, workflow_context: WorkflowContext) -> NodeResult:
"""Process user input and generate response"""
# Get user input
user_input = node_context.input_data.get("message", "")
# Get conversation history for this session
history = await self.get_history(workflow_context)
messages = await history.get_messages()
# Generate response based on input and history
response = await self.generate_response(user_input, messages)
# Save interaction to history
await self.save_interaction(
workflow_context,
Message.user(user_input),
Message.assistant(response)
)
return NodeResult(
success=True,
output={
"response": response,
"message_count": len(messages) + 2,
"session_id": workflow_context.get_global_variable("session_id")
}
)
async def generate_response(self, user_input: str, history: list[Message]) -> str:
"""Generate bot response based on input and history"""
# Simple rule-based responses
user_input_lower = user_input.lower()
if "hello" in user_input_lower or "hi" in user_input_lower:
return f"Hello! I'm {self.name}. How can I help you today?"
elif "bye" in user_input_lower or "goodbye" in user_input_lower:
return "Goodbye! It was nice chatting with you!"
elif "history" in user_input_lower:
if not history:
return "This is the beginning of our conversation."
else:
return f"We've exchanged {len(history)} messages so far."
elif "time" in user_input_lower:
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
return f"The current time is {current_time}"
elif "name" in user_input_lower:
# Check if user mentioned their name before
for msg in history:
if msg.role == "user" and "my name is" in msg.content.lower():
# Extract name from previous message
parts = msg.content.lower().split("my name is")
if len(parts) > 1:
name = parts[1].strip().split()[0]
return f"I remember! Your name is {name.title()}."
return "What's your name? You can tell me by saying 'My name is...'"
else:
# Echo with acknowledgment
return f"You said: '{user_input}'. Tell me more!"
async def main():
"""Main execution function"""
print("🤖 Simple Chat Bot Example")
print("=" * 50)
# Create chatbot
chatbot = SimpleChatBot(name="Assistant")
# Create workflow context with session
workflow_context = WorkflowContext(
workflow_id="chat_session_001",
global_variables={"session_id": "user_123"}
)
# Simulate conversation
conversations = [
"Hello!",
"What's the time?",
"My name is Alice",
"Do you remember my name?",
"Show me our history",
"Goodbye!"
]
print(f"Starting conversation with session: {workflow_context.global_variables['session_id']}\n")
for user_message in conversations:
print(f"👤 User: {user_message}")
# Create node context with user input
node_context = NodeContext(
node_id="chatbot",
input_data={"message": user_message}
)
# Execute chatbot
result = await chatbot.execute(node_context, workflow_context)
if result.success:
print(f"🤖 Bot: {result.output['response']}")
print(f" (Total messages: {result.output['message_count']})\n")
else:
print(f"❌ Error: {result.error}\n")
# Small delay for readability
await asyncio.sleep(0.5)
# Display final conversation history
print("\n" + "=" * 50)
print("📜 Conversation History:")
history = await chatbot.get_history(workflow_context)
messages = await history.get_messages()
for i, msg in enumerate(messages, 1):
role_emoji = "👤" if msg.role == "user" else "🤖"
print(f"{i}. {role_emoji} {msg.role.title()}: {msg.content}")
if __name__ == "__main__":
asyncio.run(main())
