在 LlamaIndex 中建立代理工作流

LlamaIndex 中的工作流提供了一種結構化的方式，可以將您的程式碼組織成順序的、可管理的步驟。

這樣的工作流是透過定義由 `Events` 觸發的 `Steps` 來建立的，並且它們本身會發出 `Events` 來觸發進一步的步驟。讓我們看看 Alfred 展示一個用於 RAG 任務的 LlamaIndex 工作流。

Workflow Schematic

工作流提供以下幾個主要優點

將程式碼清晰地組織成離散的步驟
用於靈活控制流的事件驅動架構
步驟之間型別安全的通訊
內建狀態管理
支援簡單和複雜的代理互動

正如你可能已經猜到的那樣，**工作流在代理的自主性與保持對整體工作流的控制之間取得了很好的平衡。**

那麼，讓我們學習如何自己建立一個工作流吧！

建立工作流

您可以按照此 Notebook 中的程式碼進行操作，您可以使用 Google Colab 執行它。

基本工作流建立

安裝 Workflow 包

如LlamaHub 部分所述，我們可以使用以下命令安裝 Workflow 包

pip install llama-index-utils-workflow

我們可以透過定義一個繼承自 `Workflow` 的類，並用 `@step` 裝飾你的函式來建立單步工作流。我們還需要新增 `StartEvent` 和 `StopEvent`，它們是用於指示工作流開始和結束的特殊事件。

from llama_index.core.workflow import StartEvent, StopEvent, Workflow, step

class MyWorkflow(Workflow):
    @step
    async def my_step(self, ev: StartEvent) -> StopEvent:
        # do something here
        return StopEvent(result="Hello, world!")


w = MyWorkflow(timeout=10, verbose=False)
result = await w.run()

如您所見，我們現在可以透過呼叫 `w.run()` 來執行工作流。

連線多個步驟

要連線多個步驟，我們**建立自定義事件以在步驟之間傳輸資料。**為此，我們需要新增一個在步驟之間傳遞的 `Event`，並將第一步的輸出傳輸到第二步。

from llama_index.core.workflow import Event

class ProcessingEvent(Event):
    intermediate_result: str

class MultiStepWorkflow(Workflow):
    @step
    async def step_one(self, ev: StartEvent) -> ProcessingEvent:
        # Process initial data
        return ProcessingEvent(intermediate_result="Step 1 complete")

    @step
    async def step_two(self, ev: ProcessingEvent) -> StopEvent:
        # Use the intermediate result
        final_result = f"Finished processing: {ev.intermediate_result}"
        return StopEvent(result=final_result)

w = MultiStepWorkflow(timeout=10, verbose=False)
result = await w.run()
result

這裡的型別提示很重要，因為它確保工作流正確執行。讓我們把事情弄得更復雜一點！

迴圈和分支

型別提示是工作流中最強大的部分，因為它允許我們建立分支、迴圈和連線以促進更復雜的工作流。

讓我們透過使用聯合運算子 `|` 來展示一個**建立迴圈**的示例。在下面的示例中，我們看到 `LoopEvent` 被用作步驟的輸入，也可以作為輸出返回。

from llama_index.core.workflow import Event
import random


class ProcessingEvent(Event):
    intermediate_result: str


class LoopEvent(Event):
    loop_output: str


class MultiStepWorkflow(Workflow):
    @step
    async def step_one(self, ev: StartEvent | LoopEvent) -> ProcessingEvent | LoopEvent:
        if random.randint(0, 1) == 0:
            print("Bad thing happened")
            return LoopEvent(loop_output="Back to step one.")
        else:
            print("Good thing happened")
            return ProcessingEvent(intermediate_result="First step complete.")

    @step
    async def step_two(self, ev: ProcessingEvent) -> StopEvent:
        # Use the intermediate result
        final_result = f"Finished processing: {ev.intermediate_result}"
        return StopEvent(result=final_result)


w = MultiStepWorkflow(verbose=False)
result = await w.run()
result

繪製工作流

我們也可以繪製工作流。讓我們使用 `draw_all_possible_flows` 函式來繪製工作流。這會將工作流儲存在一個 HTML 檔案中。

from llama_index.utils.workflow import draw_all_possible_flows

w = ... # as defined in the previous section
draw_all_possible_flows(w, "flow.html")

workflow drawing

本課程中我們將介紹最後一個很酷的技巧，即向工作流新增狀態的能力。

狀態管理

當您希望跟蹤工作流狀態，以便每個步驟都可以訪問相同狀態時，狀態管理非常有用。我們可以透過在步驟函式的引數頂部使用 `Context` 型別提示來實現這一點。

from llama_index.core.workflow import Context, StartEvent, StopEvent


@step
async def query(self, ctx: Context, ev: StartEvent) -> StopEvent:
    # store query in the context
    await ctx.store.set("query", "What is the capital of France?")

    # do something with context and event
    val = ...

    # retrieve query from the context
    query = await ctx.store.get("query")

    return StopEvent(result=val)

太棒了！現在您已經掌握了 LlamaIndex 中代理的基礎知識！

工作流還有一些更復雜的細微差別，您可以在LlamaIndex 文件中瞭解它們。

然而，還有另一種建立工作流的方法，它依賴於 `AgentWorkflow` 類。讓我們看看如何使用它來建立多代理工作流。

使用多代理工作流實現工作流自動化

除了手動建立工作流之外，我們還可以使用 **`AgentWorkflow` 類來建立多代理工作流**。`AgentWorkflow` 使用工作流代理，允許您建立一個由一個或多個代理組成的系統，這些代理可以根據其專業能力相互協作和移交任務。這使得能夠構建複雜的代理系統，其中不同的代理處理任務的不同方面。我們不是從 `llama_index.core.agent` 匯入類，而是從 `llama_index.core.agent.workflow` 匯入代理類。在 `AgentWorkflow` 建構函式中，必須將一個代理指定為根代理。當用戶訊息進來時，它首先被路由到根代理。

然後每個代理可以

直接使用其工具處理請求
將任務移交給更適合該任務的另一個代理
向用戶返回響應

讓我們看看如何建立一個多代理工作流。

from llama_index.core.agent.workflow import AgentWorkflow, ReActAgent
from llama_index.llms.huggingface_api import HuggingFaceInferenceAPI

# Define some tools
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

llm = HuggingFaceInferenceAPI(model_name="Qwen/Qwen2.5-Coder-32B-Instruct")

# we can pass functions directly without FunctionTool -- the fn/docstring are parsed for the name/description
multiply_agent = ReActAgent(
    name="multiply_agent",
    description="Is able to multiply two integers",
    system_prompt="A helpful assistant that can use a tool to multiply numbers.",
    tools=[multiply],
    llm=llm,
)

addition_agent = ReActAgent(
    name="add_agent",
    description="Is able to add two integers",
    system_prompt="A helpful assistant that can use a tool to add numbers.",
    tools=[add],
    llm=llm,
)

# Create the workflow
workflow = AgentWorkflow(
    agents=[multiply_agent, addition_agent],
    root_agent="multiply_agent",
)

# Run the system
response = await workflow.run(user_msg="Can you add 5 and 3?")

代理工具還可以修改我們之前提到的工作流狀態。在啟動工作流之前，我們可以提供一個初始狀態字典，所有代理都可以訪問該字典。狀態儲存在工作流上下文的狀態鍵中。它將被注入到 state_prompt 中，該 prompt 會增強每個新的使用者訊息。

讓我們透過修改之前的示例來注入一個計數器以計數函式呼叫

from llama_index.core.workflow import Context

# Define some tools
async def add(ctx: Context, a: int, b: int) -> int:
    """Add two numbers."""
    # update our count
    cur_state = await ctx.store.get("state")
    cur_state["num_fn_calls"] += 1
    await ctx.store.set("state", cur_state)

    return a + b

async def multiply(ctx: Context, a: int, b: int) -> int:
    """Multiply two numbers."""
    # update our count
    cur_state = await ctx.store.get("state")
    cur_state["num_fn_calls"] += 1
    await ctx.store.set("state", cur_state)

    return a * b

...

workflow = AgentWorkflow(
    agents=[multiply_agent, addition_agent],
    root_agent="multiply_agent",
    initial_state={"num_fn_calls": 0},
    state_prompt="Current state: {state}. User message: {msg}",
)

# run the workflow with context
ctx = Context(workflow)
response = await workflow.run(user_msg="Can you add 5 and 3?", ctx=ctx)

# pull out and inspect the state
state = await ctx.store.get("state")
print(state["num_fn_calls"])

恭喜！您現在已經掌握了 LlamaIndex 中代理的基礎知識！🎉

讓我們繼續進行最後一個測驗，以鞏固您的知識！🚀

< > 在 GitHub 上更新

Agents 課程