Redis 工作流存储

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

用法

运行 Redis

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

docker run --name my-redis -p 6379:6379 -d redis

redis_storage_for_workflow.py

"""
运行：`pip install openai httpx newspaper4k redis agno` 来安装依赖项
"""

import json
from typing import Iterator

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

# 使用默认的本地连接初始化 Redis 存储
storage = RedisStorage(
    # 用于命名会话的 Redis 键前缀
    prefix="agno_test",
    # Redis 主机地址
    host="localhost",
    # Redis 端口号
    port=6379,
)


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=[
            "您将收到热门新闻及其链接。",
            "仔细阅读每篇文章并思考其内容。",
            "然后生成最终的《纽约时报》风格的文章。",
            "将文章分成几个部分，并在最后提供要点。",
            "确保标题吸引人且引人入胜。",
            "分享每篇文章的分数、标题、URL 和摘要。",
            "为每个部分提供相关的标题，并在每个部分中提供详细信息/事实/流程。",
            "忽略无法阅读或理解的文章。",
            "请记住：您是为《纽约时报》写作，因此文章的质量很重要。",
        ],
    )

    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__":
    # 运行工作流
    report: Iterator[RunResponse] = HackerNewsReporter(
        storage=storage, debug_mode=False
    ).run(num_stories=5)
    # 打印报告
    pprint_run_response(report, markdown=True, show_time=True)

参数

参数	类型	描述	默认值
prefix	str	用于 Redis 键的 prefix，以 namespace 会话	Required
host	str	Redis host 地址	"localhost"
port	int	Redis 端口号	6379
db	int	Redis 数据库号	0
password	Optional[str]	需要 authentication 时的 Redis password	None
mode	Optional[Literal["agent", "team", "workflow"]]	Storage mode	"agent"

开发者资源

• 查看 Cookbook

Assistant

Responses are generated using AI and may contain mistakes.