Agent Session

此示例展示了如何创建一个使用 SQLite 数据库存储持久化内存的 Agent。在恢复对话时，我们会为 agent 设置 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)

用法

1. 创建虚拟环境

   打开 Terminal 并创建一个 python 虚拟环境。

   Mac

   python3 -m venv .venv
   source .venv/bin/activate
   
2. 安装库

   pip install openai sqlalchemy agno
   
3. 运行 Agent

   python agent_session.py