SQLite 工作流存储

Agno 支持使用 SqliteStorage 类将 SQLite 作为工作流的存储后端。

用法

sqlite_storage_for_workflow

import json
from typing import Iterator

import httpx
from agno.agent import Agent
from agno.run.response import RunResponse
from agno.storage.sqlite import SqliteStorage
from agno.tools.newspaper4k import Newspaper4kTools
from agno.utils.log import logger
from agno.utils.pprint import pprint_run_response
from agno.workflow import Workflow


class HackerNewsReporter(Workflow):
    description: str = (
        "获取 Hacker News 的热门话题，并撰写一篇相关报道。"
    )

    hn_agent: Agent = Agent(
        description="获取 Hacker News 的热门话题。 "
        "分享所有可能的信息，包括 url、得分、标题和摘要（如果可用）。",
        show_tool_calls=True,
    )

    writer: Agent = Agent(
        tools=[Newspaper4kTools()],
        description="撰写一篇引人入胜的 Hacker News 热门话题报道。",
        instructions=[
            "你将获得热门话题及其链接。",
            "仔细阅读每篇文章并思考其内容",
            "然后生成一篇符合《纽约时报》风格的最终文章",
            "将文章分成几个部分，并在结尾提供关键要点。",
            "确保标题醒目且吸引人。",
            "分享每篇文章的得分、标题、网址和摘要。",
            "为每个部分提供相关的标题，并在每个部分中提供详细信息/事实/流程",
            "忽略你无法阅读或理解的文章。",
            "请记住：你的读者是《纽约时报》的读者，因此文章的质量非常重要。",
        ],
    )

    def get_top_hackernews_stories(self, num_stories: int = 10) -> str:
        """使用此函数获取 Hacker News 的热门话题。

        Args:
            num_stories (int): 要返回的故事数量。默认为 10。

        Returns:
            str: 热门话题的 JSON 字符串。
        """

        # 获取热门话题 ID
        response = httpx.get("https://hacker-news.firebaseio.com/v0/topstories.json")
        story_ids = response.json()

        # 获取故事详情
        stories = []
        for story_id in story_ids[:num_stories]:
            story_response = httpx.get(
                f"https://hacker-news.firebaseio.com/v0/item/{story_id}.json"
            )
            story = story_response.json()
            story["username"] = story["by"]
            stories.append(story)
        return json.dumps(stories)

    def run(self, num_stories: int = 5) -> Iterator[RunResponse]:
        # 在此处为 hn_agent 设置工具，以避免循环引用
        self.hn_agent.tools = [self.get_top_hackernews_stories]

        logger.info(f"正在从 HackerNews 获取前 {num_stories} 个热门话题。")
        top_stories: RunResponse = self.hn_agent.run(num_stories=num_stories)
        if top_stories is None or not top_stories.content:
            yield RunResponse(
                run_id=self.run_id, content="抱歉，无法获取热门话题。"
            )
            return

        logger.info("正在阅读每个故事并撰写报道。")
        yield from self.writer.run(top_stories.content, stream=True)


if __name__ == "__main__":
    # 运行工作流
    storage = SqliteStorage(table_name="workflow_sessions", db_file="tmp/data.db")
    report: Iterator[RunResponse] = HackerNewsReporter(
        storage=storage, debug_mode=False
    ).run(num_stories=5)
    # 打印报告
    pprint_run_response(report, markdown=True, show_time=True)

参数

参数	类型	默认	描述
table_name	str	-	用于存储 Workflow 会话的表名。
db_url	Optional[str]	None	要连接的数据库 URL。
db_file	Optional[str]	None	要连接的数据库文件。
db_engine	Optional[Engine]	None	要使用的 SQLAlchemy 数据库引擎。
schema_version	int	1	架构版本。
auto_upgrade_schema	bool	False	是否自动升级架构。

开发者资源

• 查看 Cookbook

Assistant

Responses are generated using AI and may contain mistakes.