Agno 支持使用 SingleStore 作为 Agent 的存储后端,通过 SingleStoreStorage 类实现。

用法

这里 获取 SingleStore 的凭证。

singlestore_storage_for_agent.py
from os import getenv

from sqlalchemy.engine import create_engine

from agno.agent import Agent
from agno.storage.singlestore import SingleStoreStorage

# SingleStore 配置
USERNAME = getenv("SINGLESTORE_USERNAME")
PASSWORD = getenv("SINGLESTORE_PASSWORD")
HOST = getenv("SINGLESTORE_HOST")
PORT = getenv("SINGLESTORE_PORT")
DATABASE = getenv("SINGLESTORE_DATABASE")
SSL_CERT = getenv("SINGLESTORE_SSL_CERT", None)

# SingleStore 数据库 URL
db_url = f"mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DATABASE}?charset=utf8mb4"
if SSL_CERT:
    db_url += f"&ssl_ca={SSL_CERT}&ssl_verify_cert=true"

# 创建数据库引擎
db_engine = create_engine(db_url)

# 使用 Singlestore 数据库创建存储后端
storage = SingleStoreStorage(
    # 将会话存储在 ai.sessions 表中
    table_name="agent_sessions",
    # db_engine: Singlestore 数据库引擎
    db_engine=db_engine,
    # schema: Singlestore schema
    schema=DATABASE,
)

# 将存储添加到 Agent
agent = Agent(storage=storage)

参数

参数类型默认值描述
table_namestr-要使用的表名。
schemaOptional[str]"ai"Schema 名称。
db_urlOptional[str]None数据库 URL,如果提供了的话。
db_engineOptional[Engine]None要使用的数据库引擎。
schema_versionint1Schema 的版本号。
auto_upgrade_schemaboolFalse如果为 true,则在必要时自动升级 schema。

开发者资源