OpenAI「Agents SDK」②セッション

前回の続き。
https://zenn.dev/kun432/scraps/ced02ed77be7cd
今回はセッションに関して。

kun432

セッション

マルチターンのエージェント実行の会話履歴を保持するために、組み込みのセッションメモリが用意されている。以下はビルトインのSQLiteを使ったセッションの例。

クイックスタート

from agents import Agent, Runner, SQLiteSession
import asyncio

# エージェントを作成
agent = Agent(
    name="Assistant",
    instructions="日本語で簡潔に回答してください。",
)

async def main():
    # セッションを初期化
    session = SQLiteSession("session_1")

    # 最初の実行
    result = await Runner.run(
        agent,
        "ゴールデンゲートブリッジはどの都市にありますか？",
        session=session
    )
    print(result.final_output)

    print("-" * 20)

    # 2回目の実行 - エージェントは自動的に前回の文脈を覚えている
    result = await Runner.run(
        agent,
        "それはどの州ですか？",
        session=session
    )
    print(result.final_output)

    print("-" * 20)

    # 3回目も同様
    result = await Runner.run(
        agent,
        "人口はどれくらいですか？",
        session=session
    )
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())

出力

サンフランシスコにあります。
--------------------
カリフォルニア州です。
--------------------
サンフランシスコの人口は約88万人です。

上記は非同期の例だが、同期の場合も同様。

from agents import Agent, Runner, SQLiteSession
import asyncio

# エージェントを作成
agent = Agent(
    name="Assistant",
    instructions="日本語で簡潔に回答してください。",
)

def main():
    # セッションを初期化
    session = SQLiteSession("session_2")

    # 最初の実行
    result = Runner.run_sync(
        agent,
        "ゴールデンゲートブリッジはどの都市にありますか？",
        session=session
    )
    print(result.final_output)

    print("-" * 20)

    # 2回目の実行 - エージェントは自動的に前回の文脈を覚えている
    result = Runner.run_sync(
        agent,
        "それはどの州ですか？",
        session=session
    )
    print(result.final_output)

    print("-" * 20)

    # 3回目も同様
    result = Runner.run_sync(
        agent,
        "人口はどれくらいですか？",
        session=session
    )
    print(result.final_output)

if __name__ == "__main__":
    main()

出力

サンフランシスコにあります。
--------------------
カリフォルニア州です。
--------------------
サンフランシスコの人口は約88万人です。

仕組み

セッションが有効になっている場合、以下のような挙動になる。

実行前: ランナーがセッションの会話履歴を自動的に取得、入力アイテムの先頭に追加。
実行後: 実行中に生成されたアイテム（ユーザ入力・アシスタント応答・ツール呼び出し、等）は自動的にセッションに保存。
コンテキストの保持: 同一セッション中は完全な会話履歴が常に含まれ、これによりエージェントがコンテキストを維持。

.to_input_list() を使えば会話履歴を手動管理できるが、セッションを使えば全て自動でやってくれる。

kun432

メモリ操作

セッションには会話履歴を管理するための操作がサポートされている。

基本操作

from agents import SQLiteSession
import asyncio

async def main():
    # SQLiteを使ったセッション管理を初期化
    session = SQLiteSession(
        # セッションIDを指定
        "user_123",
        # セッションを保存するDBファイルパス（指定しない場合はインメモリ）
        "conversations.db"
        # その他、DBのテーブル名なども指定できる
    )

    # 新しいアイテムをセッションに追加
    new_items = [
        {"role": "user", "content": "こんにちは！"},
        {"role": "assistant", "content": "はい、こんにちは！今日はどのようなご要件ですか？"},
        {"role": "user", "content": "東京の天気を教えて。"},
        {"role": "assistant", "content": "東京の天気ですね。今日の東京は一日中快晴が続くでしょう。"},
    ]
    await session.add_items(new_items)

    # セッション中の全てのアイテムを取得
    items = await session.get_items()
    print(items)
    print("-" * 20)

    # 最も最近のアイテムを取得して削除
    last_item = await session.pop_item()
    print(last_item)
    print("-" * 20)

    # セッションから全てのアイテムを削除
    await session.clear_session()
    
    items = await session.get_items()
    print(items)
    print("-" * 20)

if __name__ == "__main__":
    asyncio.run(main())

結果。見やすさのため一部改行を入れている。

出力

[
    {'role': 'user', 'content': 'こんにちは！'},
    {'role': 'assistant', 'content': 'はい、こんにちは！今日はどのようなご要件ですか？'},
    {'role': 'user', 'content': '東京の天気を教えて。'},
    {'role': 'assistant', 'content': '東京の天気ですね。今日の東京は一日中快晴が続くでしょう。'}
]
--------------------
{'role': 'assistant', 'content': '東京の天気ですね。今日の東京は一日中快晴が続くでしょう。'}
--------------------
[]
--------------------

ちなみに上記の削除部分を除外すると、実際のSQLite3データベースには以下のようなデータが入っている。

sqlite3 conversations.db ".tables"

出力

agent_messages  agent_sessions

sqlite3 conversations.db ".schema agent_messages"

出力

CREATE TABLE agent_messages (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                session_id TEXT NOT NULL,
                message_data TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                FOREIGN KEY (session_id) REFERENCES agent_sessions (session_id)
                    ON DELETE CASCADE
            );
CREATE INDEX idx_agent_messages_session_id
            ON agent_messages (session_id, created_at)
        ;

sqlite3 conversations.db ".schema agent_sessions"

出力

CREATE TABLE agent_sessions (
                session_id TEXT PRIMARY KEY,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            );

sqlite3 conversations.db "SELECT * FROM agent_messages"

出力

5|user_123|{"role": "user", "content": "\u3053\u3093\u306b\u3061\u306f\uff01"}|2025-08-16 21:08:43
6|user_123|{"role": "assistant", "content": "\u306f\u3044\u3001\u3053\u3093\u306b\u3061\u306f\uff01\u4eca\u65e5\u306f\u3069\u306e\u3088\u3046\u306a\u3054\u8981\u4ef6\u3067\u3059\u304b\uff1f"}|2025-08-16 21:08:43
7|user_123|{"role": "user", "content": "\u6771\u4eac\u306e\u5929\u6c17\u3092\u6559\u3048\u3066\u3002"}|2025-08-16 21:08:43
8|user_123|{"role": "assistant", "content": "\u6771\u4eac\u306e\u5929\u6c17\u3067\u3059\u306d\u3002\u4eca\u65e5\u306e\u6771\u4eac\u306f\u4e00\u65e5\u4e2d\u5feb\u6674\u304c\u7d9a\u304f\u3067\u3057\u3087\u3046\u3002"}|2025-08-16 21:08:43

sqlite3 conversations.db "SELECT message_data FROM agent_messages" | jq -r .

出力

{
  "role": "user",
  "content": "こんにちは！"
}
{
  "role": "assistant",
  "content": "はい、こんにちは！今日はどのようなご要件ですか？"
}
{
  "role": "user",
  "content": "東京の天気を教えて。"
}
{
  "role": "assistant",
  "content": "東京の天気ですね。今日の東京は一日中快晴が続くでしょう。"
}

sqlite3 conversations.db "SELECT * FROM agent_sessions"

出力

user_123|2025-08-16 21:08:43|2025-08-16 21:08:43

修正のための `pop_item` の使用

上のサンプルコードでも使用した、会話履歴の最後のアイテムを取り出して削除する pop_itemを使うと、会話を取り消したり、やり直したい場合などに使える。

from agents import Agent, Runner, SQLiteSession
import asyncio
import json

agent = Agent(
    name="Assistant",
    instructions="日本語で簡潔に回答してください。",
)

session = SQLiteSession("correction_example")

async def main():
    # 最初の会話
    result = await Runner.run(
        agent,
        "2 + 2 は？",
        session=session
    )
    print(f"ユーザ: {result.input[0]['content']}")
    print(f"エージェント: {result.final_output}")

    print("-" * 10, "最初の会話履歴", "-" * 10)
    items = await session.get_items()
    print(json.dumps(items, indent=2, ensure_ascii=False))
    print("-" * 20)

    # ユーザが質問を修正したいとする
    assistant_item = await session.pop_item()  # エージェントの回答を削除
    user_item = await session.pop_item()  # ユーザの質問を削除

    # 修正した質問を再度送信
    result = await Runner.run(
        agent,
        "2 + 3 は？",
        session=session
    )
    print(f"ユーザ: {result.input[0]['content']}")
    print(f"エージェント: {result.final_output}")

    print("-" * 10, "修正後の会話履歴", "-" * 10)
    items = await session.get_items()
    print(json.dumps(items, indent=2, ensure_ascii=False))
    print("-" * 20)

if __name__ == "__main__":
    asyncio.run(main())

出力

ユーザ: 2 + 2 は？
エージェント: 4です。
---------- 最初の会話履歴 ----------
[
  {
    "content": "2 + 2 は？",
    "role": "user"
  },
  {
    "id": "msg_68a0f99ea5188199bcc464927d2f17d807ce076614673d4e",
    "content": [
      {
        "annotations": [],
        "text": "4です。",
        "type": "output_text",
        "logprobs": []
      }
    ],
    "role": "assistant",
    "status": "completed",
    "type": "message"
  }
]
--------------------
ユーザ: 2 + 3 は？
エージェント: 5です。
---------- 修正後の会話履歴 ----------
[
  {
    "content": "2 + 3 は？",
    "role": "user"
  },
  {
    "id": "msg_68a0f99fc1708199b672d98bb323a69e09d4cd00072f21b7",
    "content": [
      {
        "annotations": [],
        "text": "5です。",
        "type": "output_text",
        "logprobs": []
      }
    ],
    "role": "assistant",
    "status": "completed",
    "type": "message"
  }
]
--------------------

kun432

メモリオプション

メモリの使用は３パターン。

1. メモリなし（デフォルト）

from agents import Agent, Runner, SQLiteSession
import asyncio

agent = Agent(
    name="Assistant",
    instructions="日本語で簡潔に回答してください。",
)

async def main():
    # デフォルトはセッションメモリなし
    result = await Runner.run(agent, "こんにちは！私の趣味は競馬なんですよ！")
    print(result.final_output)

    print("-" * 20)
    
    result = await Runner.run(agent, "私の趣味はなんでしたっけ？")
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())

出力

こんにちは！競馬がお好きなんですね。どのレースや馬が特にお気に入りですか？
--------------------
ごめんなさい、あなたの趣味はわかりません。好きなことを教えてもらえますか？

2. SQLiteメモリ

from agents import Agent, Runner, SQLiteSession
import asyncio

agent = Agent(
    name="Assistant",
    instructions="日本語で簡潔に回答してください。",
)

async def main():
    # ファイルパスを指定しなければ、インメモリデータベースを使用（プロセス終了後にデータが失われる）
    session = SQLiteSession("session_2")

    result = await Runner.run(
        agent,
        "こんにちは！私の趣味は競馬なんですよ！",
        session=session
    )
    print(result.final_output)
   
    print("-" * 20)

    result = await Runner.run(
        agent,
        "私の趣味はなんでしたっけ？",
        session=session
    )
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())

出力

こんにちは！競馬はエキサイティングですよね。お気に入りの馬やレースはありますか？
--------------------
あなたの趣味は競馬です。

3. 複数セッション

from agents import Agent, Runner, SQLiteSession
import asyncio

agent = Agent(
    name="Assistant",
    instructions="日本語で簡潔に回答してください。",
)

async def main():
    # 異なるセッションの場合は異なる会話履歴を保持する
    session_1 = SQLiteSession("user_123", "conversations.db")
    session_2 = SQLiteSession("user_456", "conversations.db")

    print("=" * 10, "セッション1", "=" * 10)
    result = await Runner.run(
        agent,
        "こんにちは！私の趣味は競馬なんですよ！",
        session=session_1
    )
    print(result.final_output)
   
    print("-" * 20)

    result = await Runner.run(
        agent,
        "私の趣味はなんでしたっけ？",
        session=session_1
    )
    print(result.final_output)

    print("=" * 10, "セッション2", "=" * 10)

    result = await Runner.run(
        agent,
        "私の趣味はなんでしたっけ？",
        session=session_2
    )
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())

出力

========== セッション1 ==========
こんにちは！競馬が趣味なんですね。どのレースがお気に入りですか？
--------------------
あなたの趣味は競馬ですね。
========== セッション2 ==========
すみません、あなたの趣味についての情報は持っていません。興味のあるものを教えていただければ嬉しいです。

kun432

カスタムメモリ実装

よく耳にするような主要LLMエージェントフレームワークとは違って、Agents SDKの公式のメモリ実装は現状SQLiteぐらいしかない。

何かしら別のストレージを使いたい、とか、独自のセッション・メモリ管理を実装したい場合は、Sessionプロトコルに従ったクラスを実装することになる。

from agents import Agent, Runner 
from agents.memory import Session
from typing import List
import asyncio

class MyCustomSession:
    """Sessionプロトコルに従った、カスタムセッションの実装"""

    def __init__(self, session_id: str):
        self.session_id = session_id
        # 初期化処理を書く

    async def get_items(self, limit: int | None = None) -> List[dict]:
        """このセッションの会話履歴を取得する"""
        # 実装を書く
        pass

    async def add_items(self, items: List[dict]) -> None:
        """このセッションに新しいアイテムを追加する"""
        # 実装を書く
        pass

    async def pop_item(self) -> dict | None:
        """このセッションの最も最近のアイテムを取得して削除する"""
        # 実装を書く
        pass

    async def clear_session(self) -> None:
        """このセッションのすべてのアイテムを削除する"""
        # 実装を書く
        pass

agent = Agent(
    name="Assistant",
    instructions="日本語で簡潔に回答してください。",
)

async def main():
    session = MyCustomSession("my_custom_session_1")

    result = await Runner.run(
        agent,
        "こんにちは！",
        session=session
    )
    print(result.final_output)

if __name__ == "__main__":
    asyncio.run(main())

kun432

 セッション管理セッション管理におけるベストプラクティスっぽいまとめになっている。

 セッション ID の命名会話を管理する上で、意味のあるセッション ID を使用すること
ユーザー単位: "user_12345"
スレッド単位: "thread_abc123"
コンテキスト単位: "support_ticket_456"

 メモリの永続化一時的な会話には、インメモリ SQLite（SQLiteSession("session_id")）を使用
永続的な会話には、ファイルベースの SQLite（SQLiteSession("session_id", "path/to/db.sqlite")）を使用
本番環境の場合には、カスタムセッションバックエンド（Redis、PostgreSQL など）の実装を検討

 セッション管理の例(snip)
support_agent = Agent(
    name="support agent",
)
billing_agent = Agent(
    name="billing agent",
)

async def main():
    # 異なるエージェントで同一セッションを利用可能
    session = SQLiteSession("user_123")

    # 会話を新規に開始する場合にはセッションをクリアする
    await session.clear_session()
    
    # 複数のエージェントが同じ会話履歴を参照することになる
    result = await Runner.run(
        support_agent,
        "私のアカウントについて調べてほしい。",
        session=session
    )
   
    result = await Runner.run(
        billing_agent,
        "料金について教えてほしい。",
        session=session
    )
(snip)

kun432

次はツールの予定

kun432

なお、PostgreSQLやRedisなど外部のストレージを使う場合は、以下が参考になる
Redis
https://zenn.dev/lancers/articles/95a30d5aa210aa
https://github.com/rafaelpierre/openai-agents-redis
以下はマージ待ちのPR
MongoDB
https://github.com/openai/openai-agents-python/pull/1364
PostgreSQL
https://github.com/openai/openai-agents-python/pull/1163
SQLAlchemy
https://github.com/openai/openai-agents-python/pull/1357

kun432

Agents SDK自身は長期メモリを昨日としては提供していない。
サードパーティとのインテグレーションには以下のようなものがある。
https://blog.getzep.com/building-a-memory-agent-with-the-openai-agents-sdk-and-zep/
https://docs.mem0.ai/integrations/openai-agents-sdk
AsyncOpenAIクライアントを用意すればできそうな気がする
https://docs.memobase.io/templates/openai
まあ長期メモリの場合はツールとして検索するのが良さそうには思うので、都度のチャットメッセージを良しなにメモリレイヤーに登録すればいい気はする。

このスクラップは21日前にクローズされました

セッション

クイックスタート

仕組み

メモリ操作

基本操作

修正のための pop_item の使用

メモリオプション

1. メモリなし（デフォルト）

2. SQLiteメモリ

3. 複数セッション

カスタムメモリ実装

セッション管理

セッション ID の命名

メモリの永続化

セッション管理の例

修正のための `pop_item` の使用