跳到主要内容

2、Session 会话

相关概念

大模型的调用是无状态的,每次调用 Agent.run() 时,它会创建一个单一的、无状态的交互。代理会响应你的消息,之后代理就“忘记” 这次对话了,再次调用 run 方法时,只会用这次新的消息作为上下文和模型进行对话。

但大多数实际应用需要的是对话 ,而不仅仅是一次性的交互。这就是会话的作用所在。agno 为了解决 agent 记忆问题,使用 Session 和 Memory 机制。

本文聚焦在 Session 的使用上。

Session 会话

session 是会话级别,同一个用户和 agent 进行对话,比如一开始聊财经相关的内容,聊了很久以后,想换个话题,聊点体育相关的内容,为了不让之前关于财经的内容影响之后聊体育的对话内容,这时可以新建一个会话,相当于强制忘掉之前财经内容。Session 对应于 langgraph 中的短时记忆(short-term memory)。每个会话都有自己的 session_id,如果不提供的话,agno 会自己生成一个。

Memory 记忆

Memory 记忆是针对用户级别的,是帮助 agent 更好的了解用户,当对话中出现相关信息,例如用户的姓名、偏好或习惯时,具备记忆功能的智能体会自动将其存储在数据库中。之后,当这些信息再次变得相关时,智能体会自动检索并将其自然地运用到对话中。这样,智能体就能在与用户的互动中有效地学习每个用户的信息 。当对话中涉及到用户信息相关的内容,这时就需要 agent 记住,默认的 Memory instruction 是以下内容(翻译以后的)

记忆功能应记录与当前对话相关的用户个人信息,例如:

- 个人信息:姓名、年龄、职业、所在地、兴趣爱好和偏好

- 观点和偏好:用户喜欢、不喜欢、享受或感到沮丧的事物

- 用户分享的重要人生事件或经历

- 与用户当前状况、挑战或目标相关的重要背景信息

- 任何其他能够深入了解用户个性、观点或需求的信息

当然也可以自定义 Memory instruction,让 agent 记住我们想让它记住的内容。这块内容以后再详细介绍。

db

无论是 Session 会话还是 Memory 记忆,都需要将内容进行存储,而 db 则为会话和以及提供了底层的存储。如果不设置 db,即使有 session_id 和 user_id,agent(以及之后要介绍的 Teams 和 workflow) 也是无法实现记忆的,因为根本就没有将数据存储下来

agno 官方提供了很多常见的存储实现如 postgresql, mongodb, mysql, redis, sqlite 等等,本地测试过程中通常使用 sqlite 或者 InMemoryDb(内存),生产环境需要使用高可用的存储。

Session 实践

agno_agent = Agent(  
name="Agno Agent",
model=OpenAILike(
api_key=os.getenv("OPENAI_API_KEY"),
id="qwen-max"
),
add_history_to_context=True,
db=SqliteDb(db_file="agno.db"),
debug_mode=True,
debug_level=2,
markdown=True,
)

resp = agno_agent.run("你好,我叫 yyx ,你是谁? ")
print(resp.content)
resp = agno_agent.run("我叫什么名字?")
print(resp.content)

这里定义 Agent 时,需要使用 add_history_to_context=True , 这样 agent 在每次交互过程中,才会去获取历史消息并添加到本次任务消息中。

上面代码得到的结果是

(第一次调用)
DEBUG ======== user =====
DEBUG 你好,我叫 yyx ,你是谁?
DEBUG ======== assistant =====
DEBUG 你好,yyx!我是通义千问,是阿里云开发的人工智能助手。很高兴认识你!有什么我可以帮助你的吗?

(第二次调用)

DEBUG ======== user =====
DEBUG 你好,我叫 yyx ,你是谁?
DEBUG ======== assistant =====
DEBUG 你好,yyx!我是通义千问,是阿里云开发的人工智能助手。很高兴认识你!有什么我可以帮助你的吗?
DEBUG ======== user =====
DEBUG 我叫什么名字?
DEBUG ======== assistant =====
DEBUG 你刚刚告诉我,你的名字叫 yyx。

第二次调用时已经将第一次调用的消息带了上去。所以第二次的消息,agent 可以说出我的名字。

session_id 如何产生的?

上面的代码,并没有传如 session_id, agent 在调用 run 方法的时候,会使用 uuid 生成一个 session_id, 在本次脚本运行过程中,会使用相同的 session_id。

但是更加推荐的方式是客户端传入 session_id,这样使用 agno 搭建一个 agent 后端服务,用户的会话就由客户端来控制了,在使用 agentOS 进行 agent 对话时,也是由前端生成 session_id 传给后端。

agno_agent = Agent(  
name="Agno Agent",
model=OpenAILike(
api_key=os.getenv("OPENAI_API_KEY"),
id="qwen-max"
),
db=SqliteDb(db_file="agno.db"),
add_history_to_context=True,
debug_mode=True,
debug_level=2,
markdown=True,
)

resp = agno_agent.run("你好,我叫 yyx ,你是谁? ", session_id="conversion_123")
print(resp.content)
resp = agno_agent.run("我叫什么名字?", session_id="conversion_123")
print(resp.content)

# 停止脚本,注释掉
# resp = agno_agent.run("你好,我叫 yyx ,你是谁? ")
resp = agno_agent.run("我叫什么名字?",session_id="conversion_123")

这里调用 agent.run 方法时传入了 session_id 参数,当第一次调用结束以后,注释掉介绍自己的代码,直接问我的名字,此时 agent 可以获取到历史消息。

Session 管理

agent 可以对 session 数据进行管理,包括消息获取,写入,总结等等

agno_agent = Agent(  
name="Agno Agent",
model=OpenAILike(
api_key=os.getenv("OPENAI_API_KEY"),
id="qwen-max"
),
db=SqliteDb(db_file="agno.db"),
add_history_to_context=True,
debug_mode=True,
debug_level=2,
markdown=True,
)

session = agno_agent.get_session("conversion_123")
messages = session.get_messages()
for message in messages:
print(message.content)

使用 agent.get_session() 可以获得会话中所有运行的所有消息,包括工具调用和系统消息

<additional_information>
- Use markdown to format your answers.
</additional_information>
你好,我叫 yyx ,你是谁?
你好,yyx!我是Qwen,由阿里云开发的超大规模语言模。很高兴认识你!我可以帮助回答问题、提供信息、进行对话等。请问有什么我可以帮到你的吗?
我叫什么名字?
你刚才告诉我,你的名字是 yyx。如果你有其他昵称或者想用不同的名字称呼你,请告诉我!
我叫什么名字?
你叫 yyx。如果你有其他名字或者昵称,也可以告诉我!

如果要获取仅包含用户和助手消息的更简单的列表,可以使用 get_chat_history 方法:

messages = agno_agent.get_chat_history(session_id="conversion_123", last_n_runs=5)  
for message in messages:
print(message.content, "|", message.role)

get_chat_history 可以传 last_n_runs 来控制获取最近的多少轮会话。

Session Caching  会话缓存

会话缓存会将会话对象存储在内存中以提高性能。 cache_session=True 会在首次数据库读取后将已填充的会话对象保留在内存中,从而避免后续运行进行额外的查询。

from agno.agent import Agent
from agno.models.openai import OpenAIChat

agno_agent = Agent(
name="Agno Agent",
model=OpenAILike(
api_key=os.getenv("OPENAI_API_KEY"),
id="qwen-max"
),
db=SqliteDb(db_file="agno.db"),
add_history_to_context=True,
cache_session=True
markdown=True,
)

# First run loads from database and caches
agent.run("First message")

# Subsequent runs use cached session (faster)
agent.run("Second message")

总结

agno 中通过 session 和 memory 来实现短时记忆和用户信息记忆,session 是会话级别的,memory 是用户级别的。

底层都是通过 db 来实现数据的存储。