【Agent】【workflow】5.函数调用代理工作流案例

1. 案例目标

本案例展示了如何使用LlamaIndex工作流构建一个函数调用代理(Function Calling Agent)。主要目标包括：

• 实现一个能够自动选择和调用工具的智能代理
• 构建具有记忆功能的状态化工作流
• 展示如何使用支持函数调用的LLM(如OpenAI)来处理复杂任务
• 实现流式响应功能，提供更好的用户体验
• 提供完整的端到端实现，从工作流设计到实际运行

核心价值

函数调用代理是现代AI应用的核心组件，它使AI能够与外部系统交互、执行计算和访问实时数据。本案例提供了一个完整的实现框架，可作为构建更复杂AI代理的基础。

2. 技术栈与核心依赖

from llama_index.core.workflow import ( Event, Context, Workflow, StartEvent, StopEvent, step, ) 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.llms.openai import OpenAI from llama_index.core.tools import FunctionTool

核心依赖包括：

• LlamaIndex工作流框架：用于构建和编排多步骤处理流程
• 函数调用LLM：支持工具/函数调用的语言模型(如OpenAI的GPT系列)
• 聊天记忆缓冲区：用于维护对话历史和上下文
• 基础工具接口：用于定义和集成各种工具
• 函数工具：用于将Python函数转换为可调用的工具

3. 环境配置

# 安装依赖 !pip install -U llama-index # 设置API密钥 os.environ["OPENAI_API_KEY"] = "sk-proj-..."

环境配置包括：

• 安装最新版本的LlamaIndex库
• 设置OpenAI API密钥，确保能够访问支持函数调用的模型
• 可选：配置Llamatrace以可视化工作流中的每个步骤

4. 案例实现

4.1 工作流设计

函数调用代理工作流包含以下步骤：

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

4.2 定义事件

为了处理这些步骤，需要定义以下事件：

InputEvent

处理新消息并准备聊天历史的事件。

StreamEvent

处理流式响应的事件。

ToolCallEvent

触发工具调用的事件。

FunctionOutputEvent

处理工具调用结果的事件。

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

4.3 工作流实现

步骤1: 准备聊天历史

@step async def prepare_chat_history( self, ctx: Context, ev: StartEvent ) -> InputEvent: # 清除源 await ctx.store.set("sources", []) # 检查记忆是否设置 memory = await ctx.store.get("memory", default=None) if not memory: memory = ChatMemoryBuffer.from_defaults(llm=self.llm) # 获取用户输入 user_input = ev.input user_msg = ChatMessage(role="user", content=user_input) memory.put(user_msg) # 获取聊天历史 chat_history = memory.get() # 更新上下文 await ctx.store.set("memory", memory) return InputEvent(input=chat_history)

步骤2: 处理LLM输入

@step async def handle_llm_input( self, ctx: Context, ev: InputEvent ) -> ToolCallEvent | StopEvent: chat_history = ev.input # 流式响应 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 "")) # 保存最终响应，应包含所有内容 memory = await ctx.store.get("memory") memory.put(response.message) await ctx.store.set("memory", memory) # 获取工具调用 tool_calls = self.llm.get_tool_calls_from_response( response, error_on_no_tool_call=False ) if not tool_calls: sources = await ctx.store.get("sources", default=[]) return StopEvent( result={"response": response, "sources": [*sources]} ) else: return ToolCallEvent(tool_calls=tool_calls)

步骤3: 处理工具调用

@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.store.get("sources", default=[]) # 安全地调用工具 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, ) ) # 更新记忆 memory = await ctx.store.get("memory") for msg in tool_msgs: memory.put(msg) await ctx.store.set("sources", sources) await ctx.store.set("memory", memory) chat_history = memory.get() return InputEvent(input=chat_history)

4.4 工具定义

示例工具

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), ]

4.5 运行工作流

# 创建代理实例 agent = FuncationCallingAgent( llm=OpenAI(model="gpt-4o-mini"), tools=tools, timeout=120, verbose=True ) # 运行查询 ret = await agent.run(input="Hello!") print(ret["response"]) # 运行需要工具调用的查询 ret = await agent.run(input="What is (2123 + 2321) * 312?")

4.6 聊天历史管理

# 创建共享上下文以保持聊天历史 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"])

4.7 流式响应

# 使用流式事件 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

5. 案例效果

本案例实现了以下效果：

• 自动工具选择：系统能够根据用户查询自动选择合适的工具
• 多步骤工具调用：支持链式工具调用，完成复杂任务
• 对话记忆：能够记住之前的对话内容，提供连贯的交互体验
• 流式响应：提供实时的响应流，改善用户体验
• 错误处理：对工具调用中的错误进行适当处理，确保系统稳定性

示例查询结果

对于查询"What is (2123 + 2321) * 312?"，系统能够：

1. 自动识别需要使用add和multiply工具
2. 先调用add(2123, 2321)得到4444
3. 再调用multiply(4444, 312)得到1,386,528
4. 返回最终答案："The result of (2123 + 2321) * 312 is 1,386,528."

关键特性

本案例中的函数调用代理具有以下特点：

• 使用事件驱动的工作流设计，清晰分离不同处理步骤
• 支持流式响应，提供实时反馈
• 内置记忆管理，保持对话上下文
• 类型安全的工作流验证，确保事件和步骤的正确连接
• 灵活的工具集成机制，易于扩展新工具

6. 案例实现思路

本案例的实现思路如下：

1. 工作流设计：将函数调用代理过程分解为三个主要步骤，每个步骤负责特定的功能
2. 事件驱动：使用事件系统在工作流步骤间传递数据和状态
3. 循环处理：通过返回InputEvent创建循环，直到没有更多工具调用
4. 状态管理：使用上下文存储维护记忆和源信息
5. 流式处理：通过StreamEvent提供实时响应反馈

工作流数据流

1. StartEvent(用户输入) → prepare_chat_history步骤 → InputEvent(聊天历史)
2. InputEvent → handle_llm_input步骤 → ToolCallEvent(工具调用) 或 StopEvent(响应)
3. ToolCallEvent → handle_tool_calls步骤 → InputEvent(更新后的聊天历史)
4. 循环：返回步骤2，直到没有更多工具调用

7. 扩展建议

基于本案例，可以考虑以下扩展方向：

• 更多工具类型：集成更多类型的工具，如API调用、数据库查询、文件操作等
• 工具权限控制：实现工具级别的权限控制，确保安全性
• 并行工具调用：支持并行调用多个独立工具，提高效率
• 工具链优化：实现智能工具链优化，减少不必要的中间步骤
• 自定义记忆策略：实现更灵活的记忆管理策略，如长期记忆和短期记忆分离
• 多模态支持：扩展支持图像、音频等多模态输入和输出
• 工具使用统计：添加工具使用统计和分析功能，优化工具选择
• 动态工具加载：实现运行时动态加载和卸载工具的能力

8. 总结

本案例展示了如何使用LlamaIndex工作流构建一个函数调用代理，通过事件驱动的方式实现了工具选择、调用和响应生成的完整流程。案例中的关键技术点包括：

• 工作流编排：使用LlamaIndex工作流框架构建复杂的多步骤处理流程
• 事件驱动架构：通过事件系统实现步骤间的解耦和通信
• 循环处理：通过事件返回实现循环，处理多步骤工具调用
• 状态管理：使用上下文存储维护对话历史和中间结果
• 流式响应：通过StreamEvent提供实时反馈

通过这种方式，系统能够自动理解用户意图，选择合适的工具，执行必要的计算，并生成准确的响应。这种函数调用代理模式是构建现代AI应用的基础，可以广泛应用于各种需要与外部系统交互的场景。