LlamaIndex 开发智能体 Agents 要点解析

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

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)

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

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

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

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

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)

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="用于从知识库中检索相关信息")

# 测试运行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

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。