session_id，这样之前的聊天记录就会被保留。
主要特点：
- 在 SQLite 数据库中存储对话历史
- 支持跨多个会话继续对话
- 在响应中引用之前的上下文
代码
agent_session.py
import json
from typing import Optional
import typer
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.storage.sqlite import SqliteStorage
from rich.console import Console
from rich.json import JSON
from rich.panel import Panel
from rich.prompt import Prompt
from rich import print
console = Console()
def create_agent(user: str = "user"):
session_id: Optional[str] = None
# 询问用户是想开始新会话还是继续现有会话
new = typer.confirm("Do you want to start a new session?")
# 如果用户不想开始新会话，则获取现有会话
agent_storage = SqliteStorage(
table_name="agent_sessions", db_file="tmp/agents.db"
)
if not new:
existing_sessions = agent_storage.get_all_session_ids(user)
if len(existing_sessions) > 0:
session_id = existing_sessions[0]
agent = Agent(
user_id=user,
# 设置 agent 的 session_id 以恢复对话
session_id=session_id,
model=OpenAIChat(id="gpt-4o"),
storage=agent_storage,
# 将聊天历史添加到消息中
add_history_to_messages=True,
num_history_responses=3,
markdown=True,
)
if session_id is None:
session_id = agent.session_id
if session_id is not None:
print(f"Started Session: {session_id}\n")
else:
print("Started Session\n")
else:
print(f"Continuing Session: {session_id}\n")
return agent
def print_messages(agent):
"""以格式化的面板打印当前的聊天历史"""
console.print(
Panel(
JSON(
json.dumps(
[
m.model_dump(include={"role", "content"})
for m in agent.memory.messages
]
),
indent=4,
),
title=f"Chat History for session_id: {agent.session_id}",
expand=True,
)
)
def main(user: str = "user"):
agent = create_agent(user)
print("Chat with an OpenAI agent!")
exit_on = ["exit", "quit", "bye"]
while True:
message = Prompt.ask(f"[bold] :sunglasses: {user} [/bold]")
if message in exit_on:
break
agent.print_response(message=message, stream=True, markdown=True)
print_messages(agent)
if __name__ == "__main__":
typer.run(main)