> ## Documentation Index > Fetch the complete documentation index at: https://docs.upsonic.ai/llms.txt > Use this file to discover all available pages before exploring further. # Chat > Build conversational AI experiences with stateful session management ## Overview In the Upsonic framework, `Chat` is a comprehensive conversational interface that provides stateful session management for AI agents. It handles the complete lifecycle of chat interactions, including message history, memory integration, cost tracking, and both streaming and non-streaming interfaces. The Chat class serves as a high-level orchestrator that manages: * Session lifecycle and state management * Memory integration and conversation persistence * Cost and token tracking across sessions * Both blocking and streaming response modes * Error handling and retry mechanisms * Concurrent invocation control ## Chat Attributes The Chat class provides comprehensive configuration options to customize conversational behavior and session management. ### Core Attributes | Attribute | Type | Description | | - | - | - | | **session\_id** | str | Unique identifier for this chat session | | **user\_id** | str | Unique identifier for the user | | **agent** | Agent | The Agent instance to handle conversations | | **storage** | Optional\[Storage] | Storage backend for persistence (defaults to InMemoryStorage) | ### Memory Configuration | Attribute | Type | Description | | - | - | - | | **full\_session\_memory** | bool | Enable full conversation history storage (default: True) | | **summary\_memory** | bool | Enable conversation summarization (default: False) | | **user\_analysis\_memory** | bool | Enable user profile analysis (default: False) | | **user\_profile\_schema** | Optional\[Type\[BaseModel]] | Custom user profile schema | | **dynamic\_user\_profile** | bool | Enable dynamic profile schema generation (default: False) | | **num\_last\_messages** | Optional\[int] | Limit conversation history to last N messages | | **feed\_tool\_call\_results** | bool | Include tool calls in memory (default: False) | | **user\_memory\_mode** | Literal\['update', 'replace'] | How to update user profiles (default: 'update') | ### Advanced Configuration | Attribute | Type | Description | | - | - | - | | **debug** | bool | Enable debug logging (default: False) | | **max\_concurrent\_invocations** | int | Maximum concurrent invoke calls (default: 1) | | **retry\_attempts** | int | Number of retry attempts for failed calls (default: 3) | | **retry\_delay** | float | Delay between retry attempts in seconds (default: 1.0) | ## Creating a Chat Session Chat sessions are created directly in code using the constructor. Each chat session can be customized with specific memory settings, storage backends, and session management options. ### Basic Chat Creation ```python theme={null} import asyncio from upsonic import Chat, Agent async def main(): # Create agent agent = Agent("anthropic/claude-sonnet-4-5") # Create basic chat session chat = Chat( session_id="user123_session1", user_id="user123", agent=agent ) # Send a message response = await chat.invoke("Hello, how are you?") print(response) # Access chat history print(chat.all_messages) if chat.usage.cost is not None: print(f"Total cost: ${chat.usage.cost:.4f}") if __name__ == "__main__": asyncio.run(main()) ``` ### Chat with Custom Memory Settings ```python theme={null} from upsonic.storage.providers import SqliteStorage from upsonic import Chat # Create storage backend storage = SqliteStorage( sessions_table_name="sessions", profiles_table_name="profiles", db_file="chat.db" ) # Create chat with advanced memory settings chat = Chat( session_id="session1", user_id="user1", agent=agent, storage=storage, full_session_memory=True, summary_memory=True, user_analysis_memory=True, user_memory_mode='update' ) ``` ### Chat with Debug and Retry Configuration ```python theme={null} # Create chat with custom retry and debug settings chat = Chat( session_id="debug_session", user_id="user1", agent=agent, debug=True, max_concurrent_invocations=2, retry_attempts=5, retry_delay=2.0 ) ``` ## Advanced Chat Configuration ### Custom Storage Backends ```python theme={null} from upsonic.storage.providers import SqliteStorage, PostgresStorage from upsonic import Chat # SQLite storage for development and small-scale applications sqlite_storage = SqliteStorage( sessions_table_name="sessions", profiles_table_name="profiles", db_file="chat_sessions.db" # Optional: None for in-memory DB ) chat = Chat( session_id="sqlite_session", user_id="user1", agent=agent, storage=sqlite_storage ) # PostgreSQL storage for production and large-scale applications postgres_storage = PostgresStorage( sessions_table_name="sessions", profiles_table_name="profiles", db_url="postgresql://user:password@localhost:5432/chat_db", schema="public" # Optional: defaults to "public" ) chat = Chat( session_id="postgres_session", user_id="user1", agent=agent, storage=postgres_storage ) ``` ### Memory Management Strategies ```python theme={null} # Full conversation memory with summarization chat = Chat( session_id="summarized_session", user_id="user1", agent=agent, full_session_memory=True, summary_memory=True, num_last_messages=50 # Keep last 50 messages in memory ) # User profile analysis from pydantic import BaseModel class UserProfile(BaseModel): interests: list[str] = [] communication_style: str = "professional" expertise_level: str = "beginner" chat = Chat( session_id="analyzed_session", user_id="user1", agent=agent, user_analysis_memory=True, user_profile_schema=UserProfile, dynamic_user_profile=True ) ``` ## Chat Interface Methods ### Basic Messaging ```python theme={null} # Send a simple message response = await chat.invoke("What's the weather like today?") # Send a message with attachments response = await chat.invoke( "Analyze this document", attachments=["document.pdf", "data.csv"] ) # Send a Task object from upsonic import Task task = Task( description="Write a Python function to sort a list", tools=[some_tool] ) response = await chat.invoke(task) ``` ### Streaming Responses ```python theme={null} # Stream a response async for chunk in chat.invoke("Tell me a story", stream=True): print(chunk, end='', flush=True) # Using the dedicated stream method async for chunk in chat.stream("Explain quantum computing"): print(chunk, end='', flush=True) ``` ### Session Management ```python theme={null} # Get session information print(f"Session state: {chat.state}") print(f"Total messages: {len(chat.all_messages)}") print(f"Input tokens: {chat.usage.input_tokens}") print(f"Output tokens: {chat.usage.output_tokens}") if chat.usage.cost is not None: print(f"Total cost: ${chat.usage.cost:.4f}") # Session timing (wall-clock) print(f"Session duration: {chat.duration:.1f}s") # Models that contributed to this session print(f"Models used: {chat.usage.models}") ``` ## Session State Management ### Session States The Chat class manages several session states: * **IDLE**: Session is ready to accept new messages * **AWAITING\_RESPONSE**: Processing a user message * **STREAMING**: Streaming a response to the user * **ERROR**: Session encountered an error ```python theme={null} # Check session state if chat.state == SessionState.IDLE: response = await chat.invoke("Hello!") elif chat.state == SessionState.ERROR: chat.reset_session() # Reset to recover from error ``` ### Session Lifecycle ```python theme={null} # Create and use a chat session async with chat: # Send multiple messages response1 = await chat.invoke("What is AI?") response2 = await chat.invoke("How does it work?") # Session automatically cleaned up on exit ``` ## Memory Integration ### Conversation History ```python theme={null} # Access full conversation history messages = chat.all_messages for message in messages: print(f"{message.role}: {message.content}") if message.attachments: print(f" Attachments: {message.attachments}") if message.tool_calls: print(f" Tool calls: {message.tool_calls}") # Get recent messages only recent_messages = chat.get_recent_messages(count=5) ``` ### Memory Persistence ```python theme={null} # Chat automatically persists to storage chat = Chat( session_id="persistent_session", user_id="user1", agent=agent, storage=postgres_storage, full_session_memory=True ) # Memory is automatically loaded and saved response = await chat.invoke("Remember that I prefer Python over Java") # This preference is stored in user profile response = await chat.invoke("What programming language do I prefer?") # Agent will recall the preference from memory ``` ## Cost and Token Tracking ### Cost Monitoring ```python theme={null} # Track costs across the session via the unified usage view u = chat.usage if u.cost is not None: print(f"Total cost: ${u.cost:.4f}") print(f"Input tokens: {u.input_tokens}") print(f"Output tokens: {u.output_tokens}") print(f"Requests: {u.requests}") print(f"Tool calls: {u.tool_calls}") print(f"Models used: {u.models}") # JSON-friendly snapshot (for logs/dashboards) import json print(json.dumps(u.to_dict(), indent=2)) ``` ### Session Analytics ```python theme={null} # Session timing and usage aggregate print(f"Session duration: {chat.duration:.1f}s") print(f"Total messages: {len(chat.all_messages)}") print(f"Requests: {chat.usage.requests}") print(f"Tool calls: {chat.usage.tool_calls}") print(f"Model time: {chat.usage.model_execution_time:.2f}s") print(f"Tool time: {chat.usage.tool_execution_time:.2f}s") print(f"Framework time: {chat.usage.upsonic_execution_time:.2f}s") if chat.usage.cost is not None: print(f"Total cost: ${chat.usage.cost:.4f}") ``` ## Error Handling and Retry Logic ### Automatic Retry ```python theme={null} # Chat automatically retries failed requests chat = Chat( session_id="robust_session", user_id="user1", agent=agent, retry_attempts=5, retry_delay=2.0 ) # Network errors are automatically retried response = await chat.invoke("This might fail due to network issues") # Chat will retry up to 5 times with exponential backoff ``` ### Error Recovery ```python theme={null} # Check for errors and recover if chat.state == SessionState.ERROR: print("Session encountered an error") chat.reset_session() # Reset to recover response = await chat.invoke("Let's start over") ``` ## Practical Examples ### Customer Support Chat ```python theme={null} from upsonic.storage.providers import SqliteStorage from upsonic import Chat # Create storage for customer support support_storage = SqliteStorage( sessions_table_name="support_sessions", profiles_table_name="customer_profiles", db_file="support_chat.db" ) # Create a customer support chat support_chat = Chat( session_id="support_ticket_12345", user_id="customer_456", agent=Agent("anthropic/claude-sonnet-4-5"), storage=support_storage, full_session_memory=True, summary_memory=True, user_analysis_memory=True ) # Handle customer inquiry response = await support_chat.invoke("I'm having trouble with my account login") # Agent has access to conversation history and user profile # Continue the conversation response = await support_chat.invoke("The error message says 'invalid credentials'") # Agent remembers the context and can provide targeted help ``` ### Educational Assistant ```python theme={null} from upsonic.storage.providers import SqliteStorage from upsonic import Chat # Create storage for educational chat edu_storage = SqliteStorage( sessions_table_name="learning_sessions", profiles_table_name="student_profiles", db_file="education_chat.db" ) # Create an educational chat with memory edu_chat = Chat( session_id="student_learning_session", user_id="student_789", agent=Agent("anthropic/claude-sonnet-4-5"), storage=edu_storage, full_session_memory=True, user_analysis_memory=True, user_memory_mode='update' ) # Progressive learning conversation response = await edu_chat.invoke("I want to learn Python programming") # Agent creates a learning plan response = await edu_chat.invoke("What should I start with?") # Agent provides personalized recommendations based on user profile response = await edu_chat.invoke("I'm confused about functions") # Agent explains functions in context of previous conversation ``` ### Multi-User Chat System ```python theme={null} from upsonic.storage.providers import SqliteStorage from upsonic import Chat # Create multiple chat sessions for different users chats = {} # Create shared storage for multi-user system shared_storage = SqliteStorage( sessions_table_name="user_sessions", profiles_table_name="user_profiles", db_file="multi_user_chat.db" ) def create_user_chat(user_id: str, agent: Agent): return Chat( session_id=f"user_{user_id}_session", user_id=user_id, agent=agent, storage=shared_storage, full_session_memory=True, user_analysis_memory=True ) # Handle multiple users async def handle_user_message(user_id: str, message: str): if user_id not in chats: chats[user_id] = create_user_chat(user_id, agent) chat = chats[user_id] response = await chat.invoke(message) return response ``` ### Streaming Chat Interface ```python theme={null} from upsonic import Chat # Create a streaming chat for real-time applications streaming_chat = Chat( session_id="realtime_chat", user_id="user_streaming", agent=agent, debug=True ) # Stream responses for real-time UI updates async def stream_chat_response(message: str): async for chunk in streaming_chat.stream(message): # Send chunk to frontend via WebSocket await websocket.send(chunk) ``` ## Best Practices ### Performance Optimization 1. **Choose appropriate storage**: Use in-memory for development, persistent storage for production 2. **Configure memory settings**: Enable only needed memory features to optimize performance 3. **Set retry limits**: Balance between reliability and response time 4. **Use streaming for long responses**: Improve user experience with real-time output ### Session Management 1. **Use unique session IDs**: Ensure proper session isolation 2. **Implement session cleanup**: Close unused sessions to free resources 3. **Monitor session state**: Handle errors gracefully 4. **Use context managers**: Ensure proper resource cleanup ### Memory Configuration 1. **Enable summarization for long conversations**: Prevent memory overflow 2. **Use user analysis for personalization**: Improve user experience 3. **Configure message limits**: Balance between context and performance 4. **Choose appropriate memory modes**: Update vs replace based on use case ## Complete Example ```python theme={null} import asyncio from upsonic import Chat, Agent, Task from upsonic.storage.providers import SqliteStorage async def main(): # Create agent agent = Agent("anthropic/claude-sonnet-4-5") # Create storage storage = SqliteStorage( sessions_table_name="sessions", profiles_table_name="profiles", db_file="chat_sessions.db" ) # Create chat with comprehensive configuration chat = Chat( session_id="comprehensive_session", user_id="demo_user", agent=agent, storage=storage, full_session_memory=True, summary_memory=True, user_analysis_memory=True, debug=True, retry_attempts=3, retry_delay=1.5 ) try: # Basic conversation print("=== BASIC CONVERSATION ===") response = await chat.invoke("Hello! I'm interested in learning about AI.") print(f"Agent: {response}") # Follow-up with context response = await chat.invoke("What are the main types of AI?") print(f"Agent: {response}") # Streaming response print("\n=== STREAMING RESPONSE ===") async for chunk in chat.stream("Explain machine learning in detail"): print(chunk, end='', flush=True) print() # New line after streaming # Session analytics print("\n=== SESSION ANALYTICS ===") print(f"Total messages: {len(chat.all_messages)}") u = chat.usage if u.cost is not None: print(f"Total cost: ${u.cost:.4f}") print(f"Input tokens: {u.input_tokens}") print(f"Output tokens: {u.output_tokens}") print(f"Duration: {chat.duration:.1f}s") finally: # Clean up await chat.close() if __name__ == "__main__": asyncio.run(main()) ``` This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.