一、核心概念解析

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 requests
from 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)

三、智能代理创建

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 RetrieverTool
from 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 index
splitter = 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负责创建智能体。

# 测试运行agent
import asyncio

from llama_index.core.agent.workflow import FunctionAgent
from common.bedrock_model_loader import model_loaders

from fun_coll import multiply_tool,add_tool,weather_tool,search_tool
from 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 sys
from pathlib import Path

# 添加项目根目录到Python路径
project_root = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(project_root))

# ================== 初始化 大模型 Langfuse ==================
from common.muxue_model_loader import model_loaders
import common.langfuse_init_client
# ================== 初始化大模型 Langfuse  end ==================

import asyncio
from 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=20
fun_coll- multiply 中的参数：a=30, b=3
查询3: 先计算 10 加上 20，然后将结果乘以 3
回答3: 10 加上 20 的结果是 30，然后将 30 乘以 3 的结果是 90。
--------------------------------------------------