Sessions
Sessions (세션)
Agents SDK는 여러 에이전트 실행에 걸쳐 대화 기록을 자동으로 유지하는 내장 세션 메모리를 제공해서, 턴 사이에 .to_input_list()를 수동으로 처리할 필요를 없애줘요.
세션은 특정 세션의 대화 기록을 저장하고, 에이전트가 명시적 수동 메모리 관리 없이도 컨텍스트를 유지하게 해줘요. 에이전트가 이전 상호작용을 기억하길 원하는 채팅 앱이나 다중 턴 대화를 만들 때 특히 유용해요.
SDK가 클라이언트 측 메모리를 관리해 주길 원할 때 세션을 쓰세요. 같은 실행에서 세션을 실행 수준 연속 옵션인 conversation_id, previous_response_id, auto_previous_response_id와 결합할 수는 없어요. OpenAI 서버 관리 연속화를 원한다면 세션을 그 위에 겹쳐 쓰지 말고 그 메커니즘 중 하나를 고르세요.
출처: 문서
본문
빠른 시작
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"
같은 세션으로 중단된 실행 재개
실행이 승인을 위해 일시 중지되면 같은 세션 인스턴스(또는 같은 세션 ID와 같은 기본 저장소 백엔드로 구성된 다른 인스턴스)로 재개해서, 재개된 턴이 같은 저장된 대화 기록을 계속하세요.
result = await Runner.run(agent, "Delete temporary files that are no longer needed.", session=session)
if result.interruptions:
state = result.to_state()
for interruption in result.interruptions:
state.approve(interruption)
result = await Runner.run(agent, state, session=session)
핵심 세션 동작
세션 메모리가 켜져 있으면:
- 각 실행 전: 러너가 자동으로 세션의 대화 기록을 검색해 입력 항목 앞에 붙여요.
- 각 실행 후: 실행 중 생성된 모든 새 항목(사용자 입력·어시스턴트 응답·도구 호출 등)이 자동으로 세션에 저장돼요.
- 컨텍스트 보존: 같은 세션으로 하는 각 후속 실행이 전체 대화 기록을 포함해, 에이전트가 컨텍스트를 유지하게 해요.
이것으로 실행 사이에 .to_input_list()를 수동 호출하고 대화 상태를 관리할 필요가 없어져요.
기록과 새 입력이 병합되는 방식 제어
세션을 넘기면 러너는 보통 모델 입력을 이렇게 준비해요.
- 세션 기록(
session.get_items(...)에서 검색) - 새 턴 입력
RunConfig.session_input_callback으로 모델 호출 전에 그 병합 단계를 커스터마이즈하세요. 콜백은 두 목록을 받아요.
history— 검색된 세션 기록(이미 input-item 형식으로 정규화됨)new_input— 현재 턴의 새 입력 항목
모델로 보내야 할 최종 입력 항목 목록을 반환하세요.
콜백은 두 목록의 복사본을 받으므로 안전하게 변형할 수 있어요. 반환된 목록이 그 턴의 모델 입력을 제어하지만, SDK는 여전히 새 턴에 속한 항목만 저장해요. 따라서 오래된 기록을 재정렬·필터링해도 오래된 세션 항목이 새 입력으로 다시 저장되지는 않아요.
from agents import Agent, RunConfig, Runner, SQLiteSession
def keep_recent_history(history, new_input):
# Keep only the last 10 history items, then append the new turn.
return history[-10:] + new_input
agent = Agent(name="Assistant")
session = SQLiteSession("conversation_123")
result = await Runner.run(
agent,
"Continue from the latest updates only.",
session=session,
run_config=RunConfig(session_input_callback=keep_recent_history),
)
세션이 항목을 저장하는 방식을 바꾸지 않고 커스텀 가지치기·재정렬·선택적 포함이 필요할 때 쓰세요. 모델 호출 직전에 나중 최종 패스가 필요하면 running agents 가이드의 call_model_input_filter를 쓰세요.
검색 기록 제한
SessionSettings로 각 실행 전에 가져올 기록의 양을 제어하세요.
SessionSettings(limit=None)(기본) — 사용 가능한 모든 세션 항목 검색SessionSettings(limit=N)— 가장 최근 N개 항목만 검색
RunConfig.session_settings로 실행별 적용할 수 있어요.
from agents import Agent, RunConfig, Runner, SessionSettings, SQLiteSession
agent = Agent(name="Assistant")
session = SQLiteSession("conversation_123")
result = await Runner.run(
agent,
"Summarize our recent discussion.",
session=session,
run_config=RunConfig(session_settings=SessionSettings(limit=50)),
)
세션 구현이 기본 세션 설정을 노출하면, RunConfig.session_settings의 None이 아닌 각 값이 그 실행의 해당 기본값을 재정의해요. 세션의 기본 동작을 바꾸지 않고 검색 크기를 제한하려는 긴 대화에서 유용해요.
메모리 연산
기본 연산
세션은 대화 기록 관리를 위한 몇 가지 연산을 지원해요.
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}")
내장 세션 구현
SDK는 다양한 유스케이스를 위한 여러 세션 구현을 제공해요.
내장 세션 구현 고르기
상세 예제를 읽기 전에 시작점을 고르는 데 이 표를 쓰세요.
| 세션 타입 | 가장 적합한 용도 | 메모 |
|---|---|---|
SQLiteSession |
로컬 개발과 단순 앱 | 내장, 가볍고, 파일 기반 또는 인메모리 |
AsyncSQLiteSession |
aiosqlite를 쓰는 async SQLite | async 드라이버 지원 확장 백엔드 |
RedisSession |
워커·서비스 간 공유 메모리 | 저지연 분산 배포에 좋음 |
SQLAlchemySession |
기존 데이터베이스가 있는 프로덕션 앱 | SQLAlchemy 지원 데이터베이스와 동작 |
MongoDBSession |
MongoDB를 이미 쓰거나 다중 프로세스 저장이 필요한 앱 | async pymongo; 순서용 원자 시퀀스 카운터 |
DaprSession |
Dapr 사이드카가 있는 클라우드 네이티브 배포 | 여러 상태 저장소 + TTL·일관성 제어 지원 |
OpenAIConversationsSession |
OpenAI의 서버 관리 저장 | OpenAI Conversations API 기반 기록 |
OpenAIResponsesCompactionSession |
자동 압축이 있는 긴 대화 | 다른 세션 백엔드 위의 래퍼 |
AdvancedSQLiteSession |
분기/분석이 있는 SQLite | 더 무거운 기능 집합; 전용 페이지 참고 |
EncryptedSession |
다른 세션 위의 암호화 + TTL | 래퍼; 먼저 기본 백엔드를 고르세요 |
일부 구현은 추가 세부 사항이 있는 전용 페이지가 있고, 해당 하위 섹션에 인라인 링크가 있어요.
ChatKit용 Python 서버를 구현한다면, ChatKit의 스레드·항목 영속화에는 chatkit.store.Store 구현을 쓰세요. SQLAlchemySession 같은 Agents SDK 세션은 SDK 측 대화 기록을 관리하지만 ChatKit의 store를 대체하는 드롭인은 아니에요. ChatKit 데이터 저장소 구현은 chatkit-python 가이드를 참고하세요.
OpenAI Conversations API 세션
OpenAIConversationsSession으로 OpenAI의 Conversations API를 쓰세요.
from agents import Agent, Runner, 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,
"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"
OpenAI Responses 압축 세션
OpenAIResponsesCompactionSession을 사용해 Responses API(responses.compact)로 저장된 대화 기록을 압축하세요. 기본 세션을 감싸고 should_trigger_compaction에 따라 각 턴 뒤에 자동으로 압축할 수 있어요. 이것으로 OpenAIConversationsSession을 감싸지 마세요. 두 기능은 기록을 다른 방식으로 관리해요.
일반적 사용 (자동 압축)
from agents import Agent, Runner, SQLiteSession
from agents.memory import OpenAIResponsesCompactionSession
underlying = SQLiteSession("conversation_123")
session = OpenAIResponsesCompactionSession(
session_id="conversation_123",
underlying_session=underlying,
)
agent = Agent(name="Assistant")
result = await Runner.run(agent, "Hello", session=session)
print(result.final_output)
기본적으로 각 턴 뒤에 SDK는 압축 후보가 임계값을 충족하는지 확인하고 충족할 때만 압축해요.
자동 압축이 실행되면 SDK는 Runner.run(...)이 반환하거나 스트리밍 이벤트 반복자가 닫히기 전에 그것을 기다려요. 압축 요청이 보고한 사용량은 그 실행의 Usage 합계에 기여해요. 기본적으로 나중에 만드는 수동 run_compaction() 호출은 둘러싸는 실행 컨텍스트가 없어 완료된 실행의 사용량 객체를 갱신하지 않아요.
compaction_mode="previous_response_id"는 압축 세션이 유지한 Responses API 응답 ID를 사용하고, 그 응답 체인이 사용 가능한 동안 가장 잘 동작해요. compaction_mode="input"은 대신 현재 세션 항목에서 압축 요청을 재구성하는데, 응답 체인을 사용할 수 없거나 세션 내용이 진실의 원천이길 원할 때 유용해요. 기본 "auto"는 가장 안전한 가용 옵션을 고릅니다.
에이전트가 ModelSettings(store=False)로 실행되면 Responses API는 나중 조회를 위해 마지막 응답을 유지하지 않아요. 그 상태 비저장 설정에서 기본 "auto" 모드는 previous_response_id에 의존하는 대신 입력 기반 압축으로 폴백해요. 완전한 예제는 examples/memory/compaction_session_stateless_example.py를 참고하세요.
자동 압축이 스트리밍을 막을 수 있음
압축은 세션 기록을 지우고 다시 쓰므로, SDK는 실행이 완료된 것으로 간주하기 전에 압축이 끝나길 기다려요. 스트리밍 모드에서는 압축이 무거우면 마지막 출력 토큰 뒤 몇 초 동안 run.stream_events()가 열려 있을 수 있어요.
OpenAIResponsesCompactionSession.run_compaction()은 clear-and-rewrite 연산을 래퍼 경계에서 복구 가능한 교체로 취급해요. 기본 기록이 바뀐 뒤 교체가 실패하거나 취소되면, 래퍼는 이전 기록을 복원하려 시도하고 원래 예외나 취소가 호출자에게 도달하기 전에 그 복구 시도가 정리되길 기다려요. 기본 백엔드도 복구 중 실패하면 이전 기록이 복원되지 않은 채 남을 수 있고 SDK가 복구 실패를 로그로 남겨요. 래퍼는 add_items(), pop_item(), clear_session()을 원격 요청·교체·복구 단계를 포함한 전체 압축 연산과 직렬화해요. 동시 래퍼 변형은 압축을 기다렸다가 덮어쓰는 대신 결과 기록에 적용돼요. 자동 압축은 또한 어떤 래퍼 세대가 그 실행에 속하는지 기록해요. 다른 실행이 그 압축이 시작되기 전에 기록을 바꾸면 SDK는 더 새로운 기록을 교체하는 대신 오래된 압축을 건너뛰어요. 압축이 실행되는 동안 기본 세션을 직접 변형하지 마세요. 직접 변형은 이 래퍼 보장을 우회하거든요.
저지연 스트리밍이나 빠른 턴 테이킹이 필요하면 자동 압축을 끄고 턴 사이에(또는 유휴 시간에) run_compaction()을 직접 호출하세요. 자신의 기준으로 강제 압축 시점을 결정할 수 있어요.
from agents import Agent, Runner, SQLiteSession
from agents.memory import OpenAIResponsesCompactionSession
underlying = SQLiteSession("conversation_123")
session = OpenAIResponsesCompactionSession(
session_id="conversation_123",
underlying_session=underlying,
# Disable triggering the auto compaction
should_trigger_compaction=lambda _: False,
)
agent = Agent(name="Assistant")
result = await Runner.run(agent, "Hello", session=session)
# Decide when to compact (e.g., on idle, every N turns, or size thresholds).
await session.run_compaction({"force": True})
SQLite 세션
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
)
Async SQLite 세션
aiosqlite로 백업되는 SQLite 영속화를 원할 때 AsyncSQLiteSession을 쓰세요.
pip install aiosqlite
from agents import Agent, Runner
from agents.extensions.memory import AsyncSQLiteSession
agent = Agent(name="Assistant")
session = AsyncSQLiteSession("user_123", db_path="conversations.db")
result = await Runner.run(agent, "Hello", session=session)
Redis 세션
여러 워커·서비스 간 공유 세션 메모리에는 RedisSession을 쓰세요.
pip install openai-agents[redis]
from agents import Agent, Runner
from agents.extensions.memory import RedisSession
agent = Agent(name="Assistant")
session = RedisSession.from_url(
"user_123",
url="redis://localhost:6379/0",
)
result = await Runner.run(agent, "Hello", session=session)
await session.close()
from_url(...)은 Redis 클라이언트를 만들고 소유해요. close() 후에는 세션이 터미널이고 이후 세션 연산은 RuntimeError를 발생시켜요. 반복적·동시 close() 호출은 안전해요. 애플리케이션이 이미 Redis 클라이언트를 관리한다면 redis_client=...으로 RedisSession(...)을 직접 구성하세요. 그 경우 close()는 no-op이고 호출자가 클라이언트 소유권과 세션 사용성을 모두 유지해요.
SQLAlchemy 세션
SQLAlchemy가 지원하는 데이터베이스를 쓰는 프로덕션 준비된 Agents SDK 세션 영속화:
from agents.extensions.memory import SQLAlchemySession
# Using database URL
session = SQLAlchemySession.from_url(
"user_123",
url="postgresql+asyncpg://user:pass@localhost/db",
create_tables=True
)
# Using existing engine
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
session = SQLAlchemySession("user_123", engine=engine, create_tables=True)
상세 문서는 SQLAlchemy Sessions를 참고하세요.
Dapr 세션
이미 Dapr 사이드카를 실행하거나 에이전트 코드를 바꾸지 않고 구성된 state-store 백엔드를 전환하길 원할 때 DaprSession을 쓰세요.
pip install openai-agents[dapr]
from agents import Agent, Runner
from agents.extensions.memory import DaprSession
agent = Agent(name="Assistant")
async with DaprSession.from_address(
"user_123",
state_store_name="statestore",
dapr_address="localhost:50001",
) as session:
result = await Runner.run(agent, "Hello", session=session)
print(result.final_output)
메모:
from_address(...)은 Dapr 클라이언트를 만들어 소유해요. 앱이 이미 관리한다면dapr_client=...으로DaprSession(...)을 직접 구성하세요.- 컨텍스트를 나가거나
close()를 호출하면 소유 클라이언트 세션이 터미널이 되고, 이후 세션 연산은RuntimeError를 발생시키며 반복적·동시close()호출은 안전해요. 주입된 클라이언트를 쓰면close()는 no-op이고 세션은 계속 사용 가능해요. - 백킹 상태 저장소가 TTL을 지원하면
ttl=...을 넘겨 세션 데이터에 TTL 만료를 자동 적용하세요. - 상태 쓰기·삭제에 강한 일관성을 요청하려면
consistency=DAPR_CONSISTENCY_STRONG을 넘기세요(저장소 의존). 와이어 수준 읽기 일관성은get_state에 일관성을 설정하는 상위 Dapr Python 클라이언트 지원이 필요해요. - Dapr Python SDK는 HTTP 사이드카 엔드포인트도 확인해요. 로컬 개발에서는
dapr_address에 쓴 gRPC 포트뿐 아니라--dapr-http-port 3500으로 Dapr을 시작하세요. - 로컬 컴포넌트와 문제 해결을 포함한 전체 설정 워크스루는
examples/memory/dapr_session_example.py를 참고하세요.
MongoDB 세션
이미 MongoDB를 쓰거나 수평 확장·다중 프로세스 세션 저장이 필요한 애플리케이션에는 MongoDBSession을 쓰세요.
pip install openai-agents[mongodb]
from agents import Agent, Runner
from agents.extensions.memory import MongoDBSession
agent = Agent(name="Assistant")
# Create from URI — owns the client and closes it when session.close() is called
session = MongoDBSession.from_uri(
"user-123",
uri="mongodb://localhost:27017",
database="agents",
)
result = await Runner.run(agent, "Hello", session=session)
print(result.final_output)
await session.close()
메모:
from_uri(...)은AsyncMongoClient를 만들고session.close()에서 닫아요. 소유 클라이언트 세션은close()후 터미널이고, 이후 세션 연산은RuntimeError를 발생시켜요. 애플리케이션이 이미 클라이언트를 관리한다면client=...으로MongoDBSession(...)을 직접 구성하세요. 그 경우session.close()는 no-op이고, 호출자가 클라이언트 lifecycle 책임을 유지하며 세션은 계속 사용 가능해요.- MongoDB Atlas에 연결하려면 다른 변경 없이
from_uri(...)에mongodb+srv://user:***@cluster.example.mongodb.netURI를 넘기세요. - 두 컬렉션이 쓰이고 두 이름 모두
sessions_collection=(기본agent_sessions)과messages_collection=(기본agent_messages)으로 구성 가능해요. 인덱스는 첫 사용 시 자동 생성돼요. 비어 있지 않은 각add_items()호출은 하나의 논리적 배치 문서를 쓰는데, 단조 증가하는seq가 최종 항목으로 배치를 순서화해요. 레거시 항목별 메시지 문서는 여전히 읽을 수 있어요. 논리적 배치 하나가 MongoDB의 단일 문서 크기 한계 안에 들어맞아야 해요. 초대형 배치는 부분 배치를 저장하지 않고 원자적으로 실패해요. - 첫 실행 전에
await session.ping()으로 연결을 확인하세요.
고급 SQLite 세션
대화 분기·사용량 분석·구조화 쿼리가 있는 향상된 SQLite 세션:
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
상세 문서는 Advanced SQLite Sessions를 참고하세요.
암호화 세션
모든 세션 구현을 위한 투명 암호화 래퍼:
from agents.extensions.memory import EncryptedSession, SQLAlchemySession
# 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)
상세 문서는 Encrypted Sessions를 참고하세요.
다른 세션 타입
몇 가지 내장 옵션이 더 있어요. examples/memory/와 extensions/memory/ 아래의 소스 코드를 참고하세요.
운영 패턴
세션 ID 명명
대화를 조직하는 데 도움이 되는 의미 있는 세션 ID를 쓰세요.
- 사용자 기반:
"user_12345" - 스레드 기반:
"thread_abc123" - 컨텍스트 기반:
"support_ticket_456"
메모리 영속화
- 임시 대화에는 인메모리 SQLite(
SQLiteSession("session_id"))를 쓰세요. - 영속 대화에는 파일 기반 SQLite(
SQLiteSession("session_id", "path/to/db.sqlite"))를 쓰세요. - aiosqlite 기반 구현이 필요하면 async SQLite(
AsyncSQLiteSession("session_id", db_path="..."))를 쓰세요. - 공유·저지연 세션 메모리에는 Redis 백업 세션(
RedisSession.from_url("session_id", url="redis://..."))을 쓰세요. - SQLAlchemy가 지원하는 기존 데이터베이스가 있는 프로덕션 시스템에는 SQLAlchemy 기반 세션(
SQLAlchemySession("session_id", engine=engine, create_tables=True))을 쓰세요. - MongoDB를 이미 쓰거나 다중 프로세스·수평 확장 세션 저장이 필요한 애플리케이션에는 MongoDB 세션(
MongoDBSession.from_uri("session_id", uri="mongodb://localhost:27017"))을 쓰세요. - 내장 텔레메트리·tracing·데이터 격리와 30+ 데이터베이스 백엔드 지원이 있는 프로덕션 클라우드 네이티브 배포에는 Dapr 상태 저장소 세션(
DaprSession.from_address("session_id", state_store_name="statestore", dapr_address="localhost:50001"))을 쓰세요. - OpenAI Conversations API에 기록을 저장하길 선호한다면 OpenAI 호스팅 저장(
OpenAIConversationsSession())을 쓰세요. - 모든 세션을 투명 암호화·TTL 기반 만료로 감싸려면 암호화 세션(
EncryptedSession(session_id, underlying_session, encryption_key))을 쓰세요. - 더 고급 유스케이스를 위해 다른 프로덕션 시스템(예: Django)의 커스텀 세션 백엔드 구현을 고려하세요.
여러 세션
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
)
세션 공유
# 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())
커스텀 세션 구현
Session 프로토콜을 구조적으로 따르는 클래스를 만들어 자체 세션 메모리를 구현할 수 있어요. SessionABC에서 상속할 필요는 없고, session_id와 session_settings를 정의하고 네 가지 기록 메서드를 직접 구현하면 돼요.
from agents import Agent, Runner, SessionSettings
from agents.items import TResponseInputItem
class MyCustomSession:
"""Custom session implementation following the Session protocol."""
session_settings: SessionSettings | None = None
def __init__(self, session_id: str) -> None:
self.session_id = session_id
self.items: list[TResponseInputItem] = []
async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]:
if limit is None:
return list(self.items)
if limit <= 0:
return []
return list(self.items[-limit:])
async def add_items(self, items: list[TResponseInputItem]) -> None:
self.items.extend(items)
async def pop_item(self) -> TResponseInputItem | None:
return self.items.pop() if self.items else None
async def clear_session(self) -> None:
self.items.clear()
# Use your custom session
agent = Agent(name="Assistant")
result = await Runner.run(
agent,
"Hello",
session=MyCustomSession("my_session")
)
커스텀 세션에서 실행 컨텍스트 접근
Agents SDK는 테넌트 라우팅·인증·다른 앱별 저장 결정을 위해 활성 RunContextWrapper를 커스텀 세션에 전달할 수 있어요. SDK가 래퍼를 전달하게 하려면 네 가지 기록 메서드 모두에 명시적으로 명명된, 키워드 호환 wrapper 파라미터를 추가하세요.
from typing import Any
from agents import RunContextWrapper
from agents.items import TResponseInputItem
class ContextAwareSession:
async def get_items(
self,
limit: int | None = None,
*,
wrapper: RunContextWrapper[Any] | None = None,
) -> list[TResponseInputItem]: ...
async def add_items(
self,
items: list[TResponseInputItem],
*,
wrapper: RunContextWrapper[Any] | None = None,
) -> None: ...
async def pop_item(
self,
*,
wrapper: RunContextWrapper[Any] | None = None,
) -> TResponseInputItem | None: ...
async def clear_session(
self,
*,
wrapper: RunContextWrapper[Any] | None = None,
) -> None: ...
Agents SDK는 get_items, add_items, pop_item, clear_session 모두가 wrapper를 선언할 때만 이 통합을 켜요. 일반 **kwargs 파라미터는 이 시그니처 검사를 충족하지 않아요. wrapper를 생략한 기존 세션 구현은 출시된 호출 형태를 유지하고 변경 없이 계속 동작해요.
커뮤니티 세션 구현
커뮤니티가 추가 세션 구현을 개발했어요.
| 패키지 | 설명 |
|---|---|
| openai-django-sessions | Django가 지원하는 모든 데이터베이스(PostgreSQL, MySQL, SQLite 등)용 Django ORM 기반 세션 |
세션 구현을 만들었다면 문서 PR을 제출해 여기에 추가해 주세요.
API 레퍼런스
상세 API 문서는 다음을 참고하세요.
Session— 프로토콜 인터페이스OpenAIConversationsSession— OpenAI Conversations API 구현OpenAIResponsesCompactionSession— Responses API 압축 래퍼SQLiteSession— 기본 SQLite 구현AsyncSQLiteSession— aiosqlite 기반 async SQLite 구현RedisSession— Redis 백업 세션 구현SQLAlchemySession— SQLAlchemy 기반 구현MongoDBSession— MongoDB 백업 세션 구현DaprSession— Dapr 상태 저장소 구현AdvancedSQLiteSession— 분기·분석이 있는 향상된 SQLiteEncryptedSession— 모든 세션용 암호화 래퍼
더 알아보기 (Learn more)
- OpenAI Agents SDK 문서에서 더 많은 가이드를 확인하세요.