我惊了！构建AI智能体居然这么简单？LlamaIndex保姆级教程，小白也能造出“钢铁侠”！

01 前言

其通过事件驱动的方式实现了工作流编排，其中事件（Event）和上下文（Context）是两个核心概念与关键要素。

在LlamaIndex中，智能体（Agent）的构建同样基于事件编排机制，其运行逻辑正是通过Workflow来实现的。

02 相关概念

LlamaIndex中定义了三种基本的Agent模式，分别是FunctionAgent、ReActAgent、CodeActAgent。

它们都继承至BaseWorkflowAgent类，而BaseWorkflowAgent又继承至Workflow，关系如下所示：

• FunctionAgent：

  使用大模型（LLM）原生Function Call能力驱动工具调用的Agent，工具的调用由LLM直接返回的tool_calls JSON触发。要求模型具有Function Call的能力。

• ReActAgent：

  借助LLM的提示词（思考 -> 执行 -> 反馈）这种文本协议驱动的推理式Agent，需要Agent手动解析文本得到工具调用信息。比FunctionAgent适用性更广，但是可能因解析工具失败而终止。

• CodeActAgent：

  允许模型用Python代码执行操作的Agent，可以借助LlamaIndex集成的CodeInterpreterTool等工具进行实现。类似OpenAI的Code Interpreter。

03 Agent实战

下面会依次介绍下Agent的简单使用、如何在Agent中使用上下文（Context）实现记忆、Human in the loop 人参与交互、多Agent协作、实现Agent结构化输出。示例主要使用FunctionAgent演示。

定义一个Agent

# pip install llama-index-core llama-index-llms-deepseek from llama_index.core.agent import FunctionAgent from llama_index.llms.deepseek import DeepSeek def multiply(a: float, b: float) -> float: """Multiply two numbers and returns the product""" print("execute multiply") return a * b def add(a: float, b: float) -> float: """Add two numbers and returns the sum""" print("execute add") return a + b llm = DeepSeek(model="deepseek-chat", api_key="sk-...") agent = FunctionAgent( tools=[multiply, add], llm=llm, system_prompt="You are an agent that can perform basic mathematical operations using tools.", verbose=True, ) # 执行 async def main(): result = await agent.run("what is 1 add 1") print(result) if __name__ == "__main__": import asyncio asyncio.run(main())

示例首先定义两个自定义函数作为Agent的工具，通过llama-index-llms-deepseek集成并定义DeepSeek作为Agent的llm。然后实例化FunctionAgent并传入工具列表、模型和系统提示词，最后调用Agent的run方法。输出如下，可以看出Agent借助LLM成功调用了传入的工具：

execute add 1 + 1 = 2

LLM提示词在任何以LLM为主的应用中都至关重要，查看示例的Agent在调用LLM时提示词：

首先，在代码最上面引入（以此打开Debug日志）：

import logging logging.basicConfig(level=logging.DEBUG)

然后运行查看完整提示词（摘取部分内容）：

# 第一次输入模型提示词 [{'role': 'system', 'content': 'You are an agent that can perform basic mathematical operations using tools.'}, {'role': 'user', 'content': 'what is 1 add 1'}] 'tools': [{'type': 'function', 'function': {'name': 'multiply', 'description': 'multiply(a: float, b: float) -> float\nMultiply two numbers and returns the product', 'parameters': {'properties': {'a': {'title': 'A', 'type': 'number'}, 'b': {'title': 'B', 'type': 'number'}}, 'required': ['a', 'b'], 'type': 'object', 'additionalProperties': False}, 'strict': False}}, {'type': 'function', 'function': {'name': 'add', 'description': 'add(a: float, b: float) -> float\nAdd two numbers and returns the sum', 'parameters': {'properties': {'a': {'title': 'A', 'type': 'number'}, 'b': {'title': 'B', 'type': 'number'}}, 'required': ['a', 'b'], 'type': 'object', 'additionalProperties': False}, 'strict': False}}] # 第二次输入给模型提示词， # 可以看出第一次模型输出的assistant角色的tool_calls内容 [{'role': 'system', 'content': 'You are an agent that can perform basic mathematical operations using tools.'}, {'role': 'user', 'content': 'what is 1 add 1'}, {'role': 'assistant', 'content': "I'll calculate 1 + 1 for you.", 'tool_calls': [{'index': 0, 'id': 'call_00_HZ4zYEy8Id5u6hsh7EmUHJ8j', 'function': {'arguments': '{"a": 1, "b": 1}', 'name': 'add'}, 'type': 'function'}]}, {'role': 'tool', 'content': '2', 'tool_call_id': 'call_00_HZ4zYEy8Id5u6hsh7EmUHJ8j'}]

示例中使用了自定义函数作为工具，还可以使用LlamaIndex集成的工具，通过访问https://llamahub.ai/?tab=tools方便查找，提供了使用步骤，例如通过使用DuckDuckGoSearchToolSpec工具实现搜索功能，如下所示：

# 使用集成的工具 pip install llama-index-tools-duckduckgo from llama_index.tools.duckduckgo import DuckDuckGoSearchToolSpec dd_tools = DuckDuckGoSearchToolSpec().to_tool_list() dd_tools.extend([add, multiply]) # 定义Agent的tools tools=dd_tools

Agent也是基于Workflow的事件驱动的，上面的示例通过增加调用agent的events属性，可以打印出所有相关的事件，它们被定义在BaseWorkflowAgent基类中。通过BaseWorkflowAgent类定义的step可以得出，流程涉及事件及路径是：AgentWorkflowStartEvent -> AgentInput -> AgentSetup -> AgentOutput -> StopEvent | AgentInput | ToolCall (ToolCall -> ToolCallResult -> StopEvent | AgentInput ) | None。

竖线表示或的关系，圆括号可以理解为子流程。

Agent中使用上下文

在上一篇Workflow中已经介绍过关于Context诸多用法，在Agent中可以通过Context实现短期或者长期的记忆，短期记忆通过指定和使用Agent的上下文即可，长期记忆需要借助Context的序列化和外部数据库进行存储。

# 额外导入包，和Workflow一样的 from llama_index.core.workflow import Context # 定义上下文 ctx = Context(agent) # 执行 async def main(): result = await agent.run(user_msg="Hi, my name is DJ!", ctx=ctx) print(result) print("-" * 10) response2 = await agent.run(user_msg="What's my name?", ctx=ctx) print(response2) if __name__ == "__main__": import asyncio asyncio.run(main())

示例是在上文示例的基础上更新的，定义了Context作为上下文，并在agent调用run的时候传入ctx参数，输出如下：

Hi DJ! Nice to meet you. I'm here to help you with basic mathematical operations. I can add or multiply numbers for you. What would you like me to help you with today? ---------- Your name is DJ! You introduced yourself at the beginning of our conversation.

Human in the loop

人类参与循环（简写HITL）描述的是一种人机交互的过程。在Agent中，通过人类参与中间特定环节的输入和决策，来驱动Agent按预期的路径执行。

# 人类介入 from llama_index.core.agent.workflow import FunctionAgent from llama_index.llms.deepseek import DeepSeek from llama_index.core.workflow import( InputRequiredEvent, HumanResponseEvent, Context, ) llm = DeepSeek(model="deepseek-chat", api_key="sk-...") async def add_money(ctx: Context) -> str: """A task for add money""" print(ctx.to_dict()) print("starting add_money task") question = "Are you sure you want to proceed? (yes/no): " # 标准写法：只用 wait_for_event，它自己会把 waiter_event 写到流里 response = await ctx.wait_for_event( HumanResponseEvent, waiter_id=question, waiter_event=InputRequiredEvent( prefix=question, user_name="User", # 关键：带 user_name ), requirements={"user_name": "User"}, # 关键：匹配条件 ) print(f"response:{response.to_dict()}") # 根据输入处理 if response.response.strip().lower() == "yes": print("add_money response yes") return"add_money task completed successfully." else: print("add_money response no") return"add_money task aborted." # 使用 DeepSeek LLM 创建 workflow agent = FunctionAgent( tools=[add_money], llm=llm, # 使用 DeepSeek system_prompt="""You are a helpful assistant that can run add_money tools.""" ) # 执行 async def main(): handler = agent.run(user_msg="Use the add_money tool to proceed with the me 100 RMB.") async for event in handler.stream_events(): # if isinstance(event, AgentStream): # print(f"{event.delta}", end="", flush=True) if isinstance(event, InputRequiredEvent): print("需要人工介入:") response = input(event.prefix) print(f"人工输入:{response}") handler.ctx.send_event( HumanResponseEvent( response=response, user_name=event.user_name, # 用事件里的名字，和 requirements 对上 ) ) print("-"*10) # 获取最终结果 response = await handler print(f"结果: {response}") while True: await asyncio.sleep(1) if __name__ == "__main__": import asyncio asyncio.run(main())

示例首先定义了add_money工具函数，代表这是一个危险动作，需要人介入确认，工具中调用上下文Context的wait_for_event方法设置一个等待的事件HumanResponseEvent，并会发起一个InputRequiredEvent事件；而后定义FunctionAgent指定了工具、模型和系统提示词；最后调用异步迭代器获取stream_events的事件，获取InputRequiredEvent事件后，引导人类输入并将输入结果包装在HumanResponseEvent中发送。

注意，HumanResponseEvent会重新唤起tool（这里是add_money函数）的重新调用，即从头开始执行，示例输出如下：

starting add_money task 需要人工介入: Are you sure you want to proceed? (yes/no): yes 人工输入:yes starting add_money task response:{'response': 'yes', 'user_name': 'User'} add_money response yes ---------- 结果: The add_money task has been completed successfully. The operation to add 100 RMB has been processed.

踩一个坑

HITL示例执行一直非预期，输入yes后，Agent直接回复并结束，测试时使用的LlamaIndex是0.14.4版本，最后升级到0.14.10解决

大致原因：

wait_for_event 当前版本不再「阻塞等待」，而是通过抛出一个 “等待事件”专用异常来暂停当前步骤；但由于Agent内部工具执行有代码误用了except Exception，把这个异常吞掉了，导致工作流无法在等待人类输入后重回等待点。

相关issue：

https://github.com/run-llama/llama_index/pull/20173

相关代码：

https://github.com/run-llama/llama_index/pull/20173/commits/e1855c6d47b37c7cef40de9b9342d37924eee48d

查看版本：pip index versions llama-index-core

安装最新版：pip install --upgrade llama-index-core

多智能体

LlamaIndex框架定义了三种使用多Agent的模式。

1、使用内置的AgentWorkflow

AgentWorkflow继承至Workflow，通过指定多个Agent和一个root Agent，实现多智能体协作和自动交接，可以快速上手体验。

agent_workflow = AgentWorkflow( agents=[add_agent, multiply_agent], root_agent=add_agent.name, verbose=True, ) result = await agent_workflow.run("what is 2 multiply 3")

2、使用工作流编排

工作流编排模式，定义常规的Workflow，然后通过将子Agent当作工具进行调用，借助Workflow执行路径更加稳定。

3、自定义执行计划

自定义方式最为灵活，可以自定义模型调用、解析响应、业务判断、执行路径等。

篇幅原因，自定义模式示例参考官方文档：

https://developers.llamaindex.ai/python/framework/understanding/agent/multi_agent/

选型建议：

从AgentWorkflow 起步 -> 需求变复杂时切换到 Orchestrator -> 真正落地生产时选 Custom Planner。

结构化输出

在一些基于LLM的应用中，需要LLM输出结构化数据便于继续处理后续业务，对于结构化输出，一些支持FunctionCall的LLM输出会更加稳定，而不支持的LLM就要靠提示词输出响应并解析文本响应内容。

在LlamaIndex的Agent调用流程中，支持两种结构化输出方式，不管哪一种，Agent的执行流程最后会经过llama_index.core.agent.workflow.base_agent.BaseWorkflowAgent.parse_agent_output处理Agent的结果。此处会判断Agent是否设置了结构化输出，如果有则会进一步处理，示例如下：

# 结构化输出 # 1、方式一，使用output_cls，指定输出模型（Pydantic BaseModel），输出为结构化数据 # 2、方式二，使用structured_output_fn自定义解析函数，输入是一些列的ChatMessage，输出为结构化数据 from llama_index.core.agent import FunctionAgent from llama_index.llms.deepseek import DeepSeek from pydantic import BaseModel, Field import json from llama_index.core.llms import ChatMessage from typing import List, Dict, Any def multiply(a: float, b: float) -> float: """Multiply two numbers and returns the product""" print("execute multiply") return a * b ## define structured output format and tools class MathResult(BaseModel): operation: str = Field(description="the performed operation") result: int = Field(description="the result of the operation") # 自定义格式化函数 async def structured_output_parsing( messages: List[ChatMessage], ) -> Dict[str, Any]: # 内部：StructuredLLM # 重要：内部实际上是再调用一次模型，生产结构化数据： # llama_index.llms.openai.base.OpenAI._stream_chat -》 openai.resources.chat.completions.completions.AsyncCompletions.create struct_llm = llm.as_structured_llm(MathResult) messages.append( ChatMessage( role="user", content="Given the previous message history, structure the output based on the provided format.", ) ) response = await struct_llm.achat(messages) return json.loads(response.message.content) llm = DeepSeek(model="deepseek-chat", api_key="sk-...") workflow = FunctionAgent( tools=[multiply], llm=llm, system_prompt="You are an agent that can perform basic mathematical operations using tools.", output_cls=MathResult, # 第一种实现方式 # structured_output_fn=structured_output_parsing, # 第二种实现方式 verbose=True, ) # 执行 async def main(): result = await workflow.run("what is 2 multi 3") # print(type(result)) # <class 'llama_index.core.agent.workflow.workflow_events.AgentOutput'> print(result) # 获取结构化结果 print(result.structured_response) # {'operation': '2 * 3', 'result': 6} print(result.get_pydantic_model(MathResult)) # operation='2 * 3' result=6 if __name__ == "__main__": import asyncio asyncio.run(main())

示例代码中定义了两种方式：

方式一使用output_cls，指定输出模型（Pydantic BaseModel），输出为结构化数据。

方式二使用structured_output_fn自定义解析函数，自定义函数输入是一些列的ChatMessage，输出为结构化数据。

方式一、使用output_cls：

内部会结合历史对话自动生成提示词，多调用一次LLM输出结构化数据，LlamaIndex内部实现如下：

# LlamaInde定义如下 # llama_index.core.agent.utils.generate_structured_response async def generate_structured_response( messages: List[ChatMessage], llm: LLM, output_cls: Type[BaseModel] ) -> Dict[str, Any]: xml_message = messages_to_xml_format(messages) structured_response = await llm.as_structured_llm( output_cls, ).achat(messages=xml_message) return cast(Dict[str, Any], json.loads(structured_response.message.content))

最终发起LLM调用时，判断是否支持FunctionCall：

llama_index.core.llms.function_calling.FunctionCallingLLM.apredict_and_call，debug如下所示：

方式二、使用structured_output_fn

调用的自定义函数，函数内采用的仍然是通过自定义提示词，调用LLM生成结构化数据，也可以是其它方式，自定义解析函数如下：

async def structured_output_parsing( messages: List[ChatMessage], ) -> Dict[str, Any]: struct_llm = llm.as_structured_llm(MathResult) messages.append( ChatMessage( role="user", content="Given the previous message history, structure the output based on the provided format.", ) ) response = await struct_llm.achat(messages) return json.loads(response.message.content)

按示例的方式，最终还是会到方式一相同的LLM处理逻辑完成结构化输出：

04 总结

总体而言，LlamaIndex 通过事件驱动的 Workflow 与 Agent 架构，成功打通了数据获取、知识处理、模型交互之间的完整链路。其内置的 Agent 体系不仅支持工具调用、多 Agent 协同、HITL（人类参与）等复杂能力，结合事件驱动也提供了足够的开放性与灵活性。

与此同时，LlamaIndex丰富的数据连接器生态与LlamaHub的便捷检索与集成方式，让开发者无需“重复造轮子”，即可快速构建业务级 RAG 与智能体应用。

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

第一阶段（10天）：初阶应用

该阶段让大家对大模型 AI有一个最前沿的认识，对大模型 AI 的理解超过 95% 的人，可以在相关讨论时发表高级、不跟风、又接地气的见解，别人只会和 AI 聊天，而你能调教 AI，并能用代码将大模型和业务衔接。

• 大模型 AI 能干什么？
• 大模型是怎样获得「智能」的？
• 用好 AI 的核心心法
• 大模型应用业务架构
• 大模型应用技术架构
• 代码示例：向 GPT-3.5 灌入新知识
• 提示工程的意义和核心思想
• Prompt 典型构成
• 指令调优方法论
• 思维链和思维树
• Prompt 攻击和防范
• …

第二阶段（30天）：高阶应用

该阶段我们正式进入大模型 AI 进阶实战学习，学会构造私有知识库，扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架，抓住最新的技术进展，适合 Python 和 JavaScript 程序员。

• 为什么要做 RAG
• 搭建一个简单的 ChatPDF
• 检索的基础概念
• 什么是向量表示（Embeddings）
• 向量数据库与向量检索
• 基于向量检索的 RAG
• 搭建 RAG 系统的扩展知识
• 混合检索与 RAG-Fusion 简介
• 向量模型本地部署
• …

第三阶段（30天）：模型训练

恭喜你，如果学到这里，你基本可以找到一份大模型 AI相关的工作，自己也能训练 GPT 了！通过微调，训练自己的垂直大模型，能独立训练开源多模态大模型，掌握更多技术方案。

到此为止，大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗？

• 为什么要做 RAG
• 什么是模型
• 什么是模型训练
• 求解器 & 损失函数简介
• 小实验2：手写一个简单的神经网络并训练它
• 什么是训练/预训练/微调/轻量化微调
• Transformer结构简介
• 轻量化微调
• 实验数据集的构建
• …

第四阶段（20天）：商业闭环

对全球大模型从性能、吞吐量、成本等方面有一定的认知，可以在云端和本地等多种环境下部署大模型，找到适合自己的项目/创业方向，做一名被 AI 武装的产品经理。

• 硬件选型
• 带你了解全球大模型
• 使用国产大模型服务
• 搭建 OpenAI 代理
• 热身：基于阿里云 PAI 部署 Stable Diffusion
• 在本地计算机运行大模型
• 大模型的私有化部署
• 基于 vLLM 部署大模型
• 案例：如何优雅地在阿里云私有部署开源大模型
• 部署一套开源 LLM 项目
• 内容安全
• 互联网信息服务算法备案
• …

学习是一个过程，只要学习就会有挑战。天道酬勤，你越努力，就会成为越优秀的自己。

如果你能在15天内完成所有的任务，那你堪称天才。然而，如果你能完成 60-70% 的内容，你就已经开始具备成为一名大模型 AI 的正确特征了。

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】