所有工作流都附带一个 session_state 字典，您可以使用它来缓存中间结果。session_state 绑定到 session_id 并可以持久化到数据库。

为工作流提供 storage 以启用会话状态在数据库中的持久化。

例如，您可以使用 SqliteWorkflowStorage 将结果缓存到 Sqlite 数据库中。

# 创建工作流
generate_blog_post = BlogPostGenerator(
    # 为此演示修复 session_id
    session_id="my-session-id",
    storage=SqliteWorkflowStorage(
        table_name="generate_blog_post_workflows",
        db_file="tmp/workflows.db",
    ),
)

然后在 run() 方法中，您可以根据需要从 session_state 读取和写入。


class BlogPostGenerator(Workflow):
    # ... agents
    def run(self, topic: str, use_cache: bool = True) -> Iterator[RunResponse]:
        # 从 session state 缓存读取
        if use_cache and "blog_posts" in self.session_state:
            logger.info("正在检查是否存在缓存的博客文章")
            for cached_blog_post in self.session_state["blog_posts"]:
                if cached_blog_post["topic"] == topic:
                    logger.info("找到缓存的博客文章")
                    yield RunResponse(
                        run_id=self.run_id,
                        event=RunEvent.workflow_completed,
                        content=cached_blog_post["blog_post"],
                    )
                    return

        # ... 生成博客文章

        # 保存到 session state 以供将来运行
        if "blog_posts" not in self.session_state:
            self.session_state["blog_posts"] = []
        self.session_state["blog_posts"].append({"topic": topic, "blog_post": self.writer.run_response.content})

当工作流启动时，该特定 session_id 的 session_state 将从数据库读取；当工作流结束时，session_state 将存储到数据库中。

您可以随时调用 self.write_to_storage() 将 session_state 保存到数据库。以防您需要中止工作流但想存储中间结果。

查看 博客文章生成器 以了解如何使用会话状态进行缓存的示例。