状态 是 Agent 在运行期间需要维护的任何类型的数据。

Agent 一个简洁但常见的用例是为用户管理列表、项目和其他“信息”。例如,购物列表、待办事项列表、愿望清单等。

可以使用 session_state 轻松管理。Agent 在工具调用中更新 session_state,并在 descriptioninstructions 中将其暴露给模型。

Agno 提供了一个强大而优雅的状态管理系统,其工作原理如下:

  • Agent 具有 session_state 参数。
  • 我们将状态变量添加到此 session_state 字典中。
  • 我们在工具调用或其他函数中更新 session_state 字典。
  • 我们在 descriptioninstructions 中与模型共享当前的 session_state
  • session_state 与 Agent 会话一起存储,并持久化到数据库中。这意味着它在执行周期之间可用。

以下 Agent 管理购物列表的示例:

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

# 定义一个工具,该工具递增我们的计数器并返回新值
def add_item(agent: Agent, item: str) -> str:
    """将商品添加到购物列表中。"""
    agent.session_state["shopping_list"].append(item)
    return f"购物列表现在是 {agent.session_state['shopping_list']}"


# 创建一个维护状态的 Agent
agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    # 使用从 0 开始的计数器初始化会话状态
    session_state={"shopping_list": []},
    tools=[add_item],
    # 您可以在指令中使用来自会话状态的变量
    instructions="当前状态(购物列表)是:{shopping_list}",
    # 重要:将状态添加到消息中
    add_state_in_messages=True,
    markdown=True,
)

# 用法示例
agent.print_response("将牛奶、鸡蛋和面包添加到购物列表中", stream=True)
print(f"最终的会话状态:{agent.session_state}")

这是状态管理所能达到的最佳状态了。

在多个运行中维护状态

会话的一个巨大优势是能够跨多次运行维护状态。例如,假设 Agent 正在帮助用户跟踪他们的购物列表。

通过设置 add_state_in_messages=Truesession_state 字典的键可以在 descriptioninstructions 中作为变量使用。

使用此模式将 shopping_list 直接添加到指令中。

shopping_list.py
from textwrap import dedent

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


# 定义管理我们购物列表的工具
def add_item(agent: Agent, item: str) -> str:
    """将商品添加到购物列表中并返回确认。"""
    # 如果商品不在列表中,则添加它
    if item.lower() not in [i.lower() for i in agent.session_state["shopping_list"]]:
        agent.session_state["shopping_list"].append(item)
        return f"已将 '{item}' 添加到购物列表中"
    else:
        return f"'{item}' 已经在购物列表中"


def remove_item(agent: Agent, item: str) -> str:
    """按名称从购物列表中删除商品。"""
    # 区分大小写的搜索
    for i, list_item in enumerate(agent.session_state["shopping_list"]):
        if list_item.lower() == item.lower():
            agent.session_state["shopping_list"].pop(i)
            return f"已从购物列表中移除 '{list_item}'"

    return f"在购物列表中未找到 '{item}'"


def list_items(agent: Agent) -> str:
    """列出购物列表中的所有商品。"""
    shopping_list = agent.session_state["shopping_list"]

    if not shopping_list:
        return "购物列表是空的。"

    items_text = "\n".join([f"- {item}" for item in shopping_list])
    return f"当前购物列表:\n{items_text}"


# 创建一个维护状态的购物列表管理器 Agent
agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    # 使用空的购物列表初始化会话状态
    session_state={"shopping_list": []},
    tools=[add_item, remove_item, list_items],
    # 您可以在指令中使用来自会话状态的变量
    instructions=dedent("""\
        你的任务是管理一个购物列表。

        购物列表开始时是空的。你可以添加商品、按名称删除商品以及列出所有商品。

        当前购物列表:{shopping_list}
    """),
    show_tool_calls=True,
    add_state_in_messages=True,
    markdown=True,
)

# 用法示例
agent.print_response("将牛奶、鸡蛋和面包添加到购物列表中", stream=True)
print(f"会话状态:{agent.session_state}")

agent.print_response("我拿到了面包", stream=True)
print(f"会话状态:{agent.session_state}")

agent.print_response("我需要苹果和橙子", stream=True)
print(f"会话状态:{agent.session_state}")

agent.print_response("我的列表上有什么?", stream=True)
print(f"会话状态:{agent.session_state}")

agent.print_response("清空我的列表并重新开始,只保留香蕉和酸奶", stream=True)
print(f"会话状态:{agent.session_state}")

我们很喜欢能在多个运行中如此优雅地维护和传递状态。

在指令中使用状态

通过设置 add_state_in_messages=True,您可以在指令中使用来自会话状态的变量。

不要在指令中使用 f-string 语法。直接使用 {key} 语法,Agno 会为您替换值。

state_in_instructions.py
from textwrap import dedent

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


agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    # 使用变量初始化会话状态
    session_state={"user_name": "John"},
    # 您可以在指令中使用来自会话状态的变量
    instructions="用户的名字是 {user_name}",
    show_tool_calls=True,
    add_state_in_messages=True,
    markdown=True,
)

agent.print_response("我的名字是什么?", stream=True)

在数据库中持久化状态

session_state 是 Agent 会话的一部分,如果提供了 storage 驱动程序,它会在每次运行后保存到数据库中。

以下 Agent 维护购物列表并将其状态持久化到数据库的示例。多次运行此脚本以查看状态的持久化。

session_state_storage.py
"""运行 `pip install agno openai sqlalchemy` 来安装依赖项。"""

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.storage.sqlite import SqliteStorage


# 定义一个将商品添加到购物列表的工具
def add_item(agent: Agent, item: str) -> str:
    """将商品添加到购物列表中。"""
    if item not in agent.session_state["shopping_list"]:
        agent.session_state["shopping_list"].append(item)
    return f"购物列表现在是 {agent.session_state['shopping_list']}"


agent = Agent(
    model=OpenAIChat(id="gpt-4o-mini"),
    # 固定会话 ID,以便在执行周期中继续相同的会话
    session_id="fixed_id_for_demo",
    # 使用空的购物列表初始化会话状态
    session_state={"shopping_list": []},
    # 添加一个将商品添加到购物列表的工具
    tools=[add_item],
    # 将会话状态存储在 SQLite 数据库中
    storage=SqliteStorage(table_name="agent_sessions", db_file="tmp/data.db"),
    # 在指令中添加当前的购物列表状态
    instructions="当前购物列表是:{shopping_list}",
    # 重要:设置 `add_state_in_messages=True`
    # 使 `{shopping_list}` 在指令中可用
    add_state_in_messages=True,
    markdown=True,
)

# 用法示例
agent.print_response("我的购物列表有什么?", stream=True)
print(f"会话状态:{agent.session_state}")
agent.print_response("添加牛奶、鸡蛋和面包", stream=True)
print(f"会话状态:{agent.session_state}")