FunctionAgent / AgentWorkflow 基础介绍

AgentWorkflow 是一个用于运行一个或多个智能体系统的编排器。在这个示例中，我们将创建一个包含单个 FunctionAgent 的简单工作流，并使用它来演示基本功能。

%pip install llama-index

设置

在这个示例中，我们将使用 OpenAI 作为我们的LLM。要查看所有LLM，请查阅示例文档或LlamaHub获取所有支持的LLM列表及其安装/使用方法。

from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-4o-mini", api_key="sk-...")

为了让我们的智能体更加实用，我们可以为其提供可使用的工具/操作。在本例中，我们将使用Tavily来实现一个能够搜索网络信息的工具。您可以从Tavily获取免费的API密钥。

%pip install tavily-python

创建工具时，非常重要的一点是：

给工具一个合适的名称和文档字符串/描述。LLM 使用这个来理解工具的功能。
标注类型。这有助于LLM理解预期的输入和输出类型。
尽可能使用异步，因为这将使工作流程更高效。

from tavily import AsyncTavilyClient


async def search_web(query: str) -> str:
    """Useful for using the web to answer questions."""
    client = AsyncTavilyClient(api_key="tvly-...")
    return str(await client.search(query))

在定义好工具和LLM后，我们可以创建一个使用该工具的AgentWorkflow。

from llama_index.core.agent.workflow import FunctionAgent

agent = FunctionAgent(
    tools=[search_web],
    llm=llm,
    system_prompt="You are a helpful assistant that can search the web for information.",
)

运行智能体

现在我们的智能体已创建，我们可以运行它了！

response = await agent.run(user_msg="What is the weather in San Francisco?")
print(str(response))

The current weather in San Francisco is as follows:

- **Temperature**: 16.1°C (61°F)
- **Condition**: Partly cloudy
- **Wind**: 13.6 mph (22.0 kph) from the west
- **Humidity**: 64%
- **Visibility**: 16 km (9 miles)
- **Pressure**: 1017 mb (30.04 in)

For more details, you can check the full report [here](https://www.weatherapi.com/).

以上内容等同于使用单个 FunctionAgent 的 AgentWorkflow 实现：

from llama_index.core.agent.workflow import AgentWorkflow

workflow = AgentWorkflow(agents=[agent])

response = await workflow.run(user_msg="What is the weather in San Francisco?")

如果您正在创建包含多个智能体的工作流，可以将智能体列表传递给 AgentWorkflow 构造函数。在我们的多智能体工作流示例中了解更多信息。

状态维护

默认情况下，FunctionAgent 将在运行之间保持无状态。这意味着智能体不会记住之前的任何运行记录。

为了维护状态，我们需要跟踪之前的状态。由于 FunctionAgent 运行在 Workflow 中，状态存储在 Context 中。这可以在多次运行之间传递以维护状态和历史记录。

from llama_index.core.workflow import Context

ctx = Context(agent)

response = await agent.run(
    user_msg="My name is Logan, nice to meet you!", ctx=ctx
)
print(str(response))

Nice to meet you, Logan! How can I assist you today?

response = await agent.run(user_msg="What is my name?", ctx=ctx)
print(str(response))

Your name is Logan.

上下文是可序列化的，因此可以保存到数据库、文件等中，并在之后重新加载。

JsonSerializer 是一个简单的序列化器，它使用 json.dumps 和 json.loads 来序列化和反序列化上下文。

JsonPickleSerializer 是一个使用 pickle 来序列化和反序列化上下文的序列化器。如果您的上下文中存在不可序列化的对象，可以使用此序列化器。

from llama_index.core.workflow import JsonPickleSerializer, JsonSerializer

ctx_dict = ctx.to_dict(serializer=JsonSerializer())

restored_ctx = Context.from_dict(agent, ctx_dict, serializer=JsonSerializer())

response = await agent.run(
    user_msg="Do you still remember my name?", ctx=restored_ctx
)
print(str(response))

Yes, I remember your name is Logan.

流式处理

AgentWorkflow/FunctionAgent 同样支持流式传输。由于 AgentWorkflow 是一个 Workflow，它可以像其他任何 Workflow 一样进行流式传输。这是通过使用从工作流返回的处理程序来实现的。有几个关键事件会被流式传输，欢迎在下方探索。

如果您只想流式传输LLM输出，可以使用AgentStream事件。

from llama_index.core.agent.workflow import (
    AgentInput,
    AgentOutput,
    ToolCall,
    ToolCallResult,
    AgentStream,
)

handler = agent.run(user_msg="What is the weather in Saskatoon?")

async for event in handler.stream_events():
    if isinstance(event, AgentStream):
        print(event.delta, end="", flush=True)
        # print(event.response)  # the current full response
        # print(event.raw)  # the raw llm api response
        # print(event.current_agent_name)  # the current agent name
    # elif isinstance(event, AgentInput):
    #    print(event.input)  # the current input messages
    #    print(event.current_agent_name)  # the current agent name
    # elif isinstance(event, AgentOutput):
    #    print(event.response)  # the current full response
    #    print(event.tool_calls)  # the selected tool calls, if any
    #    print(event.raw)  # the raw llm api response
    # elif isinstance(event, ToolCallResult):
    #    print(event.tool_name)  # the tool name
    #    print(event.tool_kwargs)  # the tool kwargs
    #    print(event.tool_output)  # the tool output
    # elif isinstance(event, ToolCall):
    #     print(event.tool_name)  # the tool name
    #     print(event.tool_kwargs)  # the tool kwargs

The current weather in Saskatoon is as follows:

- **Temperature**: 22.2°C (72°F)
- **Condition**: Overcast
- **Humidity**: 25%
- **Wind Speed**: 6.0 mph (9.7 kph) from the northwest
- **Visibility**: 4.8 km
- **Pressure**: 1018 mb

For more details, you can check the full report [here](https://www.weatherapi.com/).

工具与状态

工具也可以被定义为能够访问工作流上下文。这意味着您可以从上下文中设置和检索变量，并在工具内部或工具之间使用它们。

注意： Context 参数应为该工具的第一个参数。

from llama_index.core.workflow import Context


async def set_name(ctx: Context, name: str) -> str:
    async with ctx.store.edit_state() as ctx_state:
        ctx_state["state"]["name"] = name
    return f"Name set to {name}"


agent = FunctionAgent(
    tools=[set_name],
    llm=llm,
    system_prompt="You are a helpful assistant that can set a name.",
    initial_state={"name": "unset"},
)

ctx = Context(agent)

response = await agent.run(user_msg="My name is Logan", ctx=ctx)
print(str(response))

state = await ctx.store.get("state")
print(state["name"])

Your name has been set to Logan.
Logan

人在回路

工具也可以被定义为需要人工介入的循环过程。这对于需要人工输入的任务非常有用，例如确认工具调用或提供反馈。

使用工作流事件，我们可以发出需要用户响应的事件。这里，我们使用内置的 InputRequiredEvent 和 HumanResponseEvent 来处理人工介入环节，但您也可以定义自己的事件。

wait_for_event 将发出 waiter_event 并等待直到看到带有指定 requirements 的 HumanResponseEvent。waiter_id 用于确保我们为每个 waiter_id 仅发送一个 waiter_event。

from llama_index.core.workflow import (
    Context,
    InputRequiredEvent,
    HumanResponseEvent,
)


async def dangerous_task(ctx: Context) -> str:
    """A dangerous task that requires human confirmation."""

    question = "Are you sure you want to proceed?"
    response = await ctx.wait_for_event(
        HumanResponseEvent,
        waiter_id=question,
        waiter_event=InputRequiredEvent(
            prefix=question,
            user_name="Logan",
        ),
        requirements={"user_name": "Logan"},
    )
    if response.response == "yes":
        return "Dangerous task completed successfully."
    else:
        return "Dangerous task aborted."


agent = FunctionAgent(
    tools=[dangerous_task],
    llm=llm,
    system_prompt="You are a helpful assistant that can perform dangerous tasks.",
)

handler = agent.run(user_msg="I want to proceed with the dangerous task.")

async for event in handler.stream_events():
    if isinstance(event, InputRequiredEvent):
        response = input(event.prefix).strip().lower()
        handler.ctx.send_event(
            HumanResponseEvent(
                response=response,
                user_name=event.user_name,
            )
        )

response = await handler
print(str(response))

The dangerous task has been completed successfully. If you need anything else, feel free to ask!

在生产环境中，你可能需要通过WebSocket或多个API请求来处理人在回路机制。

如前所述，Context 对象是可序列化的，这意味着我们也可以在运行过程中保存工作流并在之后恢复它。

注意：工作流恢复时，所有进行中的函数/步骤都将从头开始执行。

from llama_index.core.workflow import JsonSerializer

handler = agent.run(user_msg="I want to proceed with the dangerous task.")

input_ev = None
async for event in handler.stream_events():
    if isinstance(event, InputRequiredEvent):
        input_ev = event
        break

# save the context somewhere for later
ctx_dict = handler.ctx.to_dict(serializer=JsonSerializer())

# get the response from the user
response_str = input(input_ev.prefix).strip().lower()

# restore the workflow
restored_ctx = Context.from_dict(agent, ctx_dict, serializer=JsonSerializer())

handler = agent.run(ctx=restored_ctx)
handler.ctx.send_event(
    HumanResponseEvent(
        response=response_str,
        user_name=input_ev.user_name,
    )
)
response = await handler
print(str(response))

The dangerous task has been completed successfully. If you need anything else, feel free to ask!