DynamoDB 工作流存储

Agno 支持使用 DynamoDB 作为工作流的存储后端，通过 DynamoDbStorage 类实现。

用法

您需要为 DynamoDbStorage 类提供 aws_access_key_id 和 aws_secret_access_key 参数。

dynamodb_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.dynamodb import DynamoDbStorage
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="获取 hackernews 的热门新闻。 "
        "分享所有可能的信息，包括 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__":
    # 运行工作流
    report: Iterator[RunResponse] = HackerNewsReporter(
        storage=DynamoDbStorage(
            table_name="workflow_sessions", region_name="us-east-1"
        ),
        debug_mode=False,
    ).run(num_stories=5)
    # 打印报告
    pprint_run_response(report, markdown=True, show_time=True)

参数

参数	类型	默认	描述
table_name	str	-	要使用的表名。
region_name	Optional[str]	None	DynamoDB 表的区域名称。
aws_access_key_id	Optional[str]	None	AWS 访问密钥 ID，如果提供的话。
aws_secret_access_key	Optional[str]	None	AWS 秘密访问密钥，如果提供的话。
endpoint_url	Optional[str]	None	端点 URL，如果提供的话。
create_table_if_not_exists	bool	True	如果为 true，则在表不存在时创建该表。

开发者资源

• 查看 Cookbook