workflow

agno版本: 1.7.2

Workflow类

from agno.run.response import RunResponseEvent
from utils import new_openai_model
from typing import Iterator

from agno.agent import Agent
from agno.utils.pprint import pprint_run_response
from agno.workflow import Workflow


class CacheWorkflow(Workflow):
    description: str = "A workflow that caches previous outputs"

    agent = Agent(model=new_openai_model("gpt-4o-mini"))

    def run(self, message: str) -> Iterator[RunResponseEvent]:
        yield from self.agent.run(message, stream=True)


def test():
    workflow = CacheWorkflow()
    response = workflow.run(message="Tell me a joke.")
    pprint_run_response(response, markdown=True, show_time=True)

Workflow.run()应该yield一个RunResponseEvent, 文档写的是RunResponse. 实际上原文档给出的案例两个类型都yield出去了, 因此跑的时候会警告:

WARNING  Workflow.run() should only yield RunResponseEvent objects, got: <class 'agno.run.response.RunResponse'>

不过实际上并不是必须yield一个RunResponseEvent, 因此只是一个警告. 虽然说是随便yield, 还是尽量避免警告吧.

在1.7.5的版本中貌似可以yield其它东西了, 我yieldRunResponse没有警告.

然后得到一个空回复. 至少缓存工作流, 什么博客生成器是这样.

因为pprint_run_response需要接受一个RunResponse | Iterable[RunResponseEvent] | TeamRunResponse | Iterable[TeamRunResponseEvent], 所以这里空回复了.

Workflow的状态

访问session_state, 是一个以str类型为key的字典.