Postgres 工作流存储

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

用法

运行 PgVector

安装 docker desktop 并使用以下命令在端口 5532 上运行 PgVector：

docker run -d \
  -e POSTGRES_DB=ai \
  -e POSTGRES_USER=ai \
  -e POSTGRES_PASSWORD=ai \
  -e PGDATA=/var/lib/postgresql/data/pgdata \
  -v pgvolume:/var/lib/postgresql/data \
  -p 5532:5432 \
  --name pgvector \
  agno/pgvector:16

postgres_storage_for_workflow.py

import json
from typing import Iterator

import httpx
from agno.agent import Agent
from agno.run.response import RunResponse
from agno.storage.postgres import PostgresStorage
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

db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"


class HackerNewsReporter(Workflow):
    description: str = (
        "获取 Hacker News 的热门文章并撰写报告。"
    )

    hn_agent: Agent = Agent(
        description="获取 hackernews 的热门文章。 "
        "分享所有可能的信息，包括链接、分数、标题和摘要（如果可用）。",
        show_tool_calls=True,
    )

    writer: Agent = Agent(
        tools=[Newspaper4kTools()],
        description="撰写一篇引人入胜的关于 hackernews 热门文章的报告。",
        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 = PostgresStorage(table_name="agent_sessions", db_url=db_url)
    storage.drop()
    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 会话的表名。
schema	Optional[str]	"ai"	要用于此表的模式。
db_url	Optional[str]	None	要连接的数据库 URL。
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.