【Agent】3. FunctionAgent / AgentWorkflow 基础介绍

AgentWorkflow是一个用于运行一个或多个代理系统的协调器。在本案例中我们将创建一个包含单个FunctionAgent的简单工作流并以此介绍基本功能。1. 案例目标本案例的主要目标是演示如何使用LlamaIndex的FunctionAgent和AgentWorkflow构建一个简单的代理系统。具体来说我们将创建一个具有网络搜索功能的FunctionAgent展示如何运行代理并获取响应演示如何维护代理的状态和记忆展示如何实现流式输出演示如何在工具中访问工作流上下文展示如何实现人机交互循环2. 技术栈与核心依赖LlamaIndex- 用于构建代理和工作流OpenAI API- 提供语言模型Tavily API- 提供网络搜索功能FunctionAgent- 用于构建可执行工具调用的代理AgentWorkflow- 用于协调一个或多个代理的执行核心依赖包! pip install llama-index ! pip install tavily-python3. 环境配置首先需要设置OpenAI API密钥和Tavily API密钥from llama_index.llms.openai import OpenAI llm OpenAI(modelgpt-4o-mini, api_keysk-...)4. 案例实现4.1 创建网络搜索工具我们使用Tavily API创建一个网络搜索工具。创建工具时需要注意以下几点给工具一个合适的名称和文档字符串/描述LLM使用这些信息来理解工具的功能注释类型帮助LLM理解预期的输入和输出类型尽可能使用异步使工作流更高效from tavily import AsyncTavilyClient async def search_web(query: str) - str: Useful for using the web to answer questions. client AsyncTavilyClient(api_keytvly-...) return str(await client.search(query))4.2 创建FunctionAgent使用定义好的工具和LLM创建FunctionAgentfrom llama_index.core.agent.workflow import FunctionAgent agent FunctionAgent( tools[search_web], llmllm, system_promptYou are a helpful assistant that can search the web for information., )4.3 运行代理现在我们可以运行代理了response await agent.run(user_msgWhat is the weather in San Francisco?) print(str(response))4.4 使用AgentWorkflow上述操作等同于使用AgentWorkflow和单个FunctionAgentfrom llama_index.core.agent.workflow import AgentWorkflow workflow AgentWorkflow(agents[agent]) response await workflow.run(user_msgWhat is the weather in San Francisco?)4.5 维护状态默认情况下FunctionAgent在运行之间是无状态的这意味着代理不会有之前运行的任何记忆。为了维护状态我们需要跟踪之前的状态。由于FunctionAgent在Workflow中运行状态存储在Context中可以在运行之间传递以维护状态和历史记录。from llama_index.core.workflow import Context ctx Context(agent) response await agent.run( user_msgMy name is Logan, nice to meet you!, ctxctx ) print(str(response)) response await agent.run(user_msgWhat is my name?, ctxctx) print(str(response))4.6 保存和恢复状态Context是可序列化的因此可以保存到数据库、文件等并在以后加载回来。JsonSerializer是一个简单的序列化器使用json.dumps和json.loads来序列化和反序列化上下文。JsonPickleSerializer是一个使用pickle来序列化和反序列化上下文的序列化器。如果上下文中有不可序列化的对象可以使用此序列化器。from llama_index.core.workflow import JsonPickleSerializer, JsonSerializer ctx_dict ctx.to_dict(serializerJsonSerializer()) restored_ctx Context.from_dict(agent, ctx_dict, serializerJsonSerializer())4.7 流式输出AgentWorkflow/FunctionAgent也支持流式输出。由于AgentWorkflow是一个Workflow它可以像任何其他Workflow一样进行流式处理。这通过从工作流返回的处理器来实现。有几个关键事件被流式传输可以在下面探索。如果只想流式传输LLM输出可以使用AgentStream事件。from llama_index.core.agent.workflow import ( AgentInput, AgentOutput, ToolCall, ToolCallResult, AgentStream, ) handler agent.run(user_msgWhat is the weather in Saskatoon?) async for event in handler.stream_events(): if isinstance(event, AgentStream): print(event.delta, end, flushTrue) # print(event.response) # 当前的完整响应 # print(event.raw) # 原始LLM API响应 # print(event.current_agent_name) # 当前代理名称4.8 工具和状态工具也可以定义为可以访问工作流上下文。这意味着可以从上下文中设置和检索变量并在工具中或工具之间使用它们。注意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 fName set to {name} agent FunctionAgent( tools[set_name], llmllm, system_promptYou are a helpful assistant that can set a name., initial_state{name: unset}, ) ctx Context(agent) response await agent.run(user_msgMy name is Logan, ctxctx) print(str(response)) state await ctx.store.get(state) print(state[name])4.9 人机交互循环工具也可以定义为涉及人机交互循环。这对于需要人工输入的任务很有用例如确认工具调用或提供反馈。使用工作流事件我们可以发出需要用户响应的事件。这里我们使用内置的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_idquestion, waiter_eventInputRequiredEvent( prefixquestion, user_nameLogan, ), requirements{user_name: Logan}, ) if response.response yes: return Dangerous task completed successfully. else: return Dangerous task aborted. agent FunctionAgent( tools[dangerous_task], llmllm, system_promptYou are a helpful assistant that can perform dangerous tasks., )5. 案例效果5.1 网络搜索功能代理能够使用Tavily API搜索网络并回答用户问题例如查询旧金山的天气情况。代理会自动调用search_web工具获取搜索结果并基于这些结果提供答案。5.2 状态维护通过使用Context对象代理能够记住之前的交互信息。例如当用户介绍自己的名字后代理在后续对话中能够记住并正确回答我的名字是什么这类问题。5.3 流式输出代理支持流式输出用户可以实时看到代理生成的响应内容而不必等待整个响应完成。这对于长响应或需要实时反馈的场景特别有用。5.4 工具状态访问工具可以访问和修改工作流状态实现更复杂的功能。例如set_name工具可以将用户的名字存储在上下文中供后续使用。5.5 人机交互循环对于需要人工确认的任务如危险操作代理可以暂停并等待用户输入然后根据用户反馈继续执行。这种机制在需要人工干预的自动化流程中非常有用。6. 案例实现思路6.1 代理架构本案例基于LlamaIndex的FunctionAgent和AgentWorkflow构建。FunctionAgent是一个能够调用工具的代理而AgentWorkflow则是一个协调器用于管理一个或多个代理的执行。这种架构使得构建复杂的AI系统变得简单而灵活。6.2 工具定义与调用工具是代理扩展功能的关键。通过定义具有适当类型注解和文档字符串的异步函数我们可以创建代理可以调用的工具。代理根据工具的名称、描述和参数类型来决定何时以及如何调用工具。6.3 状态管理Context对象是状态管理的核心。它提供了一个可序列化的状态存储可以在代理运行之间保持信息。通过Context代理可以记住之前的交互实现更连贯的对话体验。6.4 流式处理流式处理通过事件系统实现。代理在执行过程中会发出各种事件如AgentStream、ToolCall等。用户可以监听这些事件并实时处理实现流式输出和更精细的执行控制。6.5 人机交互人机交互循环通过特殊的事件机制实现。代理可以发出InputRequiredEvent请求用户输入并通过wait_for_event等待HumanResponseEvent响应。这种机制使得代理可以在需要时暂停执行并等待人工干预。7. 扩展建议7.1 多代理系统构建具有不同专业能力的多个代理如搜索代理、计算代理、写作代理等实现代理间的协作机制使它们能够共享信息和协同工作设计代理调度策略根据任务类型自动选择最合适的代理7.2 高级工具集成集成更多外部API如数据库、文件系统、云服务等实现工具链使一个工具的输出可以作为另一个工具的输入添加工具权限控制确保代理只能访问授权的资源7.3 增强状态管理实现分层状态管理区分短期记忆和长期记忆添加状态持久化机制支持大规模状态存储和检索实现状态版本控制支持状态回滚和历史追踪7.4 高级人机交互实现多模态交互支持语音、图像等多种输入方式添加交互历史记录允许用户回顾和修改之前的交互实现交互模板为常见交互场景提供预定义流程7.5 性能优化实现工具调用缓存避免重复执行相同的工具调用添加并行执行支持使代理能够同时执行多个独立任务优化流式处理减少延迟并提高响应速度8. 总结本案例介绍了LlamaIndex的FunctionAgent和AgentWorkflow的基本功能和使用方法。通过这个案例我们了解了如何创建具有工具调用能力的代理维护代理的状态和记忆实现流式输出和实时交互构建需要人工干预的自动化流程FunctionAgent和AgentWorkflow为构建复杂的AI系统提供了强大而灵活的基础。它们不仅支持简单的工具调用还提供了状态管理、流式处理和人机交互等高级功能使得构建生产级AI应用变得更加容易。通过本案例的方法开发者可以快速构建各种智能代理系统如智能客服、自动化助手、研究助手等。这些系统可以集成各种工具和服务实现复杂的功能同时保持良好的用户体验和系统可靠性。随着AI技术的不断发展FunctionAgent和AgentWorkflow这样的工具将成为构建下一代AI应用的重要组成部分为开发者提供更强大、更灵活的构建块。