当场景复杂、涉及多模块功能时，单智能体难以满足需求，此时需构建多智能体系统：为特定领域任务创建专门智能体，让每个智能体各司其职，共同解决复杂问题。后续将逐步拆解其开发要点。
接下来，我们将逐步拆解 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 Settings
llm = OpenAI(model="gpt-4o-mini") Settings.llm = llm
3、定义工具（Tools）
在 LlamaIndex 里，“工具”是智能体的手脚；最常用包装是 FunctionTool。LlamaIndex 将任意 Python 函数包装成工具，供 Agent 决策调用；这是 FunctionAgent 模式的基础。
fun_coll.py代码如下：
from llama_index.core.tools import FunctionTooldef 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)
4、专门的Agent父类
在多智能体架构中，由于存在多个功能各异的 Agent，直接管理和调用会面临逻辑分散、重复编码等问题。因此，定义一个统一的父类 Agent是关键一步：它能为所有子 Agent 提供标准化的基础框架（如通用初始化方法、核心任务接口等），既减少重复开发，也让后续对多个 Agent 的调用、监控和协同管理更高效统一。from llama_index.core.agent.workflow import FunctionAgent
class SpecialistAgent: """专业Agent基类"""
def __init__(self, name: str, specialty: str, tools, system_prompt: str): self.name = name self.specialty = specialty self.agent = FunctionAgent( tools=tools, system_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)
代码分析：
5、多智能体管理
创建一个多智能体的管理类，创建一个“协调器”代理，这个协调器本身不直接回答问题，而是像一个项目经理，负责将用户的请求（query）分派给最合适的“专家”代理来处理。
from llama_index.core.tools import FunctionToolfrom llama_index.core.agent.workflow import FunctionAgentfrom specialist_agent import SpecialistAgent
class MultiAgentSystem: """多代理系统"""
def __init__(self): self.agents = {} self.coordinator = None
def add_agent(self, agent: SpecialistAgent): """添加代理""" self.agents[agent.name] = agent
def create_coordinator(self): """创建协调器代理""" def call_agent(agent_name: str, query: str) -> str: """调用指定的代理处理查询""" if agent_name in self.agents: 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 ret
agent_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], system_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_response else: return f"错误：找不到名为 '{agent_name}' 的代理"
return response_str
这个 MultiAgentSystem 类构建了一个框架，其中包含：
多个专家代理 (Specialist Agents)：系统中可以添加多个专门负责特定领域（如数学、天气、搜索）的代理。这些代理是从外部 specialist_agent 模块导入的，它们各自拥有处理特定任务的能力。
一个协调器代理 (Coordinator Agent)：这是一个特殊的代理，它的大脑（LLM）被赋予了两个工具：一个用来选择合适的专家，另一个用来调用这个专家；注意select_agent方法是一个非常简单的基于规则的路由逻辑。例如，如果查询中包含“计算”，它就返回 "math_agent"。在实际应用中，这里可以用更复杂的逻辑，甚至可以用另一个 LLM 来做更智能的选择。
一个处理流程 (Processing Workflow)：当用户查询进来时，首先由协调器进行分析和分派，然后系统再执行协调器的指令，调用相应的专家代理，并返回最终结果。
6、调用多智能体
主方法组装多个智能体，并且调用这些智能体。
import common.muxue_model_loaderimport common.langfuse_init_clientimport 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。
粤ICP备14082021号
免费POC，
零成本试错
