JSON 存储

Agno 支持使用本地 JSON 文件作为 Agent 的存储后端，通过 JsonStorage 类实现。

用法

json_storage_for_agent.py

import json
from typing import Iterator

import httpx
from agno.agent import Agent
from agno.run.response import RunResponse
from agno.storage.json import JsonStorage
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__":
    # 运行工作流
    report: Iterator[RunResponse] = HackerNewsReporter(
        storage=JsonStorage(dir_path="tmp/workflow_sessions_json"), debug_mode=False
    ).run(num_stories=5)
    # 打印报告
    pprint_run_response(report, markdown=True, show_time=True)

参数

参数	类型	默认	描述
dir_path	str	-	用于存储 JSON 文件的文件夹路径。

开发者资源

• 查看 Cookbook