AI Agent开发实战:LlamaIndex框架深度应用指南
本文系统介绍了使用LlamaIndex构建AI Agent的方法,从核心优势到实战开发。LlamaIndex凭借数据为核心的设计理念、极简工具封装、强大的RAG集成等特性,成为开发企业级AI Agent的理想选择。文章详细解析了LlamaIndex Agent的四大核心组件:LLM大脑、工具集、记忆系统和工作流引擎,并通过一个"个人助理Agent"的完整案例,展示了如何整合数学
本文将系统地讲解如何在 AI Agent 编写中运用 LlamaIndex,从核心概念到实战代码,从单 Agent 开发到多 Agent 协作,帮助你快速掌握生产级 AI Agent 的开发方法。
一、为什么选择 LlamaIndex 构建 AI Agent?
市面上有很多 Agent 开发框架(如 LangChain、AutoGPT 等),但 LlamaIndex 在以下几个方面具有独特优势:
- 数据即核心的设计理念:LlamaIndex 天生擅长处理结构化和非结构化数据,能够轻松将 PDF、SQL 数据库、Notion 笔记、API 接口等多种数据源转化为 Agent 可理解的知识。如果你的 Agent 需要频繁与企业内部数据交互,LlamaIndex 是最佳选择。
- 极简的工具封装:只需为 Python 函数添加类型注解,LlamaIndex 就能自动将其转换为 Agent 可调用的工具,无需复杂的配置。同时,LlamaHub 提供了数百个预构建的工具,覆盖了常见的 API 和服务。
- 强大的 RAG 集成:LlamaIndex 的核心优势在于检索增强生成(RAG),你可以将 RAG 管道直接作为 Agent 的一个工具,让 Agent 在需要时自动检索相关知识,有效减少幻觉。
- 灵活的工作流编排:2025 年推出的 Workflows 系统彻底重构了 Agent 的执行逻辑,支持复杂的多步骤任务、状态管理和多 Agent 协作,代码可读性和可维护性大幅提升。
- 丰富的 LLM 支持:LlamaIndex 兼容几乎所有主流的大语言模型,包括 OpenAI、Anthropic、Google Gemini、DeepSeek 等,甚至支持本地部署的开源模型,让你可以根据需求灵活选择。
二、LlamaIndex Agent 核心架构
在 LlamaIndex 中,一个完整的 Agent 系统由四个核心组件构成:大语言模型(LLM)、工具集(Tools)、记忆系统(Memory) 和 工作流引擎(Workflow)。
2.1 大语言模型(LLM)
LLM 是 Agent 的 “大脑”,负责理解用户意图、进行推理决策和生成自然语言响应。LlamaIndex 提供了统一的 LLM 接口,你可以轻松切换不同的模型提供商。
from llama_index.llms.openai import OpenAI
from llama_index.llms.deepseek import DeepSeek
from llama_index.llms.google_genai import Gemini
# 使用 OpenAI GPT-4o
llm = OpenAI(model="gpt-4o", api_key="your-api-key")
# 使用 DeepSeek V3
llm = DeepSeek(model="deepseek-chat", api_key="your-api-key")
# 使用 Google Gemini 1.5 Pro
llm = Gemini(model="gemini-1.5-pro", api_key="your-api-key")
2.3 记忆系统(Memory)
记忆是 Agent 的 “第二大脑”,让 Agent 能够记住过去的对话内容和任务执行结果,保持交互的连续性。LlamaIndex 提供了分层的记忆系统:
- 短期记忆:存储当前会话的聊天历史,默认使用 ChatMemoryBuffer
- 长期记忆:将重要信息持久化存储,支持事实提取和语义检索
from llama_index.core.memory import Memory, ChatMemoryBuffer
from llama_index.core.memory.blocks import FactExtractionMemoryBlock
# 基础短期记忆
short_term_memory = ChatMemoryBuffer.from_defaults(token_limit=40000)
# 高级记忆系统(支持长期记忆)
long_term_block = FactExtractionMemoryBlock(llm=llm)
memory = Memory.from_defaults(
session_id="user-123",
token_limit=40000,
long_term_memory_blocks=[long_term_block]
)
2.4 工作流引擎(Workflow)
Workflows 是 LlamaIndex 2025 年推出的核心特性,它将 Agent 的执行过程抽象为一系列步骤和事件,支持复杂的任务编排、状态管理和错误处理。
LlamaIndex 提供了预构建的 AgentWorkflow,专门用于快速开发单 Agent 和多 Agent 系统:
from llama_index.core.agent.workflow import AgentWorkflow, FunctionAgent
# 创建一个简单的 FunctionAgent
agent = FunctionAgent(
name="Assistant",
description="A helpful assistant that can answer questions and perform calculations",
llm=llm,
tools=[calculator_tool, rag_tool]
)
# 创建工作流并运行
workflow = AgentWorkflow(agents=[agent], root_agent="Assistant")
response = await workflow.run(user_msg="What is 123 * 456?")
print(response)
三、从零开始构建第一个 LlamaIndex Agent
让我们通过一个完整的实战案例,学习如何构建一个具备数学计算和文档检索能力的 “个人助理 Agent”。
3.1 环境准备
首先,安装必要的依赖包:
pip install llama-index llama-index-llms-openai python-dotenv
创建一个 .env 文件,存储你的 API 密钥:
OPENAI_API_KEY=your-openai-api-key
3.2 定义工具集
我们将为 Agent 定义两个工具:一个计算器工具和一个文档检索工具。
import os
from dotenv import load_dotenv
from llama_index.core.tools import FunctionTool, QueryEngineTool
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.llms.openai import OpenAI
# 加载环境变量
load_dotenv()
# 初始化 LLM
llm = OpenAI(model="gpt-4o-mini", api_key=os.getenv("OPENAI_API_KEY"))
# 1. 计算器工具
def add(a: float, b: float) -> float:
"""Add two numbers and returns the sum"""
return a + b
def subtract(a: float, b: float) -> float:
"""Subtract b from a and returns the difference"""
return a - b
def multiply(a: float, b: float) -> float:
"""Multiply two numbers and returns the product"""
return a * b
def divide(a: float, b: float) -> float:
"""Divide a by b and returns the quotient"""
if b == 0:
raise ValueError("Cannot divide by zero")
return a / b
calculator_tools = [
FunctionTool.from_defaults(fn=add),
FunctionTool.from_defaults(fn=subtract),
FunctionTool.from_defaults(fn=multiply),
FunctionTool.from_defaults(fn=divide)
]
# 2. 文档检索工具
# 确保你有一个 ./data 目录,里面包含一些文档文件
if os.path.exists("./data"):
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents, llm=llm)
query_engine = index.as_query_engine(llm=llm)
rag_tool = QueryEngineTool.from_defaults(
query_engine=query_engine,
name="document_retrieval",
description="Useful for answering questions about the internal documents in the data directory"
)
else:
print("Warning: ./data directory not found, document retrieval tool will not be available")
rag_tool = None
3.3 初始化并运行 Agent
现在,我们将使用 ReActAgent(LlamaIndex 最常用的 Agent 类型)来整合这些工具:
from llama_index.core.agent import ReActAgent
# 准备工具列表
tools = calculator_tools
if rag_tool:
tools.append(rag_tool)
# 初始化 Agent
agent = ReActAgent.from_tools(
tools=tools,
llm=llm,
verbose=True, # 开启详细日志,查看 Agent 的思考过程
max_iterations=10 # 限制最大迭代次数,防止无限循环
)
# 运行 Agent
print("=== 个人助理 Agent 已启动 ===")
print("输入 'exit' 退出")
while True:
user_input = input("\n你: ")
if user_input.lower() == "exit":
print("Agent: 再见!")
break
response = agent.chat(user_input)
print(f"\nAgent: {response}")
3.4 测试 Agent
运行代码后,你可以尝试以下问题来测试 Agent 的能力:
- 数学计算:计算 1234 * 5678 + 9012
- 文档检索:根据文档内容,总结一下项目的主要目标
- 混合任务:先计算 2026 年有多少天,然后在文档中查找与这个数字相关的内容
你会看到 Agent 会自动分析问题,决定是否需要调用工具,然后根据工具的返回结果生成最终答案。
四、LlamaIndex Agent 高级特性
4.1 结构化输出
在很多实际应用中,我们需要 Agent 输出结构化的数据(如 JSON),以便后续程序处理。LlamaIndex 提供了两种结构化输出方式:
from pydantic import BaseModel, Field
from llama_index.core.agent import FunctionAgent
# 定义输出模型
class MathResult(BaseModel):
operation: str = Field(description="The operation performed")
input_values: list[float] = Field(description="The input values")
result: float = Field(description="The result of the operation")
# 方式一:使用 output_cls 参数
agent = FunctionAgent(
llm=llm,
tools=calculator_tools,
output_cls=MathResult
)
response = agent.run("Calculate 123 * 456")
print(f"Operation: {response.operation}")
print(f"Result: {response.result}")
4.2 多 Agent 协作
对于复杂的任务,单个 Agent 往往难以胜任。LlamaIndex 的 AgentWorkflow 支持轻松构建多 Agent 系统,让不同的专家 Agent 协作完成任务。
下面是一个 “报告生成系统” 的示例,包含三个专家 Agent:
from llama_index.core.agent.workflow import AgentWorkflow, FunctionAgent
from llama_index.tools.tavily import TavilyToolSpec
# 初始化网络搜索工具
tavily_tool = TavilyToolSpec(api_key="your-tavily-api-key")
search_tools = tavily_tool.to_tool_list()
# 1. 研究 Agent:负责搜索信息
research_agent = FunctionAgent(
name="ResearchAgent",
description="Search the web for information on a given topic and record notes",
system_prompt="""You are a research expert. Search the web for information on the given topic,
take detailed notes, and then hand off to the WriteAgent to write the report.""",
llm=llm,
tools=search_tools,
can_handoff_to=["WriteAgent"]
)
# 2. 写作 Agent:负责撰写报告
write_agent = FunctionAgent(
name="WriteAgent",
description="Write a comprehensive report based on the research notes",
system_prompt="""You are a professional writer. Use the research notes provided by the ResearchAgent
to write a well-structured, comprehensive report. After writing, hand off to the ReviewAgent for review.""",
llm=llm,
can_handoff_to=["ReviewAgent"]
)
# 3. 审核 Agent:负责审核报告
review_agent = FunctionAgent(
name="ReviewAgent",
description="Review the report and provide feedback or final approval",
system_prompt="""You are a strict editor. Review the report written by the WriteAgent.
If the report is good, return it as the final answer. If it needs improvement,
provide detailed feedback and hand off back to the WriteAgent.""",
llm=llm,
can_handoff_to=["WriteAgent"]
)
# 创建多 Agent 工作流
workflow = AgentWorkflow(
agents=[research_agent, write_agent, review_agent],
root_agent="ResearchAgent"
)
# 运行工作流
response = await workflow.run(
user_msg="Write a comprehensive report on the latest developments in AI agents in 2026"
)
print(response)
4.3 状态管理
在多步骤任务中,Agent 需要在不同步骤之间传递数据。LlamaIndex 的 Workflow 系统提供了内置的状态管理功能:
from llama_index.core.workflow import Context, StartEvent, StopEvent, step, Workflow
from llama_index.core.llms import ChatMessage
class ReportWorkflow(Workflow):
@step
async def research(self, ctx: Context, ev: StartEvent) -> StopEvent:
# 从事件中获取用户输入
topic = ev.topic
# 执行搜索
search_results = await tavily_tool.asearch(topic)
# 将搜索结果存储到状态中
await ctx.set("search_results", search_results)
# 进入下一步
return await self.write(ctx)
@step
async def write(self, ctx: Context) -> StopEvent:
# 从状态中获取搜索结果
search_results = await ctx.get("search_results")
# 生成报告
prompt = f"Write a report on {await ctx.get('topic')} based on the following information: {search_results}"
report = await llm.achat(prompt)
# 将报告存储到状态中
await ctx.set("report", report.response)
# 返回最终结果
return StopEvent(result=report.response)
# 运行工作流
workflow = ReportWorkflow()
result = await workflow.run(topic="AI agents in 2026")
print(result)
五、生产级 Agent 设计模式
5.1 路由模式
当你的 Agent 需要处理多种不同类型的任务时,可以使用路由模式。引入一个专门的 “路由 Agent”,它的唯一职责是根据用户输入,将任务分配给最合适的专家 Agent。
from llama_index.core.agent import ReActAgent
from llama_index.core.tools import QueryEngineTool
# 定义专家 Agent
math_agent = ReActAgent.from_tools(calculator_tools, llm=llm)
document_agent = ReActAgent.from_tools([rag_tool], llm=llm)
# 定义路由工具
def route_to_math_agent(query: str) -> str:
"""Use this tool when the user asks a math question or needs to perform calculations"""
return math_agent.chat(query).response
def route_to_document_agent(query: str) -> str:
"""Use this tool when the user asks a question about the internal documents"""
return document_agent.chat(query).response
# 创建路由 Agent
router_agent = ReActAgent.from_tools(
tools=[
FunctionTool.from_defaults(fn=route_to_math_agent),
FunctionTool.from_defaults(fn=route_to_document_agent)
],
llm=llm,
system_prompt="""You are a router. Your only job is to decide which agent to use based on the user's query.
Do not attempt to answer the question yourself. Always use one of the available tools."""
)
5.2 错误处理与重试
生产环境中,Agent 可能会遇到各种错误(如 API 调用失败、工具执行异常等)。LlamaIndex 支持在工作流中添加错误处理和重试逻辑:
from llama_index.core.workflow import Context, step, Workflow, StartEvent, StopEvent
class RobustWorkflow(Workflow):
@step
async def research(self, ctx: Context, ev: StartEvent) -> StopEvent:
retry_count = await ctx.get("retry_count", default=0)
try:
# 尝试执行搜索
search_results = await tavily_tool.asearch(ev.topic)
await ctx.set("retry_count", 0) # 成功后重置重试次数
return StopEvent(result=search_results)
except Exception as e:
await ctx.set("retry_count", retry_count + 1)
if retry_count < 3:
print(f"Search failed, retrying... (attempt {retry_count + 1})")
return await self.research(ctx, ev) # 重试
else:
raise RuntimeError(f"Search failed after 3 attempts: {e}")
5.3 可观测性
在生产环境中,监控 Agent 的运行状态至关重要。LlamaIndex 集成了多种可观测性工具,如 LangSmith、Phoenix 等:
from llama_index.core.callbacks import CallbackManager
from llama_index.callbacks.langsmith import LangSmithCallbackHandler
# 初始化 LangSmith 回调处理器
langsmith_handler = LangSmithCallbackHandler(
api_key="your-langsmith-api-key",
project_name="llamaindex-agent-demo"
)
# 创建回调管理器
callback_manager = CallbackManager([langsmith_handler])
# 初始化 Agent 时传入回调管理器
agent = ReActAgent.from_tools(
tools=tools,
llm=llm,
callback_manager=callback_manager
)
六、常见问题与解决方案
6.1 Agent 陷入无限循环
问题:Agent 不断重复调用同一个工具,无法完成任务。
解决方案:
- 设置 max_iterations 参数,限制 Agent 的最大迭代次数
- 优化工具描述,让 Agent 更清楚工具的使用场景
- 使用更强大的 LLM(如 GPT-4o 代替 GPT-4o-mini)
- 在系统提示词中明确要求 Agent 在获得足够信息后直接回答问题
6.2 Agent 不调用工具,直接回答问题
问题:Agent 忽略可用工具,仅凭自身知识回答问题,导致幻觉。
解决方案:
- 优化系统提示词,强调必须使用工具来获取准确信息
- 改进工具描述,明确说明工具的优势和适用场景
- 使用 FunctionAgent 代替 ReActAgent,它对函数调用的支持更好
- 在提示词中提供工具使用的示例
6.3 工具调用参数错误
问题:Agent 调用工具时传入了错误的参数类型或格式。
解决方案:
- 确保函数有清晰的类型注解和文档字符串
- 使用 Pydantic 模型定义工具的输入参数
- 在工具内部添加参数验证逻辑
- 使用支持结构化输出的 LLM
更多推荐




所有评论(0)