セッション
Agents SDK は、複数のエージェント実行にまたがる会話履歴を自動的に維持する組み込みのセッション メモリを提供し、ターン間で手動で .to_input_list()
を扱う必要をなくします。
セッションは特定のセッションの会話履歴を保存し、明示的な手動メモリ管理なしにエージェントがコンテキストを維持できるようにします。これは、エージェントに過去のやり取りを記憶させたいチャット アプリケーションやマルチターン会話の構築に特に有用です。
クイックスタート
from agents import Agent, Runner, SQLiteSession
# Create agent
agent = Agent(
name="Assistant",
instructions="Reply very concisely.",
)
# Create a session instance with a session ID
session = SQLiteSession("conversation_123")
# First turn
result = await Runner.run(
agent,
"What city is the Golden Gate Bridge in?",
session=session
)
print(result.final_output) # "San Francisco"
# Second turn - agent automatically remembers previous context
result = await Runner.run(
agent,
"What state is it in?",
session=session
)
print(result.final_output) # "California"
# Also works with synchronous runner
result = Runner.run_sync(
agent,
"What's the population?",
session=session
)
print(result.final_output) # "Approximately 39 million"
仕組み
セッション メモリが有効な場合:
- 各実行の前 : ランナーはセッションの会話履歴を自動的に取得し、入力アイテムの先頭に付加します。
- 各実行の後 : 実行中に生成されたすべての新規アイテム(ユーザー入力、アシスタントの応答、ツール呼び出しなど)が自動的にセッションに保存されます。
- コンテキストの保持 : 同じセッションでの後続の各実行には会話履歴全体が含まれ、エージェントはコンテキストを維持できます。
これにより、ターン間で .to_input_list()
を手動で呼び出したり、会話状態を管理したりする必要がなくなります。
メモリ操作
基本操作
セッションは会話履歴を管理するためにいくつかの操作をサポートします:
from agents import SQLiteSession
session = SQLiteSession("user_123", "conversations.db")
# Get all items in a session
items = await session.get_items()
# Add new items to a session
new_items = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi there!"}
]
await session.add_items(new_items)
# Remove and return the most recent item
last_item = await session.pop_item()
print(last_item) # {"role": "assistant", "content": "Hi there!"}
# Clear all items from a session
await session.clear_session()
修正のための pop_item の使用
pop_item
メソッドは、会話の最後のアイテムを取り消したり変更したりしたい場合に特に役立ちます:
from agents import Agent, Runner, SQLiteSession
agent = Agent(name="Assistant")
session = SQLiteSession("correction_example")
# Initial conversation
result = await Runner.run(
agent,
"What's 2 + 2?",
session=session
)
print(f"Agent: {result.final_output}")
# User wants to correct their question
assistant_item = await session.pop_item() # Remove agent's response
user_item = await session.pop_item() # Remove user's question
# Ask a corrected question
result = await Runner.run(
agent,
"What's 2 + 3?",
session=session
)
print(f"Agent: {result.final_output}")
メモリ オプション
メモリなし(デフォルト)
SQLite メモリ
from agents import SQLiteSession
# In-memory database (lost when process ends)
session = SQLiteSession("user_123")
# Persistent file-based database
session = SQLiteSession("user_123", "conversations.db")
# Use the session
result = await Runner.run(
agent,
"Hello",
session=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,
"Hello",
session=session_1
)
result2 = await Runner.run(
agent,
"Hello",
session=session_2
)
カスタム メモリ実装
Session
プロトコルに従うクラスを作成することで、独自のセッション メモリを実装できます:
from agents.memory import Session
from typing import List
class MyCustomSession:
"""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[dict]:
"""Retrieve conversation history for this session."""
# Your implementation here
pass
async def add_items(self, items: List[dict]) -> None:
"""Store new items for this session."""
# Your implementation here
pass
async def pop_item(self) -> dict | 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")
)
セッション管理
セッション ID の命名
会話を整理するのに役立つ意味のあるセッション ID を使用します:
- ユーザー ベース:
"user_12345"
- スレッド ベース:
"thread_abc123"
- コンテキスト ベース:
"support_ticket_456"
メモリの永続化
- 一時的な会話にはインメモリ SQLite(
SQLiteSession("session_id")
)を使用します - 永続的な会話にはファイル ベースの SQLite(
SQLiteSession("session_id", "path/to/db.sqlite")
)を使用します - 本番システム向けにはカスタム セッション バックエンド(Redis、PostgreSQL など)の実装を検討します
セッション管理
# Clear a session when conversation should start fresh
await session.clear_session()
# Different agents can share the same session
support_agent = Agent(name="Support")
billing_agent = Agent(name="Billing")
session = SQLiteSession("user_123")
# Both agents will see the same conversation history
result1 = await Runner.run(
support_agent,
"Help me with my account",
session=session
)
result2 = await Runner.run(
billing_agent,
"What are my charges?",
session=session
)
完全な例
セッション メモリが実際に動作する完全な例を次に示します:
import asyncio
from agents import Agent, Runner, SQLiteSession
async def main():
# Create an agent
agent = Agent(
name="Assistant",
instructions="Reply very concisely.",
)
# Create a session instance that will persist across runs
session = SQLiteSession("conversation_123", "conversation_history.db")
print("=== Sessions Example ===")
print("The agent will remember previous messages automatically.\n")
# First turn
print("First turn:")
print("User: What city is the Golden Gate Bridge in?")
result = await Runner.run(
agent,
"What city is the Golden Gate Bridge in?",
session=session
)
print(f"Assistant: {result.final_output}")
print()
# Second turn - the agent will remember the previous conversation
print("Second turn:")
print("User: What state is it in?")
result = await Runner.run(
agent,
"What state is it in?",
session=session
)
print(f"Assistant: {result.final_output}")
print()
# Third turn - continuing the conversation
print("Third turn:")
print("User: What's the population of that state?")
result = await Runner.run(
agent,
"What's the population of that state?",
session=session
)
print(f"Assistant: {result.final_output}")
print()
print("=== Conversation Complete ===")
print("Notice how the agent remembered the context from previous turns!")
print("Sessions automatically handles conversation history.")
if __name__ == "__main__":
asyncio.run(main())
API リファレンス
詳細な API ドキュメントは以下をご覧ください:
Session
- プロトコル インターフェースSQLiteSession
- SQLite 実装