fromagentsimportAgent,Runner,SQLiteSession# Create agentagent=Agent(name="Assistant",instructions="Reply very concisely.",)# Create a session instance with a session IDsession=SQLiteSession("conversation_123")# First turnresult=awaitRunner.run(agent,"What city is the Golden Gate Bridge in?",session=session)print(result.final_output)# "San Francisco"# Second turn - agent automatically remembers previous contextresult=awaitRunner.run(agent,"What state is it in?",session=session)print(result.final_output)# "California"# Also works with synchronous runnerresult=Runner.run_sync(agent,"What's the population?",session=session)print(result.final_output)# "Approximately 39 million"

fromagentsimportSQLiteSessionsession=SQLiteSession("user_123","conversations.db")# Get all items in a sessionitems=awaitsession.get_items()# Add new items to a sessionnew_items=[{"role":"user","content":"Hello"},{"role":"assistant","content":"Hi there!"}]awaitsession.add_items(new_items)# Remove and return the most recent itemlast_item=awaitsession.pop_item()print(last_item)# {"role": "assistant", "content": "Hi there!"}# Clear all items from a sessionawaitsession.clear_session()

fromagentsimportAgent,Runner,SQLiteSessionagent=Agent(name="Assistant")session=SQLiteSession("correction_example")# Initial conversationresult=awaitRunner.run(agent,"What's 2 + 2?",session=session)print(f"Agent:{result.final_output}")# User wants to correct their questionassistant_item=awaitsession.pop_item()# Remove agent's responseuser_item=awaitsession.pop_item()# Remove user's question# Ask a corrected questionresult=awaitRunner.run(agent,"What's 2 + 3?",session=session)print(f"Agent:{result.final_output}")

# Default behavior - no session memoryresult=awaitRunner.run(agent,"Hello")

fromagentsimportOpenAIConversationsSessionsession=OpenAIConversationsSession()# Optionally resume a previous conversation by passing a conversation ID# session = OpenAIConversationsSession(conversation_id="conv_123")result=awaitRunner.run(agent,"Hello",session=session,)

fromagentsimportSQLiteSession# In-memory database (lost when process ends)session=SQLiteSession("user_123")# Persistent file-based databasesession=SQLiteSession("user_123","conversations.db")# Use the sessionresult=awaitRunner.run(agent,"Hello",session=session)

fromagentsimportAgent,Runner,SQLiteSessionagent=Agent(name="Assistant")# Different sessions maintain separate conversation historiessession_1=SQLiteSession("user_123","conversations.db")session_2=SQLiteSession("user_456","conversations.db")result1=awaitRunner.run(agent,"Hello",session=session_1)result2=awaitRunner.run(agent,"Hello",session=session_2)

importasynciofromagentsimportAgent,Runnerfromagents.extensions.memory.sqlalchemy_sessionimportSQLAlchemySessionasyncdefmain():agent=Agent("Assistant")session=SQLAlchemySession.from_url("user-123",url="sqlite+aiosqlite:///:memory:",create_tables=True,# Auto-create tables for the demo)result=awaitRunner.run(agent,"Hello",session=session)if__name__=="__main__":asyncio.run(main())

importasynciofromagentsimportAgent,Runnerfromagents.extensions.memory.sqlalchemy_sessionimportSQLAlchemySessionfromsqlalchemy.ext.asyncioimportcreate_async_engineasyncdefmain():# In your application, you would use your existing engineengine=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=awaitRunner.run(agent,"Hello",session=session)print(result.final_output)awaitengine.dispose()if__name__=="__main__":asyncio.run(main())

importasynciofromagentsimportAgent,Runnerfromagents.extensions.memoryimportEncryptedSession,SQLAlchemySessionasyncdefmain():# 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 expirationsession=EncryptedSession(session_id="user-123",underlying_session=underlying_session,encryption_key="your-encryption-key",# Use a secure key from your secrets managementttl=600,# 10 minutes - items older than this are silently skipped)agent=Agent("Assistant")result=awaitRunner.run(agent,"Hello",session=session)print(result.final_output)if__name__=="__main__":asyncio.run(main())

重要的安全注意事项

• 安全存储你的加密密钥（如环境变量、密钥管理服务）
• 过期令牌根据应用服务的系统时钟被拒绝——请确保所有服务均通过 NTP 同步时间，以避免因时钟漂移导致的误拒
• 底层会话仍存储加密数据，因此你依然可以掌控你的数据库基础设施

fromagents.memory.sessionimportSessionABCfromagents.itemsimportTResponseInputItemfromtypingimportListclassMyCustomSession(SessionABC):"""Custom session implementation following the Session protocol."""def__init__(self,session_id:str):self.session_id=session_id# Your initialization hereasyncdefget_items(self,limit:int|None=None)->List[TResponseInputItem]:"""Retrieve conversation history for this session."""# Your implementation herepassasyncdefadd_items(self,items:List[TResponseInputItem])->None:"""Store new items for this session."""# Your implementation herepassasyncdefpop_item(self)->TResponseInputItem|None:"""Remove and return the most recent item from this session."""# Your implementation herepassasyncdefclear_session(self)->None:"""Clear all items for this session."""# Your implementation herepass# Use your custom sessionagent=Agent(name="Assistant")result=awaitRunner.run(agent,"Hello",session=MyCustomSession("my_session"))

# Clear a session when conversation should start freshawaitsession.clear_session()# Different agents can share the same sessionsupport_agent=Agent(name="Support")billing_agent=Agent(name="Billing")session=SQLiteSession("user_123")# Both agents will see the same conversation historyresult1=awaitRunner.run(support_agent,"Help me with my account",session=session)result2=awaitRunner.run(billing_agent,"What are my charges?",session=session)

importasynciofromagentsimportAgent,Runner,SQLiteSessionasyncdefmain():# Create an agentagent=Agent(name="Assistant",instructions="Reply very concisely.",)# Create a session instance that will persist across runssession=SQLiteSession("conversation_123","conversation_history.db")print("=== Sessions Example ===")print("The agent will remember previous messages automatically.\n")# First turnprint("First turn:")print("User: What city is the Golden Gate Bridge in?")result=awaitRunner.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 conversationprint("Second turn:")print("User: What state is it in?")result=awaitRunner.run(agent,"What state is it in?",session=session)print(f"Assistant:{result.final_output}")print()# Third turn - continuing the conversationprint("Third turn:")print("User: What's the population of that state?")result=awaitRunner.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())