github-featured/openai--openai-agents-python
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

Document AdvancedSQLiteSession and refactor session docs (#1791)

Browse Source 
Co-authored-by: Kazuhiro Sera <seratch@openai.com>
This commit is contained in:
Hassan Abu Alhaj
2025-10-17 03:03:23 +03:00
committed by GitHub
parent 3b7aed7e7c
commit 2ccdbdfd3d
8 changed files with 737 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/ref/extensions/memory/advanced_sqlite_session.md
+2 -2
View File
@@ -1,3 +1,3 @@
# `Advanced Sqlite Session`
# `AdvancedSQLiteSession`
::: agents.extensions.memory.advanced_sqlite_session
::: agents.extensions.memory.advanced_sqlite_session.AdvancedSQLiteSession
docs/ref/memory.md
+1
View File
@@ -6,3 +6,4 @@
members:
- Session
- SQLiteSession
- OpenAIConversationsSession
docs/running_agents.md
+2 -2
View File
@@ -89,7 +89,7 @@ async def main():
### Automatic conversation management with Sessions
For a simpler approach, you can use [Sessions](sessions.md) to automatically handle conversation history without manually calling `.to_input_list()`:
For a simpler approach, you can use [Sessions](sessions/index.md) to automatically handle conversation history without manually calling `.to_input_list()`:
```python
from agents import Agent, Runner, SQLiteSession
@@ -119,7 +119,7 @@ Sessions automatically:
- Stores new messages after each run
- Maintains separate conversations for different session IDs
See the [Sessions documentation](sessions.md) for more details.
See the [Sessions documentation](sessions/index.md) for more details.
### Server-managed conversations
docs/sessions/advanced_sqlite_session.md
+303
View File
@@ -0,0 +1,303 @@
# Advanced SQLite Sessions
`AdvancedSQLiteSession` is an enhanced version of the basic `SQLiteSession` that provides advanced conversation management capabilities including conversation branching, detailed usage analytics, and structured conversation queries.
## Features
- **Conversation branching**: Create alternative conversation paths from any user message
- **Usage tracking**: Detailed token usage analytics per turn with full JSON breakdowns
- **Structured queries**: Get conversations by turns, tool usage statistics, and more
- **Branch management**: Independent branch switching and management
- **Message structure metadata**: Track message types, tool usage, and conversation flow
## Quick start
```python
from agents import Agent, Runner
from agents.extensions.memory import AdvancedSQLiteSession
# Create agent
agent = Agent(
name="Assistant",
instructions="Reply very concisely.",
)
# Create an advanced session
session = AdvancedSQLiteSession(
session_id="conversation_123",
db_path="conversations.db",
create_tables=True
)
# First conversation turn
result = await Runner.run(
agent,
"What city is the Golden Gate Bridge in?",
session=session
)
print(result.final_output) # "San Francisco"
# IMPORTANT: Store usage data
await session.store_run_usage(result)
# Continue conversation
result = await Runner.run(
agent,
"What state is it in?",
session=session
)
print(result.final_output) # "California"
await session.store_run_usage(result)
```
## Initialization
```python
from agents.extensions.memory import AdvancedSQLiteSession
# Basic initialization
session = AdvancedSQLiteSession(
session_id="my_conversation",
create_tables=True # Auto-create advanced tables
)
# With persistent storage
session = AdvancedSQLiteSession(
session_id="user_123",
db_path="path/to/conversations.db",
create_tables=True
)
# With custom logger
import logging
logger = logging.getLogger("my_app")
session = AdvancedSQLiteSession(
session_id="session_456",
create_tables=True,
logger=logger
)
```
### Parameters
- `session_id` (str): Unique identifier for the conversation session
- `db_path` (str | Path): Path to SQLite database file. Defaults to `:memory:` for in-memory storage
- `create_tables` (bool): Whether to automatically create the advanced tables. Defaults to `False`
- `logger` (logging.Logger | None): Custom logger for the session. Defaults to module logger
## Usage tracking
AdvancedSQLiteSession provides detailed usage analytics by storing token usage data per conversation turn. **This is entirely dependent on the `store_run_usage` method being called after each agent run.**
### Storing usage data
```python
# After each agent run, store the usage data
result = await Runner.run(agent, "Hello", session=session)
await session.store_run_usage(result)
# This stores:
# - Total tokens used
# - Input/output token breakdown
# - Request count
# - Detailed JSON token information (if available)
```
### Retrieving usage statistics
```python
# Get session-level usage (all branches)
session_usage = await session.get_session_usage()
if session_usage:
print(f"Total requests: {session_usage['requests']}")
print(f"Total tokens: {session_usage['total_tokens']}")
print(f"Input tokens: {session_usage['input_tokens']}")
print(f"Output tokens: {session_usage['output_tokens']}")
print(f"Total turns: {session_usage['total_turns']}")
# Get usage for specific branch
branch_usage = await session.get_session_usage(branch_id="main")
# Get usage by turn
turn_usage = await session.get_turn_usage()
for turn_data in turn_usage:
print(f"Turn {turn_data['user_turn_number']}: {turn_data['total_tokens']} tokens")
if turn_data['input_tokens_details']:
print(f" Input details: {turn_data['input_tokens_details']}")
if turn_data['output_tokens_details']:
print(f" Output details: {turn_data['output_tokens_details']}")
# Get usage for specific turn
turn_2_usage = await session.get_turn_usage(user_turn_number=2)
```
## Conversation branching
One of the key features of AdvancedSQLiteSession is the ability to create conversation branches from any user message, allowing you to explore alternative conversation paths.
### Creating branches
```python
# Get available turns for branching
turns = await session.get_conversation_turns()
for turn in turns:
print(f"Turn {turn['turn']}: {turn['content']}")
print(f"Can branch: {turn['can_branch']}")
# Create a branch from turn 2
branch_id = await session.create_branch_from_turn(2)
print(f"Created branch: {branch_id}")
# Create a branch with custom name
branch_id = await session.create_branch_from_turn(
2,
branch_name="alternative_path"
)
# Create branch by searching for content
branch_id = await session.create_branch_from_content(
"weather",
branch_name="weather_focus"
)
```
### Branch management
```python
# List all branches
branches = await session.list_branches()
for branch in branches:
current = " (current)" if branch["is_current"] else ""
print(f"{branch['branch_id']}: {branch['user_turns']} turns, {branch['message_count']} messages{current}")
# Switch between branches
await session.switch_to_branch("main")
await session.switch_to_branch(branch_id)
# Delete a branch
await session.delete_branch(branch_id, force=True) # force=True allows deleting current branch
```
### Branch workflow example
```python
# Original conversation
result = await Runner.run(agent, "What's the capital of France?", session=session)
await session.store_run_usage(result)
result = await Runner.run(agent, "What's the weather like there?", session=session)
await session.store_run_usage(result)
# Create branch from turn 2 (weather question)
branch_id = await session.create_branch_from_turn(2, "weather_focus")
# Continue in new branch with different question
result = await Runner.run(
agent,
"What are the main tourist attractions in Paris?",
session=session
)
await session.store_run_usage(result)
# Switch back to main branch
await session.switch_to_branch("main")
# Continue original conversation
result = await Runner.run(
agent,
"How expensive is it to visit?",
session=session
)
await session.store_run_usage(result)
```
## Structured queries
AdvancedSQLiteSession provides several methods for analyzing conversation structure and content.
### Conversation analysis
```python
# Get conversation organized by turns
conversation_by_turns = await session.get_conversation_by_turns()
for turn_num, items in conversation_by_turns.items():
print(f"Turn {turn_num}: {len(items)} items")
for item in items:
if item["tool_name"]:
print(f" - {item['type']} (tool: {item['tool_name']})")
else:
print(f" - {item['type']}")
# Get tool usage statistics
tool_usage = await session.get_tool_usage()
for tool_name, count, turn in tool_usage:
print(f"{tool_name}: used {count} times in turn {turn}")
# Find turns by content
matching_turns = await session.find_turns_by_content("weather")
for turn in matching_turns:
print(f"Turn {turn['turn']}: {turn['content']}")
```
### Message structure
The session automatically tracks message structure including:
- Message types (user, assistant, tool_call, etc.)
- Tool names for tool calls
- Turn numbers and sequence numbers
- Branch associations
- Timestamps
## Database schema
AdvancedSQLiteSession extends the basic SQLite schema with two additional tables:
### message_structure table
```sql
CREATE TABLE message_structure (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
message_id INTEGER NOT NULL,
branch_id TEXT NOT NULL DEFAULT 'main',
message_type TEXT NOT NULL,
sequence_number INTEGER NOT NULL,
user_turn_number INTEGER,
branch_turn_number INTEGER,
tool_name TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (session_id) REFERENCES agent_sessions(session_id) ON DELETE CASCADE,
FOREIGN KEY (message_id) REFERENCES agent_messages(id) ON DELETE CASCADE
);
```
### turn_usage table
```sql
CREATE TABLE turn_usage (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
branch_id TEXT NOT NULL DEFAULT 'main',
user_turn_number INTEGER NOT NULL,
requests INTEGER DEFAULT 0,
input_tokens INTEGER DEFAULT 0,
output_tokens INTEGER DEFAULT 0,
total_tokens INTEGER DEFAULT 0,
input_tokens_details JSON,
output_tokens_details JSON,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (session_id) REFERENCES agent_sessions(session_id) ON DELETE CASCADE,
UNIQUE(session_id, branch_id, user_turn_number)
);
```
## Complete example
Check out the [complete example](https://github.com/openai/openai-agents-python/tree/main/examples/memory/advanced_sqlite_session_example.py) for a comprehensive demonstration of all features.
## API Reference
- [`AdvancedSQLiteSession`][agents.extensions.memory.advanced_sqlite_session.AdvancedSQLiteSession] - Main class
- [`Session`][agents.memory.session.Session] - Base session protocol
docs/sessions/encrypted_session.md
+175
View File
@@ -0,0 +1,175 @@
# Encrypted Sessions
`EncryptedSession` provides transparent encryption for any session implementation, securing conversation data with automatic expiration of old items.
## Features
- **Transparent encryption**: Wraps any session with Fernet encryption
- **Per-session keys**: Uses HKDF key derivation for unique encryption per session
- **Automatic expiration**: Old items are silently skipped when TTL expires
- **Drop-in replacement**: Works with any existing session implementation
## Installation
Encrypted sessions require the `encrypt` extra:
```bash
pip install openai-agents[encrypt]
```
## Quick start
```python
import asyncio
from agents import Agent, Runner
from agents.extensions.memory import EncryptedSession, SQLAlchemySession
async def main():
agent = Agent("Assistant")
# Create underlying session
underlying_session = SQLAlchemySession.from_url(
"user-123",
url="sqlite+aiosqlite:///:memory:",
create_tables=True
)
# Wrap with encryption
session = EncryptedSession(
session_id="user-123",
underlying_session=underlying_session,
encryption_key="your-secret-key-here",
ttl=600 # 10 minutes
)
result = await Runner.run(agent, "Hello", session=session)
print(result.final_output)
if __name__ == "__main__":
asyncio.run(main())
```
## Configuration
### Encryption key
The encryption key can be either a Fernet key or any string:
```python
from agents.extensions.memory import EncryptedSession
# Using a Fernet key (base64-encoded)
session = EncryptedSession(
session_id="user-123",
underlying_session=underlying_session,
encryption_key="your-fernet-key-here",
ttl=600
)
# Using a raw string (will be derived to a key)
session = EncryptedSession(
session_id="user-123",
underlying_session=underlying_session,
encryption_key="my-secret-password",
ttl=600
)
```
### TTL (Time To Live)
Set how long encrypted items remain valid:
```python
# Items expire after 1 hour
session = EncryptedSession(
session_id="user-123",
underlying_session=underlying_session,
encryption_key="secret",
ttl=3600 # 1 hour in seconds
)
# Items expire after 1 day
session = EncryptedSession(
session_id="user-123",
underlying_session=underlying_session,
encryption_key="secret",
ttl=86400 # 24 hours in seconds
)
```
## Usage with different session types
### With SQLite sessions
```python
from agents import SQLiteSession
from agents.extensions.memory import EncryptedSession
# Create encrypted SQLite session
underlying = SQLiteSession("user-123", "conversations.db")
session = EncryptedSession(
session_id="user-123",
underlying_session=underlying,
encryption_key="secret-key"
)
```
### With SQLAlchemy sessions
```python
from agents.extensions.memory import EncryptedSession, SQLAlchemySession
# Create encrypted SQLAlchemy session
underlying = SQLAlchemySession.from_url(
"user-123",
url="postgresql+asyncpg://user:pass@localhost/db",
create_tables=True
)
session = EncryptedSession(
session_id="user-123",
underlying_session=underlying,
encryption_key="secret-key"
)
```
!!! warning "Advanced Session Features"
When using `EncryptedSession` with advanced session implementations like `AdvancedSQLiteSession`, note that:
- Methods like `find_turns_by_content()` won't work effectively since message content is encrypted
- Content-based searches operate on encrypted data, limiting their effectiveness
## Key derivation
EncryptedSession uses HKDF (HMAC-based Key Derivation Function) to derive unique encryption keys per session:
- **Master key**: Your provided encryption key
- **Session salt**: The session ID
- **Info string**: `"agents.session-store.hkdf.v1"`
- **Output**: 32-byte Fernet key
This ensures that:
- Each session has a unique encryption key
- Keys cannot be derived without the master key
- Session data cannot be decrypted across different sessions
## Automatic expiration
When items exceed the TTL, they are automatically skipped during retrieval:
```python
# Items older than TTL are silently ignored
items = await session.get_items() # Only returns non-expired items
# Expired items don't affect session behavior
result = await Runner.run(agent, "Continue conversation", session=session)
```
## API Reference
- [`EncryptedSession`][agents.extensions.memory.encrypt_session.EncryptedSession] - Main class
- [`Session`][agents.memory.session.Session] - Base session protocol
docs/sessions.md → docs/sessions/index.md
+157 -183
View File
@@ -113,37 +113,49 @@ result = await Runner.run(
print(f"Agent: {result.final_output}")
```
## Memory options
## Session types
### No memory (default)
The SDK provides several session implementations for different use cases:
### OpenAI Conversations API sessions
Use [OpenAI's Conversations API](https://platform.openai.com/docs/api-reference/conversations) through `OpenAIConversationsSession`.
```python
# Default behavior - no session memory
result = await Runner.run(agent, "Hello")
```
from agents import Agent, Runner, OpenAIConversationsSession
### OpenAI Conversations API memory
Use the [OpenAI Conversations API](https://platform.openai.com/docs/api-reference/conversations/create) to persist
[conversation state](https://platform.openai.com/docs/guides/conversation-state?api-mode=responses#using-the-conversations-api) without managing your own database. This is helpful when you already rely on OpenAI-hosted infrastructure
for storing conversation history.
```python
from agents import OpenAIConversationsSession
# Create agent
agent = Agent(
name="Assistant",
instructions="Reply very concisely.",
)
# Create a new conversation
session = OpenAIConversationsSession()
# Optionally resume a previous conversation by passing a conversation ID
# session = OpenAIConversationsSession(conversation_id="conv_123")
# Start conversation
result = await Runner.run(
agent,
"Hello",
session=session,
"What city is the Golden Gate Bridge in?",
session=session
)
print(result.final_output) # "San Francisco"
# Continue the conversation
result = await Runner.run(
agent,
"What state is it in?",
session=session
)
print(result.final_output) # "California"
```
### SQLite memory
### SQLite sessions
The default, lightweight session implementation using SQLite:
```python
from agents import SQLiteSession
@@ -162,183 +174,79 @@ result = await Runner.run(
)
```
### Multiple sessions
### SQLAlchemy sessions
Production-ready sessions using any SQLAlchemy-supported database:
```python
from agents import Agent, Runner, SQLiteSession
from agents.extensions.memory import SQLAlchemySession
agent = Agent(name="Assistant")
# Different sessions maintain separate conversation histories
session_1 = SQLiteSession("user_123", "conversations.db")
session_2 = SQLiteSession("user_456", "conversations.db")
result1 = await Runner.run(
agent,
"Hello",
session=session_1
# Using database URL
session = SQLAlchemySession.from_url(
"user_123",
url="postgresql+asyncpg://user:pass@localhost/db",
create_tables=True
)
result2 = await Runner.run(
agent,
"Hello",
session=session_2
)
```
### SQLAlchemy-powered sessions
For more advanced use cases, you can use a SQLAlchemy-powered session backend. This allows you to use any database supported by SQLAlchemy (PostgreSQL, MySQL, SQLite, etc.) for session storage.
**Example 1: Using `from_url` with in-memory SQLite**
This is the simplest way to get started, ideal for development and testing.
```python
import asyncio
from agents import Agent, Runner
from agents.extensions.memory.sqlalchemy_session import SQLAlchemySession
async def main():
agent = Agent("Assistant")
session = SQLAlchemySession.from_url(
"user-123",
url="sqlite+aiosqlite:///:memory:",
create_tables=True, # Auto-create tables for the demo
)
result = await Runner.run(agent, "Hello", session=session)
if __name__ == "__main__":
asyncio.run(main())
```
**Example 2: Using an existing SQLAlchemy engine**
In a production application, you likely already have a SQLAlchemy `AsyncEngine` instance. You can pass it directly to the session.
```python
import asyncio
from agents import Agent, Runner
from agents.extensions.memory.sqlalchemy_session import SQLAlchemySession
# Using existing engine
from sqlalchemy.ext.asyncio import create_async_engine
async def main():
# In your application, you would use your existing engine
engine = create_async_engine("sqlite+aiosqlite:///conversations.db")
agent = Agent("Assistant")
session = SQLAlchemySession(
"user-456",
engine=engine,
create_tables=True, # Auto-create tables for the demo
)
result = await Runner.run(agent, "Hello", session=session)
print(result.final_output)
await engine.dispose()
if __name__ == "__main__":
asyncio.run(main())
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
session = SQLAlchemySession("user_123", engine=engine, create_tables=True)
```
See [SQLAlchemy Sessions](sqlalchemy_session.md) for detailed documentation.
### Advanced SQLite sessions
Enhanced SQLite sessions with conversation branching, usage analytics, and structured queries:
```python
from agents.extensions.memory import AdvancedSQLiteSession
# Create with advanced features
session = AdvancedSQLiteSession(
session_id="user_123",
db_path="conversations.db",
create_tables=True
)
# Automatic usage tracking
result = await Runner.run(agent, "Hello", session=session)
await session.store_run_usage(result) # Track token usage
# Conversation branching
await session.create_branch_from_turn(2) # Branch from turn 2
```
See [Advanced SQLite Sessions](advanced_sqlite_session.md) for detailed documentation.
### Encrypted sessions
For applications requiring encryption of conversation data at rest, you can use `EncryptedSession` to wrap any session backend with transparent encryption and automatic TTL-based expiration. This requires the `encrypt` extra: `pip install openai-agents[encrypt]`.
The `EncryptedSession` uses Fernet encryption with per-session key derivation (HKDF) and supports automatic expiration of old messages. When items exceed the TTL, they are silently skipped during retrieval.
**Example: Encrypting SQLAlchemy session data**
Transparent encryption wrapper for any session implementation:
```python
import asyncio
from agents import Agent, Runner
from agents.extensions.memory import EncryptedSession, SQLAlchemySession
async def main():
# Create underlying session (works with any SessionABC implementation)
underlying_session = SQLAlchemySession.from_url(
session_id="user-123",
url="postgresql+asyncpg://app:secret@db.example.com/agents",
create_tables=True,
)
# Wrap with encryption and TTL-based expiration
session = EncryptedSession(
session_id="user-123",
underlying_session=underlying_session,
encryption_key="your-encryption-key", # Use a secure key from your secrets management
ttl=600, # 10 minutes - items older than this are silently skipped
)
agent = Agent("Assistant")
result = await Runner.run(agent, "Hello", session=session)
print(result.final_output)
if __name__ == "__main__":
asyncio.run(main())
```
**Key features:**
- **Transparent encryption**: Automatically encrypts all session items before storage and decrypts on retrieval
- **Per-session key derivation**: Uses HKDF with the session ID as salt to derive unique encryption keys
- **TTL-based expiration**: Automatically expires old messages based on configurable time-to-live (default: 10 minutes)
- **Flexible key input**: Accepts either Fernet keys or raw strings as encryption keys
- **Wraps any session**: Works with SQLite, SQLAlchemy, or custom session implementations
!!! warning "Important security notes"
- Store your encryption key securely (e.g., environment variables, secrets manager)
- Expired tokens are rejected based on the application server's system clock - ensure all servers are time-synchronized with NTP to avoid valid tokens being rejected due to clock drift
- The underlying session still stores encrypted data, so you maintain control over your database infrastructure
## Custom memory implementations
You can implement your own session memory by creating a class that follows the [`Session`][agents.memory.session.Session] protocol:
```python
from agents.memory.session import SessionABC
from agents.items import TResponseInputItem
from typing import List
class MyCustomSession(SessionABC):
"""Custom session implementation following the Session protocol."""
def __init__(self, session_id: str):
self.session_id = session_id
# Your initialization here
async def get_items(self, limit: int | None = None) -> List[TResponseInputItem]:
"""Retrieve conversation history for this session."""
# Your implementation here
pass
async def add_items(self, items: List[TResponseInputItem]) -> None:
"""Store new items for this session."""
# Your implementation here
pass
async def pop_item(self) -> TResponseInputItem | None:
"""Remove and return the most recent item from this session."""
# Your implementation here
pass
async def clear_session(self) -> None:
"""Clear all items for this session."""
# Your implementation here
pass
# Use your custom session
agent = Agent(name="Assistant")
result = await Runner.run(
agent,
"Hello",
session=MyCustomSession("my_session")
# Create underlying session
underlying_session = SQLAlchemySession.from_url(
"user_123",
url="sqlite+aiosqlite:///conversations.db",
create_tables=True
)
# Wrap with encryption and TTL
session = EncryptedSession(
session_id="user_123",
underlying_session=underlying_session,
encryption_key="your-secret-key",
ttl=600 # 10 minutes
)
result = await Runner.run(agent, "Hello", session=session)
```
See [Encrypted Sessions](encrypted_session.md) for detailed documentation.
## Session management
### Session ID naming
@@ -358,12 +266,32 @@ Use meaningful session IDs that help you organize conversations:
- Use encrypted sessions (`EncryptedSession(session_id, underlying_session, encryption_key)`) to wrap any session with transparent encryption and TTL-based expiration
- Consider implementing custom session backends for other production systems (Redis, Django, etc.) for more advanced use cases
### Session management
### Multiple sessions
```python
# Clear a session when conversation should start fresh
await session.clear_session()
from agents import Agent, Runner, SQLiteSession
agent = Agent(name="Assistant")
# Different sessions maintain separate conversation histories
session_1 = SQLiteSession("user_123", "conversations.db")
session_2 = SQLiteSession("user_456", "conversations.db")
result1 = await Runner.run(
agent,
"Help me with my account",
session=session_1
)
result2 = await Runner.run(
agent,
"What are my charges?",
session=session_2
)
```
### Session sharing
```python
# Different agents can share the same session
support_agent = Agent(name="Support")
billing_agent = Agent(name="Billing")
@@ -446,12 +374,58 @@ if __name__ == "__main__":
asyncio.run(main())
```
## Custom session implementations
You can implement your own session memory by creating a class that follows the [`Session`][agents.memory.session.Session] protocol:
```python
from agents.memory.session import SessionABC
from agents.items import TResponseInputItem
from typing import List
class MyCustomSession(SessionABC):
"""Custom session implementation following the Session protocol."""
def __init__(self, session_id: str):
self.session_id = session_id
# Your initialization here
async def get_items(self, limit: int | None = None) -> List[TResponseInputItem]:
"""Retrieve conversation history for this session."""
# Your implementation here
pass
async def add_items(self, items: List[TResponseInputItem]) -> None:
"""Store new items for this session."""
# Your implementation here
pass
async def pop_item(self) -> TResponseInputItem | None:
"""Remove and return the most recent item from this session."""
# Your implementation here
pass
async def clear_session(self) -> None:
"""Clear all items for this session."""
# Your implementation here
pass
# Use your custom session
agent = Agent(name="Assistant")
result = await Runner.run(
agent,
"Hello",
session=MyCustomSession("my_session")
)
```
## API Reference
For detailed API documentation, see:
- [`Session`][agents.memory.Session] - Protocol interface
- [`SQLiteSession`][agents.memory.SQLiteSession] - SQLite implementation
- [`OpenAIConversationsSession`](ref/memory/openai_conversations_session.md) - OpenAI Conversations API implementation
- [`Session`][agents.memory.session.Session] - Protocol interface
- [`OpenAIConversationsSession`][agents.memory.OpenAIConversationsSession] - OpenAI Conversations API implementation
- [`SQLiteSession`][agents.memory.sqlite_session.SQLiteSession] - Basic SQLite implementation
- [`SQLAlchemySession`][agents.extensions.memory.sqlalchemy_session.SQLAlchemySession] - SQLAlchemy-powered implementation
- [`EncryptedSession`][agents.extensions.memory.encrypt_session.EncryptedSession] - Encrypted session wrapper with TTL
- [`AdvancedSQLiteSession`][agents.extensions.memory.advanced_sqlite_session.AdvancedSQLiteSession] - Enhanced SQLite with branching and analytics
- [`EncryptedSession`][agents.extensions.memory.encrypt_session.EncryptedSession] - Encrypted wrapper for any session
docs/sessions/sqlalchemy_session.md
+76
View File
@@ -0,0 +1,76 @@
# SQLAlchemy Sessions
`SQLAlchemySession` uses SQLAlchemy to provide a production-ready session implementation, allowing you to use any database supported by SQLAlchemy (PostgreSQL, MySQL, SQLite, etc.) for session storage.
## Installation
SQLAlchemy sessions require the `sqlalchemy` extra:
```bash
pip install openai-agents[sqlalchemy]
```
## Quick start
### Using database URL
The simplest way to get started:
```python
import asyncio
from agents import Agent, Runner
from agents.extensions.memory import SQLAlchemySession
async def main():
agent = Agent("Assistant")
# Create session using database URL
session = SQLAlchemySession.from_url(
"user-123",
url="sqlite+aiosqlite:///:memory:",
create_tables=True
)
result = await Runner.run(agent, "Hello", session=session)
print(result.final_output)
if __name__ == "__main__":
asyncio.run(main())
```
### Using existing engine
For applications with existing SQLAlchemy engines:
```python
import asyncio
from agents import Agent, Runner
from agents.extensions.memory import SQLAlchemySession
from sqlalchemy.ext.asyncio import create_async_engine
async def main():
# Create your database engine
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
agent = Agent("Assistant")
session = SQLAlchemySession(
"user-456",
engine=engine,
create_tables=True
)
result = await Runner.run(agent, "Hello", session=session)
print(result.final_output)
# Clean up
await engine.dispose()
if __name__ == "__main__":
asyncio.run(main())
```
## API Reference
- [`SQLAlchemySession`][agents.extensions.memory.sqlalchemy_session.SQLAlchemySession] - Main class
- [`Session`][agents.memory.session.Session] - Base session protocol
mkdocs.yml
+21 -5
View File
@@ -57,7 +57,11 @@ plugins:
- Documentation:
- agents.md
- running_agents.md
- sessions.md
- Sessions:
- sessions/index.md
- sessions/sqlalchemy_session.md
- sessions/advanced_sqlite_session.md
- sessions/encrypted_session.md
- results.md
- streaming.md
- repl.md
@@ -146,7 +150,7 @@ plugins:
- ref/extensions/litellm.md
- ref/extensions/memory/sqlalchemy_session.md
- ref/extensions/memory/encrypt_session.md
- ref/extensions/memory/advanced_sqlite_session.md
- locale: ja
name: 日本語
build: true
@@ -157,7 +161,11 @@ plugins:
- ドキュメント:
- agents.md
- running_agents.md
- sessions.md
- セッション:
- sessions/index.md
- sessions/sqlalchemy_session.md
- sessions/advanced_sqlite_session.md
- sessions/encrypted_session.md
- results.md
- streaming.md
- repl.md
@@ -192,7 +200,11 @@ plugins:
- 문서:
- agents.md
- running_agents.md
- sessions.md
- 세션:
- sessions/index.md
- sessions/sqlalchemy_session.md
- sessions/advanced_sqlite_session.md
- sessions/encrypted_session.md
- results.md
- streaming.md
- repl.md
@@ -227,7 +239,11 @@ plugins:
- 文档:
- agents.md
- running_agents.md
- sessions.md
- 会话:
- sessions/index.md
- sessions/sqlalchemy_session.md
- sessions/advanced_sqlite_session.md
- sessions/encrypted_session.md
- results.md
- streaming.md
- repl.md