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=DATABASE,
)
# 将存储添加到 Agent
agent = Agent(storage=storage)
参数
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
table_name | str | - | 要使用的表名。 |
schema | Optional[str] | "ai" | Schema 名称。 |
db_url | Optional[str] | None | 数据库 URL,如果提供了的话。 |
db_engine | Optional[Engine] | None | 要使用的数据库引擎。 |
schema_version | int | 1 | Schema 的版本号。 |
auto_upgrade_schema | bool | False | 如果为 true,则在必要时自动升级 schema。 |
开发人员资源
- 查看 Cookbook