一、核心概念解析
1.1 LlamaIndex框架概述
LlamaIndex是一个强大的数据框架，专为构建基于大型语言模型（LLM）的应用程序而设计。它提供了从数据摄取、索引构建到查询执行的全套工具，特别适合构建智能代理系统。
核心组件包括：
- 数据连接器（Data Connectors）：用于从各种数据源（如PDF、文档、数据库等）读取数据
- 数据索引（Data Indexes）：将数据转换为结构化格式，便于高效检索
- 查询引擎（Query Engines）：处理用户查询并生成响应
- 智能代理（Agents）：能够使用工具执行复杂任务的自主系统
- 工作流（Workflows）：协调多个组件和步骤的执行流程
1.2 智能代理（Agent）核心概念
智能代理是能够理解目标、规划步骤并使用工具执行任务的系统。在LlamaIndex中，智能代理具有以下特点：
- 自主性：能够自主决定使用哪些工具以及如何使用
- 反应性：能够响应环境变化和用户输入
- 主动性：能够主动采取行动以达成目标
- 社交能力：能够与其他代理或系统进行交互
LlamaIndex提供了多种代理类型，如FunctionAgent、ReActAgent等，每种类型适用于不同的场景。
1.3 工具（Tools）概念
工具是智能代理执行特定任务的能力扩展。LlamaIndex支持多种类型的工具：
- 检索工具（Retriever Tools）：用于从知识库中检索信息
- 函数工具（Function Tools）：执行特定计算或操作
- 查询工具（Query Tools）：执行复杂查询
- API工具（API Tools）：与外部服务交互
1.4 工作流（Workflow）概念
工作流是定义多个步骤执行顺序和条件的框架。在LlamaIndex中，工作流由以下元素组成：
- 事件（Events）：工作流中传递的信息单元
- 步骤（Steps）：处理事件并产生新事件的函数
- 上下文（Context）：在工作流中共享的状态信息
二、工具集成
2.1 检索工具
检索工具允许代理从知识库中获取信息：
from llama_index.core.tools import RetrieverTool
def create_retriever_tool(index: VectorStoreIndex, name: str, description: str): """创建检索工具""" # 创建检索器 retriever = index.as_retriever(similarity_top_k=3)
# 创建检索工具 retriever_tool = RetrieverTool.from_defaults( retriever=retriever, name=name, description=description )
return retriever_tool
2.2 函数工具
函数工具允许代理执行特定计算或操作：
from llama_index.core.tools import FunctionTool
# 定义计算函数def multiply(a: float, b: float) -> float: """两个数相乘""" return a * b
def add(a: float, b: float) -> float: """两个数相加""" return a + b
def get_weather(city: str) -> str: """获取指定城市的天气信息（模拟）""" # 这里可以接入真实的天气API weather_data = { "北京": "晴天，温度25°C", "上海": "多云，温度28°C", "广州": "雨天，温度30°C" } return weather_data.get(city, f"抱歉，没有{city}的天气信息")
# 创建函数工具multiply_tool = FunctionTool.from_defaults(fn=multiply)add_tool = FunctionTool.from_defaults(fn=add)weather_tool = FunctionTool.from_defaults(fn=get_weather)
2.3 API工具
API工具允许代理与外部服务交互：
import requestsfrom llama_index.core.tools import FunctionTool
def search_web(query: str) -> str: """使用网络搜索API搜索信息（模拟）""" # 这里可以接入真实的搜索API，如Google Search API、Bing Search API等 # 以下是模拟实现 return f"关于'{query}'的搜索结果：这是一个模拟的搜索结果。"
# 创建API工具search_tool = FunctionTool.from_defaults(fn=search_web)
三、智能代理创建
导入一些包，可自行删减。
pip install llama-index-corepip install llama-index-llms-openaipip install llama-index-embeddings-huggingfacepip install llama-index-vector-stores-chromapip install chromadb
3.1 基础智能代理
现在，我们可以创建一个基础的智能代理：
from llama_index.core.agent.workflow import FunctionAgent
def create_basic_agent(tools, system_prompt=None): """创建基础智能代理""" if system_prompt is None: system_prompt = "你是一个有用的助手，可以使用提供的工具来回答问题和执行任务。"
agent = FunctionAgent( tools=tools, llm=Settings.llm, system_prompt=system_prompt, verbose=True # 启用详细输出，便于调试 )
return agent
3.2 高级智能代理
我们可以创建一个更高级的智能代理，结合检索和计算能力：
def create_advanced_agent(index: VectorStoreIndex): """创建高级智能代理""" # 创建工具 retriever_tool = create_retriever_tool( index, name="knowledge_retriever", description="用于从知识库中检索相关信息" )
# 添加函数工具 tools = [ retriever_tool, multiply_tool, add_tool, weather_tool, search_tool ]
# 创建系统提示 system_prompt = """ 你是一个智能助手，可以使用多种工具来回答问题和执行任务。
当用户提问时，你应该： 1. 首先尝试使用knowledge_retriever工具从知识库中查找相关信息 2. 如果需要计算，使用multiply或add工具 3. 如果需要天气信息，使用get_weather工具 4. 如果需要网络搜索，使用search_web工具
请根据用户需求选择合适的工具，并综合所有信息给出准确、有用的回答。 """
# 创建代理 agent = FunctionAgent( tools=tools, llm=Settings.llm, system_prompt=system_prompt, verbose=True )
return agent
3.3 代理与工作流结合
我们可以将代理集成到工作流中，创建更复杂的系统：
class AgentEvent(Event): response: str
class AgentWorkflow(Workflow): def __init__(self, agent, **kwargs): super().__init__(**kwargs) self.agent = agent
@step async def process_with_agent(self, ev: StartEvent) -> AgentEvent: """使用代理处理输入""" query = ev.get("query")
# 使用代理处理查询 response = await self.agent.run(query)
return AgentEvent(response=str(response))
@step async def finalize(self, ev: AgentEvent) -> StopEvent: """最终化结果""" return StopEvent(result=ev.response)
async def run_agent_workflow(agent, query: str): """运行代理工作流""" workflow = AgentWorkflow(agent=agent, timeout=30, verbose=False) result = await workflow.run(query=query) return result
3.4 完整的示例
（1）fun_coll.py中创建所需的方法。
from llama_index.core.tools import FunctionTool# 定义一些简单的工具函数def multiply(a: float, b: float) -> float: """两个数相乘""" print(f"fun_coll- multiply 中的参数：a={a}, b={b}") return a * b
def add(a: float, b: float) -> float: """两个数相加""" print(f"fun_coll- add 中的参数：a={a}, b={b}") return a + b
def divide(a: float, b: float) -> float: """两个数相除，除数不能为零""" if b == 0: raise ValueError("除数不能为零") return a / b
def get_weather(city: str) -> str: """获取指定城市的天气信息（模拟）""" weather_data = { "北京": "晴天，温度25°C", "上海": "多云，温度28°C", "广州": "雨天，温度30°C" } print(f"fun_coll- get_weather 中的参数：city={city}") return weather_data.get(city, f"抱歉，没有{city}的天气信息")
def search(query: str) -> str: """模拟搜索功能，根据查询词返回结果""" search_results = { "Python": "Python是一种高级编程语言", "AI": "AI指人工智能", "LlamaIndex": "LlamaIndex是一个用于构建LLM应用的框架" } ret= search_results.get(query, f"未找到关于 '{query}' 的结果") print(f"fun_coll- search 中的返回值：{ret}") return ret
# 创建工具实例multiply_tool = FunctionTool.from_defaults(fn=multiply)add_tool = FunctionTool.from_defaults(fn=add)divide_tool = FunctionTool.from_defaults(fn=divide)weather_tool = FunctionTool.from_defaults(fn=get_weather)search_tool = FunctionTool.from_defaults(fn=search)
（2）fun_retiever.py中创建检索器。
from llama_index.core.tools import RetrieverToolfrom llama_index.core import ( VectorStoreIndex, SimpleDirectoryReader, Response,)from llama_index.core.node_parser import SentenceSplitter
# ......这里造查询引擎......documents = SimpleDirectoryReader(input_files=[r"D:\muxue\docs\jianlai.txt"]).load_data()# create vector indexsplitter = SentenceSplitter(chunk_size=512)vector_index = VectorStoreIndex.from_documents( documents, transformations=[splitter])query_engine = vector_index.as_query_engine()
# 创建工具def create_retriever_tool(index: VectorStoreIndex, name: str, description: str): """创建检索工具""" # 创建检索器 retriever = index.as_retriever(similarity_top_k=3)
# 创建检索工具 tool = RetrieverTool.from_defaults( retriever=retriever, name=name, description=description )
return tool
my_retriever_tool=create_retriever_tool(index=vector_index,name="knowledge_retriever",description="用于从知识库中检索相关信息")
（3）my_agent.py负责创建智能体。
# 测试运行agentimport asyncio
from llama_index.core.agent.workflow import FunctionAgentfrom common.bedrock_model_loader import model_loaders
from fun_coll import multiply_tool,add_tool,weather_tool,search_toolfrom fun_retriever import my_retriever_tool
async def create_advanced_agent(): """创建高级智能代理""" # 添加函数工具 tools = [ my_retriever_tool, multiply_tool, add_tool, weather_tool, search_tool ]
# 创建系统提示 system_prompt = """ 你是一个智能助手，可以使用多种工具来回答问题和执行任务。
当用户提问时，你应该： 1. 首先尝试使用knowledge_retriever工具从知识库中查找相关信息 2. 如果需要计算，使用multiply或add工具 3. 如果需要天气信息，使用get_weather工具 4. 如果需要网络搜索，使用search_web工具
请根据用户需求选择合适的工具，并综合所有信息给出准确、有用的回答。 """
agent = FunctionAgent( tools=tools, #llm=model_loaders.llm, # 使用model_loaders中的llm system_prompt=system_prompt, verbose= False # True # 启用详细输出，便于调试 )
return agent
（4）main_test.py中调用智能体。
import sysfrom pathlib import Path
# 添加项目根目录到Python路径project_root = Path(__file__).resolve().parent.parentsys.path.insert(0, str(project_root))
# ================== 初始化 大模型 Langfuse ==================from common.muxue_model_loader import model_loadersimport common.langfuse_init_client# ================== 初始化大模型 Langfuse end ==================
import asynciofrom my_agent import create_advanced_agent
async def test_agent(): print("开始测试智能代理...") agent = await create_advanced_agent()
# 测试 0: 从知识库中检索答案 query1 = "成平安是什么性格？" response1 = await agent.run(query1) print(f"查询1: {query1}") print(f"回答1: {response1}") print("-" * 50)
# 测试1: 数学计算 query1 = "计算 123 乘以 456 等于多少？" response1 = await agent.run(query1) print(f"查询1: {query1}") print(f"回答1: {response1}") print("-" * 50)
# 测试2: 天气查询 query2 = "北京的天气怎么样？" response2 = await agent.run(query2) print(f"查询2: {query2}") print(f"回答2: {response2}") print("-" * 50)
# 测试3: 组合任务 query3 = "先计算 10 加上 20，然后将结果乘以 3" response3 = await agent.run(query3) print(f"查询3: {query3}") print(f"回答3: {response3}") print("-" * 50)
# 运行测试if __name__ == "__main__": asyncio.run(test_agent())
执行结果如下：
查询1: 成平安是什么性格？回答1: 根据知识库中的信息，成平安是一个勤劳、坚韧、不轻言放弃的人。他自学拉坯技艺，即使在贫困和被人嘲笑的情况下，也坚持不懈地练习。他早起晚睡，养成了良好的作息习惯。此外，他还表现出了一定的谦逊和内敛，在被宋集薪嘲笑时，他没有反驳，而是默不作声。总的来说 ，成平安是一个坚韧、勤劳、谦逊的人。--------------------------------------------------fun_coll- multiply 中的参数：a=123, b=456查询1: 计算 123 乘以 456 等于多少？回答1: 123 乘以 456 等于 56088。--------------------------------------------------fun_coll- get_weather 中的参数：city=北京查询2: 北京的天气怎么样？回答2: 北京现在是晴天，温度为25°C。--------------------------------------------------fun_coll- add 中的参数：a=10, b=20fun_coll- multiply 中的参数：a=30, b=3查询3: 先计算 10 加上 20，然后将结果乘以 3回答3: 10 加上 20 的结果是 30，然后将 30 乘以 3 的结果是 90。
将项目拆分为多个.py 文件，核心是遵循 “单一职责” 原则：让每个文件专注处理特定功能模块，既避免代码混杂导致的相互干扰，也能显著降低后续扩展、查阅与理解的成本，让项目结构更清晰、维护更高效。
[登录查看剩余 70% 内容](javascript:void (0);)