函数调用 Agent 的工作流¶

本 notebook 详细介绍了如何从零开始设置 Workflow 来构建函数调用 Agent。

函数调用 Agent 的工作原理是利用支持工具/函数 API 的 LLM（如 OpenAI、Ollama、Anthropic 等）来调用函数和使用工具。

我们的工作流将是有状态的（带记忆），并且能够调用 LLM 来选择工具和处理用户输入消息。

In [ ]

!pip install -U llama-index

!pip install -U llama-index

In [ ]

import os

os.environ["OPENAI_API_KEY"] = "sk-proj-..."

import os os.environ["OPENAI_API_KEY"] = "sk-proj-..."

[可选] 使用 Llamatrace 设置可观测性¶

设置追踪以便可视化工作流中的每个步骤。

由于工作流是异步优先的，这在 notebook 中运行良好。如果您在自己的代码中运行，如果尚未运行异步事件循环，则需要使用 asyncio.run() 来启动一个。

async def main():
    <async code>

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

设计工作流¶

一个 Agent 由几个步骤组成：

1. 处理最新的用户输入消息，包括添加到记忆和获取最新的聊天历史
2. 使用工具 + 聊天历史调用 LLM
3. 解析工具调用（如果有的话）
4. 如果有工具调用，则执行它们，并循环直到没有工具调用
5. 当没有工具调用时，返回 LLM 响应

工作流事件¶

为了处理这些步骤，我们需要定义几个事件：

1. 用于处理新消息并准备聊天历史的事件
2. 用于处理流式响应的事件
3. 用于触发工具调用的事件
4. 用于处理工具调用结果的事件

其他步骤将使用内置的 StartEvent 和 StopEvent 事件。

In [ ]

from llama_index.core.llms import ChatMessage
from llama_index.core.tools import ToolSelection, ToolOutput
from llama_index.core.workflow import Event


class InputEvent(Event):
    input: list[ChatMessage]


class StreamEvent(Event):
    delta: str


class ToolCallEvent(Event):
    tool_calls: list[ToolSelection]


class FunctionOutputEvent(Event):
    output: ToolOutput

from llama_index.core.llms import ChatMessage from llama_index.core.tools import ToolSelection, ToolOutput from llama_index.core.workflow import Event class InputEvent(Event): input: list[ChatMessage] class StreamEvent(Event): delta: str class ToolCallEvent(Event): tool_calls: list[ToolSelection] class FunctionOutputEvent(Event): output: ToolOutput

工作流本身¶

定义好事件后，我们可以构建我们的工作流和步骤。

请注意，工作流会自动使用类型注解进行自我验证，因此我们步骤上的类型注解非常有用！

In [ ]

from typing import Any, List

from llama_index.core.llms.function_calling import FunctionCallingLLM
from llama_index.core.memory import ChatMemoryBuffer
from llama_index.core.tools.types import BaseTool
from llama_index.core.workflow import (
    Context,
    Workflow,
    StartEvent,
    StopEvent,
    step,
)
from llama_index.llms.openai import OpenAI


class FuncationCallingAgent(Workflow):
    def __init__(
        self,
        *args: Any,
        llm: FunctionCallingLLM | None = None,
        tools: List[BaseTool] | None = None,
        **kwargs: Any,
    ) -> None:
        super().__init__(*args, **kwargs)
        self.tools = tools or []

        self.llm = llm or OpenAI()
        assert self.llm.metadata.is_function_calling_model

    @step
    async def prepare_chat_history(
        self, ctx: Context, ev: StartEvent
    ) -> InputEvent:
        # clear sources
        await ctx.set("sources", [])

        # check if memory is setup
        memory = await ctx.get("memory", default=None)
        if not memory:
            memory = ChatMemoryBuffer.from_defaults(llm=self.llm)

        # get user input
        user_input = ev.input
        user_msg = ChatMessage(role="user", content=user_input)
        memory.put(user_msg)

        # get chat history
        chat_history = memory.get()

        # update context
        await ctx.set("memory", memory)

        return InputEvent(input=chat_history)

    @step
    async def handle_llm_input(
        self, ctx: Context, ev: InputEvent
    ) -> ToolCallEvent | StopEvent:
        chat_history = ev.input

        # stream the response
        response_stream = await self.llm.astream_chat_with_tools(
            self.tools, chat_history=chat_history
        )
        async for response in response_stream:
            ctx.write_event_to_stream(StreamEvent(delta=response.delta or ""))

        # save the final response, which should have all content
        memory = await ctx.get("memory")
        memory.put(response.message)
        await ctx.set("memory", memory)

        # get tool calls
        tool_calls = self.llm.get_tool_calls_from_response(
            response, error_on_no_tool_call=False
        )

        if not tool_calls:
            sources = await ctx.get("sources", default=[])
            return StopEvent(
                result={"response": response, "sources": [*sources]}
            )
        else:
            return ToolCallEvent(tool_calls=tool_calls)

    @step
    async def handle_tool_calls(
        self, ctx: Context, ev: ToolCallEvent
    ) -> InputEvent:
        tool_calls = ev.tool_calls
        tools_by_name = {tool.metadata.get_name(): tool for tool in self.tools}

        tool_msgs = []
        sources = await ctx.get("sources", default=[])

        # call tools -- safely!
        for tool_call in tool_calls:
            tool = tools_by_name.get(tool_call.tool_name)
            additional_kwargs = {
                "tool_call_id": tool_call.tool_id,
                "name": tool.metadata.get_name(),
            }
            if not tool:
                tool_msgs.append(
                    ChatMessage(
                        role="tool",
                        content=f"Tool {tool_call.tool_name} does not exist",
                        additional_kwargs=additional_kwargs,
                    )
                )
                continue

            try:
                tool_output = tool(**tool_call.tool_kwargs)
                sources.append(tool_output)
                tool_msgs.append(
                    ChatMessage(
                        role="tool",
                        content=tool_output.content,
                        additional_kwargs=additional_kwargs,
                    )
                )
            except Exception as e:
                tool_msgs.append(
                    ChatMessage(
                        role="tool",
                        content=f"Encountered error in tool call: {e}",
                        additional_kwargs=additional_kwargs,
                    )
                )

        # update memory
        memory = await ctx.get("memory")
        for msg in tool_msgs:
            memory.put(msg)

        await ctx.set("sources", sources)
        await ctx.set("memory", memory)

        chat_history = memory.get()
        return InputEvent(input=chat_history)

from typing import Any, List from llama_index.core.llms.function_calling import FunctionCallingLLM from llama_index.core.memory import ChatMemoryBuffer from llama_index.core.tools.types import BaseTool from llama_index.core.workflow import ( Context, Workflow, StartEvent, StopEvent, step, ) from llama_index.llms.openai import OpenAI class FuncationCallingAgent(Workflow): def __init__( self, *args: Any, llm: FunctionCallingLLM | None = None, tools: List[BaseTool] | None = None, **kwargs: Any, ) -> None: super().__init__(*args, **kwargs) self.tools = tools or [] self.llm = llm or OpenAI() assert self.llm.metadata.is_function_calling_model @step async def prepare_chat_history( self, ctx: Context, ev: StartEvent ) -> InputEvent: # clear sources await ctx.set("sources", []) # check if memory is setup memory = await ctx.get("memory", default=None) if not memory: memory = ChatMemoryBuffer.from_defaults(llm=self.llm) # get user input user_input = ev.input user_msg = ChatMessage(role="user", content=user_input) memory.put(user_msg) # get chat history chat_history = memory.get() # update context await ctx.set("memory", memory) return InputEvent(input=chat_history) @step async def handle_llm_input( self, ctx: Context, ev: InputEvent ) -> ToolCallEvent | StopEvent: chat_history = ev.input # stream the response response_stream = await self.llm.astream_chat_with_tools( self.tools, chat_history=chat_history ) async for response in response_stream: ctx.write_event_to_stream(StreamEvent(delta=response.delta or "")) # save the final response, which should have all content memory = await ctx.get("memory") memory.put(response.message) await ctx.set("memory", memory) # get tool calls tool_calls = self.llm.get_tool_calls_from_response( response, error_on_no_tool_call=False ) if not tool_calls: sources = await ctx.get("sources", default=[]) return StopEvent( result={"response": response, "sources": [*sources]} ) else: return ToolCallEvent(tool_calls=tool_calls) @step async def handle_tool_calls( self, ctx: Context, ev: ToolCallEvent ) -> InputEvent: tool_calls = ev.tool_calls tools_by_name = {tool.metadata.get_name(): tool for tool in self.tools} tool_msgs = [] sources = await ctx.get("sources", default=[]) # call tools -- safely! for tool_call in tool_calls: tool = tools_by_name.get(tool_call.tool_name) additional_kwargs = { "tool_call_id": tool_call.tool_id, "name": tool.metadata.get_name(), } if not tool: tool_msgs.append( ChatMessage( role="tool", content=f"Tool {tool_call.tool_name} does not exist", additional_kwargs=additional_kwargs, ) ) continue try: tool_output = tool(**tool_call.tool_kwargs) sources.append(tool_output) tool_msgs.append( ChatMessage( role="tool", content=tool_output.content, additional_kwargs=additional_kwargs, ) ) except Exception as e: tool_msgs.append( ChatMessage( role="tool", content=f"Encountered error in tool call: {e}", additional_kwargs=additional_kwargs, ) ) # update memory memory = await ctx.get("memory") for msg in tool_msgs: memory.put(msg) await ctx.set("sources", sources) await ctx.set("memory", memory) chat_history = memory.get() return InputEvent(input=chat_history)

就是这样！让我们稍微探索一下我们编写的工作流。

prepare_chat_history()：这是我们的主要入口点。它负责将用户消息添加到记忆中，并使用记忆获取最新的聊天历史。它返回一个 InputEvent。

handle_llm_input()：由 InputEvent 触发，它使用聊天历史和工具来提示 LLM。如果找到工具调用，则发出一个 ToolCallEvent。否则，我们认为工作流完成并发出一个 StopEvent。

handle_tool_calls()：由 ToolCallEvent 触发，它会调用工具并进行错误处理，然后返回工具输出。此事件会触发一个 循环，因为它会发出一个 InputEvent，将我们带回 handle_llm_input()。

运行工作流！¶

注意： 使用循环时，我们需要注意运行时长。这里我们将超时设置为 120 秒。

In [ ]

from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI


def add(x: int, y: int) -> int:
    """Useful function to add two numbers."""
    return x + y


def multiply(x: int, y: int) -> int:
    """Useful function to multiply two numbers."""
    return x * y


tools = [
    FunctionTool.from_defaults(add),
    FunctionTool.from_defaults(multiply),
]

agent = FuncationCallingAgent(
    llm=OpenAI(model="gpt-4o-mini"), tools=tools, timeout=120, verbose=True
)

ret = await agent.run(input="Hello!")

from llama_index.core.tools import FunctionTool from llama_index.llms.openai import OpenAI def add(x: int, y: int) -> int: """Useful function to add two numbers.""" return x + y def multiply(x: int, y: int) -> int: """Useful function to multiply two numbers.""" return x * y tools = [ FunctionTool.from_defaults(add), FunctionTool.from_defaults(multiply), ] agent = FuncationCallingAgent( llm=OpenAI(model="gpt-4o-mini"), tools=tools, timeout=120, verbose=True ) ret = await agent.run(input="Hello!")

Running step prepare_chat_history
Step prepare_chat_history produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event StopEvent

In [ ]

print(ret["response"])

print(ret["response"])

assistant: Hello! How can I assist you today?

In [ ]

ret = await agent.run(input="What is (2123 + 2321) * 312?")

ret = await agent.run(input="What is (2123 + 2321) * 312?")

Running step prepare_chat_history
Step prepare_chat_history produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event ToolCallEvent
Running step handle_tool_calls
Step handle_tool_calls produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event ToolCallEvent
Running step handle_tool_calls
Step handle_tool_calls produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event StopEvent

聊天历史¶

默认情况下，工作流会为每次运行创建一个新的 Context。这意味着聊天历史不会在多次运行之间保留。但是，我们可以将自己的 Context 传递给工作流以保留聊天历史。

In [ ]

from llama_index.core.workflow import Context

ctx = Context(agent)

ret = await agent.run(input="Hello! My name is Logan.", ctx=ctx)
print(ret["response"])

ret = await agent.run(input="What is my name?", ctx=ctx)
print(ret["response"])

from llama_index.core.workflow import Context ctx = Context(agent) ret = await agent.run(input="Hello! My name is Logan.", ctx=ctx) print(ret["response"]) ret = await agent.run(input="What is my name?", ctx=ctx) print(ret["response"])

Running step prepare_chat_history
Step prepare_chat_history produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event StopEvent
assistant: Hello, Logan! How can I assist you today?
Running step prepare_chat_history
Step prepare_chat_history produced event InputEvent
Running step handle_llm_input
Step handle_llm_input produced event StopEvent
assistant: Your name is Logan.

流式处理¶

使用 .run() 方法返回的 handler，我们还可以访问流式事件。

In [ ]

agent = FuncationCallingAgent(
    llm=OpenAI(model="gpt-4o-mini"), tools=tools, timeout=120, verbose=False
)

handler = agent.run(input="Hello! Write me a short story about a cat.")

async for event in handler.stream_events():
    if isinstance(event, StreamEvent):
        print(event.delta, end="", flush=True)

response = await handler
# print(ret["response"])

agent = FuncationCallingAgent( llm=OpenAI(model="gpt-4o-mini"), tools=tools, timeout=120, verbose=False ) handler = agent.run(input="Hello! Write me a short story about a cat.") async for event in handler.stream_events(): if isinstance(event, StreamEvent): print(event.delta, end="", flush=True) response = await handler # print(ret["response"])

Once upon a time in a quaint little village, there lived a curious cat named Whiskers. Whiskers was no ordinary cat; he had a beautiful coat of orange and white fur that shimmered in the sunlight, and his emerald green eyes sparkled with mischief.

Every day, Whiskers would explore the village, visiting the bakery for a whiff of freshly baked bread and the flower shop to sniff the colorful blooms. The villagers adored him, often leaving out little treats for their favorite feline.

One sunny afternoon, while wandering near the edge of the village, Whiskers stumbled upon a hidden path that led into the woods. His curiosity piqued, he decided to follow the path, which was lined with tall trees and vibrant wildflowers. As he ventured deeper, he heard a soft, melodic sound that seemed to beckon him.

Following the enchanting music, Whiskers soon found himself in a clearing where a group of woodland creatures had gathered. They were having a grand celebration, complete with dancing, singing, and a feast of berries and nuts. The animals welcomed Whiskers with open paws, inviting him to join their festivities.

Whiskers, delighted by the warmth and joy of his new friends, danced and played until the sun began to set. As the sky turned shades of pink and orange, he realized it was time to return home. The woodland creatures gifted him a small, sparkling acorn as a token of their friendship.

From that day on, Whiskers would often visit the clearing, sharing stories of the village and enjoying the company of his woodland friends. He learned that adventure and friendship could be found in the most unexpected places, and he cherished every moment spent in the magical woods.

And so, Whiskers continued to live his life filled with curiosity, laughter, and the warmth of friendship, reminding everyone that sometimes, the best adventures are just a whisker away.