上一章节已经介绍了LlamaIndex开发智能体的方法，是开发单智能体，本章讲一下 开发多智能体Agents。 开发多智能体涉及到很多方面，我们一步一步学习。

在一些复杂的场景，涉及到众多模块和功能，单智能体无法满足，则需要使用多智能体来解决需求。创建多个专门的智能体，每个代理负责特定领域的任务。

开发步骤如下：

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，便于调用和管理。

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 类构建了一个框架，其中包含：

1. 多个专家代理 (Specialist Agents)：系统中可以添加多个专门负责特定领域（如数学、天气、搜索）的代理。这些代理是从外部 specialist_agent 模块导入的，它们各自拥有处理特定任务的能力。

2. 一个协调器代理 (Coordinator Agent)：这是一个特殊的代理，它的大脑（LLM）被赋予了两个工具：一个用来选择合适的专家，另一个用来调用这个专家；注意select_agent方法是一个非常简单的基于规则的路由逻辑。例如，如果查询中包含“计算”，它就返回 "math_agent"。在实际应用中，这里可以用更复杂的逻辑，甚至可以用另一个 LLM 来做更智能的选择。

3. 一个处理流程 (Processing Workflow)：当用户查询进来时，首先由协调器进行分析和分派，然后系统再执行协调器的指令，调用相应的专家代理，并返回最终结果。

5、调用多智能体

主方法组装多个智能体，并且调用这些智能体。

# ================== 初始化 大模型 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())

 代码的主要作用是：

1. 创建并配置代理：实例化多个具有不同专业技能（数学、天气、搜索、通用）的 SpecialistAgent。

2. 组装系统：将这些专家代理添加到一个 MultiAgentSystem 实例中，并创建起核心的协调器。

3. 运行和测试：通过一系列不同类型的问题来测试这个多代理系统，验证协调器是否能正确地将问题分派给合适的专家，并打印出最终结果。

执行结果如下：

多代理系统处理 - 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，并且参数也会变化；等多次调用后依旧找不到正确的结果，则会返回未找到之类的结果。

下一章节将智能体的介绍记忆和上下文。