当场景复杂、涉及多模块功能时，单智能体难以满足需求，此时需构建多智能体系统：为特定领域任务创建专门智能体，让每个智能体各司其职，共同解决复杂问题。后续将逐步拆解其开发要点。
接下来，我们将逐步拆解 LlamaIndex 多智能体的开发流程，帮助大家系统掌握从需求分析到落地实现的完整方法。
开发步骤如下：
1、安装与环境
导包所需的包。
pip install -U llama-index llama-index-llms-openai
2、选择 LLM 与全局设置
muxue_model_loader.py文件代码：
from llama_index.llms.openai import OpenAIfrom llama_index.core import Settingsllm = OpenAI(model="gpt-4o-mini") # 可换任意支持函数调用的模型Settings.llm = llm # 让后续 Agent/Workflow 继承该 LLM
3、定义工具（Tools）
在 LlamaIndex 里，“工具”是智能体的手脚；最常用包装是 FunctionTool。LlamaIndex 将任意 Python 函数包装成工具，供 Agent 决策调用；这是 FunctionAgent 模式的基础。
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 * bdef add(a: float, b: float) -> float:"""两个数相加"""print(f"fun_coll- add 中的参数：a={a}, b={b}")return a + bdef divide(a: float, b: float) -> float:"""两个数相除，除数不能为零"""if b == 0:raise ValueError("除数不能为零")return a / bdef 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)
4、专门的Agent父类
from llama_index.core.agent.workflow import FunctionAgentclass SpecialistAgent:"""专业Agent基类"""def __init__(self, name: str, specialty: str, tools, system_prompt: str):self.name = nameself.specialty = specialtyself.agent = FunctionAgent(tools=tools,# llm=model_loaders.llm, # 使用model_loaders中的llmsystem_prompt=system_prompt,verbose=True)async def handle_query(self, query: str):"""处理查询"""print(f"[{self.name}] 处理查询: {query}")response = await self.agent.run(query)return str(response)
代码分析：
初始化方法中，设置智能体的名称，FunctionAgent接收多个工具tools，以及非常重要的提示词；verbose=True开启详细日志打印，若不需要则设置为False，或不写。
执行Agent的方法，handle_query做了一个简单的封装，便于统一管理和执行agent。
5、多智能体管理
创建一个多智能体的管理类，创建一个“协调器”代理，这个协调器本身不直接回答问题，而是像一个项目经理，负责将用户的请求（query）分派给最合适的“专家”代理来处理。
from llama_index.core.tools import FunctionToolfrom llama_index.core.agent.workflow import FunctionAgentfrom specialist_agent import SpecialistAgentclass MultiAgentSystem:"""多代理系统"""def __init__(self):self.agents = {}self.coordinator = Nonedef add_agent(self, agent: SpecialistAgent):"""添加代理"""self.agents[agent.name] = agentdef create_coordinator(self):"""创建协调器代理"""# 创建代理调用工具def call_agent(agent_name: str, query: str) -> str:"""调用指定的代理处理查询"""if agent_name in self.agents:# 注意：这里需要异步执行，但在同步函数中无法直接await# 我们将返回一个标记，表示需要调用哪个代理print(f"call_agent 中的返回值：CALL_AGENT:{agent_name}:{query}")return f"CALL_AGENT:{agent_name}:{query}"else:print(f"call_agent 错误：找不到名为 '{agent_name}' 的代理")return f"错误：找不到名为 '{agent_name}' 的代理"agent_caller_tool = FunctionTool.from_defaults(fn=call_agent)# 创建代理选择工具def select_agent(query: str) -> str:"""根据查询选择最合适的代理"""ret=""# 这里可以实现更复杂的代理选择逻辑if "计算" in query or "数学" in query:ret= "math_agent"elif "天气" in query:ret= "weather_agent"elif "搜索" in query or "网络" in query:ret= "search_agent"else:ret= "general_agent"print(f"select_agent中的返回值：{ret}")return retagent_selector_tool = FunctionTool.from_defaults(fn=select_agent)# 创建协调器系统提示coordinator_prompt = """你是一个代理协调器，负责将用户查询分配给最合适的专业代理。当收到用户查询时：1. 使用select_agent工具确定应该处理该查询的代理2. 使用call_agent工具调用选定的代理处理查询3. 重要：在调用call_agent时，必须原样传递用户查询，不要修改或改写查询内容4. 注意：你不需要直接返回最终响应给用户，原样返回call_agent的返回值可用的代理有：- math_agent: 专门处理数学计算- weather_agent: 专门处理天气查询- search_agent: 专门处理网络搜索- general_agent: 处理一般查询"""# 创建协调器代理self.coordinator = FunctionAgent(tools=[agent_selector_tool, agent_caller_tool],# llm=model_loaders.llm, # 使用model_loaders中的llmsystem_prompt=coordinator_prompt,verbose=True)async def process_query(self, query: str):"""处理查询"""if not self.coordinator:raise ValueError("协调器尚未创建，请先调用create_coordinator方法")print(f"多代理系统处理查询: {query}")# 获取协调器的初步响应coordinator_response = await self.coordinator.run(query)response_str = str(coordinator_response)# 检查是否需要调用专业代理if response_str.startswith("CALL_AGENT:"):# 解析代理名称和查询parts = response_str.split(":", 2)if len(parts) == 3:agent_name = parts[1]agent_query = parts[2]# 调用专业代理if agent_name in self.agents:agent_response = await self.agents[agent_name].handle_query(agent_query)return agent_responseelse:return f"错误：找不到名为 '{agent_name}' 的代理"# 如果不需要调用专业代理，直接返回协调器的响应return response_str
这个 MultiAgentSystem 类构建了一个框架，其中包含：
多个专家代理 (Specialist Agents)：系统中可以添加多个专门负责特定领域（如数学、天气、搜索）的代理。这些代理是从外部 specialist_agent 模块导入的，它们各自拥有处理特定任务的能力。
一个协调器代理 (Coordinator Agent)：这是一个特殊的代理，它的大脑（LLM）被赋予了两个工具：一个用来选择合适的专家，另一个用来调用这个专家；注意select_agent方法是一个非常简单的基于规则的路由逻辑。例如，如果查询中包含“计算”，它就返回 "math_agent"。在实际应用中，这里可以用更复杂的逻辑，甚至可以用另一个 LLM 来做更智能的选择。
一个处理流程 (Processing Workflow)：当用户查询进来时，首先由协调器进行分析和分派，然后系统再执行协调器的指令，调用相应的专家代理，并返回最终结果。
6、调用多智能体
主方法组装多个智能体，并且调用这些智能体。
# ================== 初始化 大模型 Langfuse ==================import common.muxue_model_loaderimport common.langfuse_init_client# ================== 初始化大模型 Langfuse end ==================import asynciofrom fun_coll import *from specialist_agent import SpecialistAgentfrom multi_agent_system import MultiAgentSystem# 使用示例async def multi_agent_example():# 创建多代理系统multi_agent_system = MultiAgentSystem()# 创建专业代理math_agent = SpecialistAgent(name="math_agent",specialty="数学计算",tools=[multiply_tool, add_tool],system_prompt="你是一个数学专家，专门处理各种数学计算问题。")weather_agent = SpecialistAgent(name="weather_agent",specialty="天气查询",tools=[weather_tool],system_prompt="你是一个天气专家，专门提供天气信息和预报。")search_agent = SpecialistAgent(name="search_agent",specialty="网络搜索",tools=[search_tool],system_prompt="你是一个搜索专家，专门进行网络搜索和信息检索。")general_agent = SpecialistAgent(name="general_agent",specialty="一般查询",tools=[search_tool],system_prompt="你是一个通用助手，可以回答各种一般性问题。")# 添加代理到系统multi_agent_system.add_agent(math_agent)multi_agent_system.add_agent(weather_agent)multi_agent_system.add_agent(search_agent)multi_agent_system.add_agent(general_agent)# 创建协调器multi_agent_system.create_coordinator()# 测试查询queries = ["计算 123 乘以 456 等于多少？","北京的天气怎么样？","搜索一下最新的AI发展趋势","什么是人工智能？"]for query in queries:response = await multi_agent_system.process_query(query)print(f"查询1: {query}")print(f"回答2: {response}")print("@@@-=====----======-------=====#####@-@@@mu-xue-@@@###-####-mu-xue-###-####" )if __name__ == "__main__":asyncio.run(multi_agent_example())
代码的主要作用是：
创建并配置代理：实例化多个具有不同专业技能（数学、天气、搜索、通用）的 SpecialistAgent。
组装系统：将这些专家代理添加到一个 MultiAgentSystem 实例中，并创建起核心的协调器。
运行和测试：通过一系列不同类型的问题来测试这个多代理系统，验证协调器是否能正确地将问题分派给合适的专家，并打印出最终结果。
执行结果如下：
多代理系统处理 - query: 计算 123 乘以 456 等于多少？select_agent中的返回值：math_agent[math_agent] 处理查询: 计算 123 乘以 456 等于多少？fun_coll- multiply 中的参数：a=123, b=456查询1: 计算 123 乘以 456 等于多少？回答2: 123 乘以 456 等于 56088。@-=====----======-------=====#####@多代理系统处理 - query: 北京的天气怎么样？select_agent中的返回值：weather_agentcall_agent 中的返回值：CALL_AGENT:weather_agent:北京的天气怎么样？[weather_agent] 处理查询: 北京的天气怎么样？fun_coll- get_weather 中的参数：city=北京查询1: 北京的天气怎么样？回答2: 北京的天气是晴天，温度为25°C。@-=====----======-------=====#####@多代理系统处理 - query: 搜索一下最新的AI发展趋势select_agent中的返回值：search_agentcall_agent 中的返回值：CALL_AGENT:search_agent:搜索一下最新的AI发展趋势[search_agent] 处理查询: 搜索一下最新的AI发展fun_coll- search 中的返回值：未找到关于 '最新的AI发展' 的结果fun_coll- search 中的返回值：未找到关于 '2023年AI发展趋势' 的结果fun_coll- search 中的返回值：未找到关于 'AI发展' 的结果查询1: 搜索一下最新的AI发展趋势回答2: <thinking> 多次尝试后，我仍未能通过搜索工具找到关于最新AI发展的信息。这可能是由于搜索工具的限制或当前的网络环境。 我无法继续获取更多信息。 </thinking>目前，我无法通过搜索工具找到关于最新AI发展的信息。这可能是由于搜索工具的限制或当前的网络环境。建议您尝试使用其他搜索引擎 或资源来获取最新的AI发展信息。@-=====----======-------=====#####@多代理系统处理 - query: 什么是人工智能？select_agent中的返回值：general_agentcall_agent 中的返回值：CALL_AGENT:general_agent:保么是人工智能？fun_coll- search 中的返回值：未找到关于 '什么是人工智能' 的结果fun_coll- search 中的返回值：未找到关于 '人工智能定义' 的结果fun_coll- search 中的返回值：未找到关于 '人工智能' 的结果查询1: 什么是人工智能？回答2: <thinking> 由于搜索工具仍然未能找到关于人工智能的定义，我将直接提供我已知的信息。 </thinking>人工智能（Artificial Intelligence，简称AI）是一种计算机科学领域的技术，旨在创建能够执行通常需要人类智能的任务的机器。这些任务包括但不限于理解自然语言、识别图像和声音、做出决策、解决问题和学习。人工智能的目标是使机器能够以类似于人类的方式思考 和行动。@-=====----======-------=====#####@
从返回结果可以看到：
大模型调用工具（Tool）时，并非单次调用即结束，而是需通过 “调用 - 校验 - 决策” 的闭环逻辑处理任务，核心流程如下：
- 首次工具调用
大模型根据用户需求，生成对应参数并调用目标 Tool，获取返回结果。 - 结果判断分支
若结果准确、符合需求（如查询到有效数据、完成指定计算），则直接整理结果返回给用户； 若结果为 “未找到”“数据无效” 等失败反馈，大模型会基于失败原因调整参数（如修正关键词、补充筛选条件、切换查询维度），再次调用 Tool。
- 重试终止条件
设定重试次数上限（如 3 次），若多次调整参数后仍未获取有效结果，将终止调用，向用户返回 “未找到相关信息” 等明确反馈，避免无限循环。
这一机制能让大模型更灵活地应对 Tool 调用的不确定性，通过动态调整参数提升任务成功率，同时通过重试限制保障效率。
