Singlestore Team 存储

Agno 支持使用 Singlestore 作为 Team 的存储后端，通过 SingleStoreStorage 类实现。

用法

从这里获取 Singlestore 的凭证。

singlestore_storage_for_team.py

"""
运行: `pip install openai duckduckgo-search newspaper4k lxml_html_clean agno` 来安装依赖
"""

import os
from os import getenv
from typing import List

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.storage.singlestore import SingleStoreStorage
from agno.team import Team
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.hackernews import HackerNewsTools
from agno.utils.certs import download_cert
from pydantic import BaseModel
from sqlalchemy.engine import create_engine

# 配置 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)


# 如果未提供 SSL_CERT，则下载证书
if not SSL_CERT:
    SSL_CERT = download_cert(
        cert_url="https://portal.singlestore.com/static/ca/singlestore_bundle.pem",
        filename="singlestore_bundle.pem",
    )
    if SSL_CERT:
        os.environ["SINGLESTORE_SSL_CERT"] = SSL_CERT


# 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)


class Article(BaseModel):
    title: str
    summary: str
    reference_links: List[str]


hn_researcher = Agent(
    name="HackerNews 调研员",
    model=OpenAIChat("gpt-4o"),
    role="从 hackernews 获取热门故事。",
    tools=[HackerNewsTools()],
)

web_searcher = Agent(
    name="网络搜索员",
    model=OpenAIChat("gpt-4o"),
    role="搜索网络信息",
    tools=[DuckDuckGoTools()],
    add_datetime_to_instructions=True,
)


hn_team = Team(
    name="HackerNews 小组",
    mode="coordinate",
    model=OpenAIChat("gpt-4o"),
    members=[hn_researcher, web_searcher],
    storage=SingleStoreStorage(
        table_name="agent_sessions",
        db_engine=db_engine,
        schema=DATABASE,
        auto_upgrade_schema=True,
    ),
    instructions=[
        "首先，在 hackernews 上搜索用户询问的内容。",
        "然后，让网络搜索员搜索每个故事以获取更多信息。",
        "最后，提供一个深思熟虑且引人入胜的摘要。",
    ],
    response_model=Article,
    show_tool_calls=True,
    markdown=True,
    debug_mode=True,
    show_members_responses=True,
)

hn_team.print_response("写一篇关于 hackernews 上排名前 2 的故事的文章。")

参数

参数	类型	默认值	描述
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