LlamaIndex 开发多智能体 Agents 入门基础
我们将逐步拆解 LlamaIndex 多智能体的开发流程,帮助大家系统掌握从需求分析到落地实现的完整方法。
本章将聚焦多智能体开发。
当场景复杂、涉及多模块功能时,单智能体难以满足需求,此时需构建多智能体系统:为特定领域任务创建专门智能体,让每个智能体各司其职,共同解决复杂问题。后续将逐步拆解其开发要点。
接下来,我们将逐步拆解 LlamaIndex 多智能体的开发流程,帮助大家系统掌握从需求分析到落地实现的完整方法。

开发步骤如下:
1、安装与环境
导包所需的包。
pip install -U llama-index llama-index-llms-openai
2、选择 LLM 与全局设置
muxue_model_loader.py文件代码:
from llama_index.llms.openai import OpenAI
from llama_index.core import Settings
llm = 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 * 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,
# llm=model_loaders.llm, # 使用model_loaders中的llm
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)
代码分析:
- 初始化方法中,设置智能体的名称,FunctionAgent接收多个工具tools,以及非常重要的提示词;verbose=True开启详细日志打印,若不需要则设置为False,或不写。
- 执行Agent的方法,handle_query做了一个简单的封装,便于统一管理和执行agent。

5、多智能体管理
创建一个多智能体的管理类,创建一个“协调器”代理,这个协调器本身不直接回答问题,而是像一个项目经理,负责将用户的请求(query)分派给最合适的“专家”代理来处理。
from llama_index.core.tools import FunctionTool
from llama_index.core.agent.workflow import FunctionAgent
from 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:
# 注意:这里需要异步执行,但在同步函数中无法直接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 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],
# llm=model_loaders.llm, # 使用model_loaders中的llm
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、调用多智能体
主方法组装多个智能体,并且调用这些智能体。
# ================== 初始化 大模型 Langfuse ==================
import common.muxue_model_loader
import common.langfuse_init_client
# ================== 初始化大模型 Langfuse end ==================
import asyncio
from fun_coll import *
from specialist_agent import SpecialistAgent
from 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_agent
call_agent 中的返回值:CALL_AGENT:weather_agent:北京的天气怎么样?
[weather_agent] 处理查询: 北京的天气怎么样?
fun_coll- get_weather 中的参数:city=北京
查询1: 北京的天气怎么样?
回答2: 北京的天气是晴天,温度为25°C。
@-=====----======-------=====#####@
多代理系统处理 - query: 搜索一下最新的AI发展趋势
select_agent中的返回值:search_agent
call_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_agent
call_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 调用的不确定性,通过动态调整参数提升任务成功率,同时通过重试限制保障效率。

想入门 AI 大模型却找不到清晰方向?备考大厂 AI 岗还在四处搜集零散资料?别再浪费时间啦!2025 年 AI 大模型全套学习资料已整理完毕,从学习路线到面试真题,从工具教程到行业报告,一站式覆盖你的所有需求,现在全部免费分享!
👇👇扫码免费领取全部内容👇👇

一、学习必备:100+本大模型电子书+26 份行业报告 + 600+ 套技术PPT,帮你看透 AI 趋势
想了解大模型的行业动态、商业落地案例?大模型电子书?这份资料帮你站在 “行业高度” 学 AI:
1. 100+本大模型方向电子书

2. 26 份行业研究报告:覆盖多领域实践与趋势
报告包含阿里、DeepSeek 等权威机构发布的核心内容,涵盖:
- 职业趋势:《AI + 职业趋势报告》《中国 AI 人才粮仓模型解析》;
- 商业落地:《生成式 AI 商业落地白皮书》《AI Agent 应用落地技术白皮书》;
- 领域细分:《AGI 在金融领域的应用报告》《AI GC 实践案例集》;
- 行业监测:《2024 年中国大模型季度监测报告》《2025 年中国技术市场发展趋势》。
3. 600+套技术大会 PPT:听行业大咖讲实战
PPT 整理自 2024-2025 年热门技术大会,包含百度、腾讯、字节等企业的一线实践:

- 安全方向:《端侧大模型的安全建设》《大模型驱动安全升级(腾讯代码安全实践)》;
- 产品与创新:《大模型产品如何创新与创收》《AI 时代的新范式:构建 AI 产品》;
- 多模态与 Agent:《Step-Video 开源模型(视频生成进展)》《Agentic RAG 的现在与未来》;
- 工程落地:《从原型到生产:AgentOps 加速字节 AI 应用落地》《智能代码助手 CodeFuse 的架构设计》。
二、求职必看:大厂 AI 岗面试 “弹药库”,300 + 真题 + 107 道面经直接抱走
想冲字节、腾讯、阿里、蔚来等大厂 AI 岗?这份面试资料帮你提前 “押题”,拒绝临场慌!

1. 107 道大厂面经:覆盖 Prompt、RAG、大模型应用工程师等热门岗位
面经整理自 2021-2025 年真实面试场景,包含 TPlink、字节、腾讯、蔚来、虾皮、中兴、科大讯飞、京东等企业的高频考题,每道题都附带思路解析:

2. 102 道 AI 大模型真题:直击大模型核心考点
针对大模型专属考题,从概念到实践全面覆盖,帮你理清底层逻辑:

3. 97 道 LLMs 真题:聚焦大型语言模型高频问题
专门拆解 LLMs 的核心痛点与解决方案,比如让很多人头疼的 “复读机问题”:

三、路线必明: AI 大模型学习路线图,1 张图理清核心内容
刚接触 AI 大模型,不知道该从哪学起?这份「AI大模型 学习路线图」直接帮你划重点,不用再盲目摸索!

路线图涵盖 5 大核心板块,从基础到进阶层层递进:一步步带你从入门到进阶,从理论到实战。

L1阶段:启航篇丨极速破界AI新时代
L1阶段:了解大模型的基础知识,以及大模型在各个行业的应用和分析,学习理解大模型的核心原理、关键技术以及大模型应用场景。

L2阶段:攻坚篇丨RAG开发实战工坊
L2阶段:AI大模型RAG应用开发工程,主要学习RAG检索增强生成:包括Naive RAG、Advanced-RAG以及RAG性能评估,还有GraphRAG在内的多个RAG热门项目的分析。

L3阶段:跃迁篇丨Agent智能体架构设计
L3阶段:大模型Agent应用架构进阶实现,主要学习LangChain、 LIamaIndex框架,也会学习到AutoGPT、 MetaGPT等多Agent系统,打造Agent智能体。

L4阶段:精进篇丨模型微调与私有化部署
L4阶段:大模型的微调和私有化部署,更加深入的探讨Transformer架构,学习大模型的微调技术,利用DeepSpeed、Lamam Factory等工具快速进行模型微调,并通过Ollama、vLLM等推理部署框架,实现模型的快速部署。

L5阶段:专题集丨特训篇 【录播课】

四、资料领取:全套内容免费抱走,学 AI 不用再找第二份
不管你是 0 基础想入门 AI 大模型,还是有基础想冲刺大厂、了解行业趋势,这份资料都能满足你!
现在只需按照提示操作,就能免费领取:
👇👇扫码免费领取全部内容👇👇

2025 年想抓住 AI 大模型的风口?别犹豫,这份免费资料就是你的 “起跑线”!
更多推荐


所有评论(0)