Skip to content

Conversations

The conversations module provides high-level APIs for managing conversation state, message history, and metadata in the Agency framework.

Components

ConversationManager

Central API for conversation lifecycle management:

from agency.conversations.manager import ConversationManager

manager = ConversationManager(
    base_dir="data/conversations",
    session_dir="data/streamlit_sessions"
)

Core Models

ChatMessage

Represents individual messages in conversations:

from agency.conversations.models import ChatMessage

message = ChatMessage(
    role="user",
    content="Hello, how can you help me?",
    timestamp="2:30 PM",
    metadata={"user_id": "123"}
)

Conversation

Container for conversation metadata and messages:

from agency.conversations.models import Conversation

conversation = Conversation(
    id="conv_abc123",
    name="Customer Support Chat",
    created_at="2024-09-05T10:00:00Z",
    session_info=session_info,
    config=agency_config
)

SessionInfo

Tracks session identifiers across systems:

from agency.conversations.models import SessionInfo

session_info = SessionInfo(
    session_id="session_xyz",
    user_id="user_123",
    openai_session_id="conv_abc123"
)

Usage Examples

Creating a New Conversation

# Initialize manager
manager = ConversationManager()

# Create conversation with custom config
conversation = manager.create_conversation(
    name="Technical Support",
    config=AgencyConfig(
        name="TechSupport",
        strategy_type="router",
        agents=["technical", "general"]
    )
)

print(f"Created conversation: {conversation.id}")

Adding Messages

# Add user message
user_message = manager.add_message(
    conversation_id=conversation.id,
    role="user",
    content="I'm having trouble with my account",
    metadata={"priority": "high"}
)

# Process and get response
response = manager.process_message(
    conversation_id=conversation.id,
    message="I'm having trouble with my account"
)

Loading Existing Conversations

# Load conversation by ID
conversation = manager.load_conversation("conv_abc123")

# Get conversation history
messages = manager.get_messages(conversation.id)

# Get conversation summaries
summaries = manager.list_conversations()
for summary in summaries:
    print(f"{summary.id}: {summary.name} ({summary.message_count} messages)")

# Search conversations by content
results = manager.search_conversations("billing issue")

# Filter by date range
from datetime import datetime, timedelta

recent = manager.list_conversations(
    since=datetime.now() - timedelta(days=7)
)

Storage Layer

ConversationStorage

Handles file-based persistence:

from agency.conversations.storage import ConversationStorage

storage = ConversationStorage("data/conversations")

# Save conversation
storage.save_conversation(conversation)

# Load conversation
loaded = storage.load_conversation(conversation.id)

# List all conversations
all_conversations = storage.list_conversations()

File Structure

Conversations are stored as JSON files:

data/conversations/
├── conv_abc123.json
├── conv_def456.json
└── summaries.json

Integration with Agency

Using with AgencyFactory

from agency.factories.agency_factory import AgencyFactory

# Create conversation
manager = ConversationManager()
conversation = manager.create_conversation("Support Chat")

# Create agency with conversation context
agency = AgencyFactory.create_agency(
    config=conversation.config,
    session_id=conversation.session_info.session_id,
    user_id=conversation.session_info.user_id
)

# Process message
result = await agency.run("How can I help you?")

# Save response
manager.add_message(
    conversation.id,
    "assistant",
    result.response
)

Streamlit Integration

The conversation manager integrates seamlessly with Streamlit applications:

import streamlit as st
from agency.conversations.manager import ConversationManager

# Initialize in session state
if 'conversation_manager' not in st.session_state:
    st.session_state.conversation_manager = ConversationManager()

# Create or load conversation
if 'current_conversation' not in st.session_state:
    st.session_state.current_conversation = (
        st.session_state.conversation_manager.create_conversation()
    )

Configuration Options

Custom Storage Paths

manager = ConversationManager(
    base_dir="/custom/path/conversations",
    session_dir="/custom/path/sessions"
)

Agency Configuration

from agency.conversations.models import AgencyConfig

config = AgencyConfig(
    name="CustomerSupport",
    strategy_type="router",
    agents=["billing", "technical", "general"],
    settings={
        "max_tokens": 4000,
        "temperature": 0.7
    }
)

Error Handling

try:
    conversation = manager.load_conversation("invalid_id")
except FileNotFoundError:
    print("Conversation not found")
    conversation = manager.create_conversation("New Chat")

try:
    manager.process_message(conversation.id, "Hello")
except Exception as e:
    print(f"Processing error: {e}")
    # Handle gracefully

Best Practices

  1. Unique Naming: Use descriptive conversation names for better organization
  2. Metadata Usage: Store relevant context in message metadata
  3. Regular Cleanup: Archive or delete old conversations periodically
  4. Error Handling: Always handle file I/O and processing errors
  5. Session Management: Properly link conversations to user sessions

Next Steps