Agentic AI - 第二课 - 多Agent协同

背景简介

随着人工智能技术的快速发展，单一Agent已难以满足复杂任务的需求。多Agent协同系统通过分工合作，能够更高效地处理多样化任务。本文基于ReAct范式，深入探讨如何构建一个包含Router、Executor、Validator和Responder四个角色的多Agent协同系统，实现智能化的任务处理流程。

前置信息

完成【Agentic AI - 第一课 - 从零实现基于 ReAct 范式的智能 Agent】

详细信息

系统依赖

# requirements.txt
ollama
aiohttp
pydantic
asyncio
typing

系统架构设计

我们的多Agent协同系统采用四层架构，每个Agent各司其职：

Router（路由器）：负责用户意图识别和任务分发
Executor（执行器）：负责工具调用和参数提取
Validator（验证器）：负责结果验证和错误检测
Responder（响应器）：负责生成自然语言回复
多Agent 协同代码

import asyncio
from typing import Dict, Any, Callable
from app.ai_agent.agent_tools import tool_calculator, tool_get_weather
from app.ai_agent.agent_tools import TOOLS
from app.core.pers_ollama import OllamaClient

# ========== 工具注册表 ==========
TOOL_REGISTRY: Dict[str, Callable] = {
    "tool_get_weather": tool_get_weather,
    "tool_calculator": tool_calculator,
}

class MultiAgentBot:
    """终极版：Router -> Executor -> Validator -> Responder (修复工具调用问题)"""
    
    def __init__(self, base_url: str = "http://localhost:11434", model: str = "llama3.1:8b"):
        self.client = OllamaClient(base_url=base_url, model=model, temperature=0.2)
        self.model = model

    async def close(self):
        await self.client.close()

    # ================= Agent 1: Router =================
    ROUTER_PROMPT = (
        "意图分类（只输出模式名）：\n"
        "- CHAT: 闲聊、常识\n"
        "- WEATHER: 查天气\n"
        "- CALCULATOR: 算数\n\n"
        "示例：\n"
        "用户: 月亮为什么圆 -> CHAT\n"
        "用户: 上海冷不冷 -> WEATHER\n"
        "用户: 123*456 -> CALCULATOR\n\n"
        "用户问题: {user_message}\n模式:"
    )

    async def agent_router(self, user_message: str) -> str:
        response = await self.client.chat(messages=[{"role": "user", "content": self.ROUTER_PROMPT.format(user_message=user_message)}], tools=[])
        c = response.content.strip().upper()
        if "CALC" in c: return "CALCULATOR"
        if "WEATHER" in c: return "WEATHER"
        return "CHAT"

    # ================= Agent 2: Executor (修复版) =================
    EXECUTOR_SYSTEM_PROMPT = (
        "你是工具调用专家。你的任务是提取参数并调用指定工具。\n"
        "当前模式：{tool_type}\n\n"
        "重要规则：\n"
        "1. 必须使用系统提供的工具。\n"
        "2. 【中文城市名处理】：如果用户问的是中文城市名（如兰州、上海），"
        "必须**逐字复制**用户问题中的城市名作为参数，严禁翻译、拼音化或猜测。\n\n"
        "【示例】\n"
        "用户: 兰州的天气热吗？ -> 调用 tool_get_weather(city='兰州')\n"
        "用户: 上海冷不冷？ -> 调用 tool_get_weather(city='上海')\n"
        "用户: 123 * 456 是多少？ -> 调用 tool_calculator(expression='123 * 456')\n\n"
        "3. 如果收到错误反馈，请根据反馈修正参数。"
    )


    async def agent_executor(self, user_message: str, tool_type: str, history_msgs: list = None):
        if history_msgs is None:
            history_msgs = []
        
        system_content = self.EXECUTOR_SYSTEM_PROMPT.format(tool_type=tool_type)
        
        # 构建消息列表
        # 如果有历史消息（说明是重试），则不再追加原始 user_message，直接让模型回复最后的反馈
        if history_msgs:
            messages = [{"role": "system", "content": system_content}] + history_msgs
        else:
            messages = [{"role": "system", "content": system_content}] + [{"role": "user", "content": user_message}]
        
        # 【关键修复】：根据类型筛选可用工具，并传递给 client
        allowed_tools = []
        if tool_type == "WEATHER":
            allowed_tools = [t for t in TOOLS if t["function"]["name"] == "tool_get_weather"]
        elif tool_type == "CALCULATOR":
            allowed_tools = [t for t in TOOLS if t["function"]["name"] == "tool_calculator"]
            
        # 【关键修复】：这里必须传递 allowed_tools，不能是空列表！
        return await self.client.chat(messages=messages, tools=allowed_tools)

    # ================= Agent 3: Validator (带具体纠错原因) =================
    # 关键点：允许输出 "FAIL: 原因"，直接告诉模型改哪里
    VALIDATOR_PROMPT = (
        "你是参数审核员。请判断工具调用是否成功且符合用户意图。\n"
        "你需要查看：【用户问题】、【工具参数】、【工具执行结果】。\n\n"
        "输出要求：\n"
        "1. 成功：只输出 'PASS'。\n"
        "2. 失败：输出 'FAIL: 原因'。\n\n"
        "【示例】\n"
        "示例 1 (正确提取):\n"
        "用户: 兰州的天气\n"
        "参数: {{'city': '兰州'}}\n"
        "结果: 晴，25度\n"
        "结果: PASS\n\n"
        "示例 2 (提取错误导致未找到城市):\n"
        "用户: 兰州的天气\n"
        "参数: {{'city': '勗安'}}\n"
        "结果: 未找到城市: 勗安\n"
        "结果: FAIL: 工具未找到城市 '勗安'。用户问的是 '兰州'，请直接使用 '兰州' 作为参数。\n\n"
        "示例 3 (中文乱码/错字):\n"
        "用户: 上海天气\n"
        "参数: {{'city': '上晦'}}\n"
        "结果: 无法获取天气信息\n"
        "结果: FAIL: 参数城市名 '上晦' 错误，请修正为 '上海'。\n\n"
        "示例 4 (错误的表达式):\n"
        "用户: 计算一个半径为3的圆的面积\n"
        "参数: {{'expression': 'π*3^2'}}\n"
        "结果: 表达式无效或包含非法操作:\n"
        "结果: FAIL: 表达式非法，转化为纯数字的公式，不使用特殊字符，例如'3.14159 * 3 * 3'。\n\n"
        "【当前任务】\n"
        "用户: {user_message}\n"
        "参数: {tool_args}\n"
        "工具结果: {tool_result}\n"
        "结果:"
    )


    async def agent_validator(
        self, 
        user_message: str, 
        tool_name: str, 
        tool_args: dict, 
        tool_type: str, 
        tool_result_str: str  # 【新增】接收工具执行后的结果字符串
    )  -> tuple[bool, str]:  
        
        prompt = self.VALIDATOR_PROMPT.format(
            user_message=user_message,
            tool_args=tool_args,
            tool_result=tool_result_str  # 【新增】注入结果
        )
        response = await self.client.chat(messages=[{"role": "user", "content": prompt}], tools=[])
        content = response.content.strip()
        
        if "PASS" in content.upper():
            return True, "PASS"
        else:
            return False, content


    # ================= Agent 4: Responder =================
    RESPONDER_PROMPT = (
        "你是一个友好的助手。请根据工具结果回答用户的问题。\n"
        "用户问题：{user_message}\n"
        "工具返回结果：{tool_result}\n\n"
        "请用自然、口语化的中文回答。"
    )

    async def agent_responder(self, user_message: str, tool_result: str) -> str:
        prompt = self.RESPONDER_PROMPT.format(
            user_message=user_message,
            tool_result=tool_result
        )
        response = await self.client.chat(messages=[{"role": "user", "content": prompt}], tools=[])
        return response.content

    # ================= 主流程 =================
    async def chat(self, user_message: str) -> str:
        print(f"\n👤 用户: {user_message}")
        print("-" * 60)

        # 1. Router
        print("🧠 [Router] 分析意图...")
        mode = await self.agent_router(user_message)
        print(f"🧠 [Router] 模式: {mode}")

        if mode == "CHAT":
            return await self.agent_responder(user_message, "这是一个常识问题，不需要工具。")

        # 2. Executor -> Validator Loop
        conversation_history = [] 
        
        for attempt in range(5):
            print(f"🔧 [Executor] 尝试 {attempt + 1}...")
            
            # 【关键修复】：调用 agent_executor，而不是直接调 client.chat
            # agent_executor 会处理历史消息和工具传递
            exec_response = await self.agent_executor(user_message, mode, conversation_history)
            
            if not exec_response.tool_calls:
                print(f"❌ [Executor] 没有生成工具调用。内容: {exec_response.content}")
                # 如果没有生成工具，构造一个错误反馈迫使重试（或者直接失败）
                if attempt == 0:
                    conversation_history.append({
                        "role": "user", 
                        "content": "你没有调用工具！请立即调用工具，不要只用文字描述。"
                    })
                    continue
                else:
                    return "Executor 执行失败，模型拒绝调用工具。"

            # 记录历史
            conversation_history.append({
                "role": "assistant",
                "content": exec_response.content,
                "tool_calls": exec_response.tool_calls
            })

            tool_call = exec_response.tool_calls[0]
            tool_name = tool_call["function"]["name"]
            tool_args = tool_call["function"]["arguments"]
            
            print(f"🔧 参数: {tool_args}")
            
            # 真实执行工具
            try:
                tool_result = TOOL_REGISTRY[tool_name](**tool_args)
                # 转成字符串方便 Validator 读取
                tool_result_str = str(tool_result)
                print(f"✅ [Tool Result] {tool_result_str[:50]}...")
            except Exception as e:
                tool_result = f"工具执行报错: {str(e)}"
                tool_result_str = tool_result

            # 3. Validator (传入 tool_result_str)
            print("🔍 [Validator] 检查中...")
            
            # 【修改】这里增加了 tool_result_str 参数
            is_valid, feedback = await self.agent_validator(
                user_message, 
                tool_name, 
                tool_args, 
                mode, 
                tool_result_str
            )
            
            if is_valid:
                print("✅ [Validator] 通过")
                return await self.agent_responder(user_message, str(tool_result))
            else:
                print(f"❌ [Validator] 拒绝: {feedback}")
                # 如果是因为“未找到城市”失败，反馈会非常明确告诉 Executor 改成 '兰州'
                conversation_history.append({
                    "role": "user",
                    "content": f"上一次调用失败。原因：{feedback}。请修正参数并再次调用工具。"
                })

        return "抱歉，尝试多次后仍未通过验证。"



# ================= 测试入口 =================

async def main():
    bot = MultiAgentBot()
    
    test_questions = [
        #"上海现在冷不冷？",        # 期望：修正参数后回答
        #"123 * 456 是多少？",     # 期望：计算后回答
        #"今天北京天气怎么样？",     # 期望：直接回答
        #"月亮为什么是圆的",
        #"兰州的天气热吗？",
        # "随便告诉我一个中国城市的天气，不要是北京，上海，兰州",
        "计算一个半径为3的圆的面积",
        "半径为5cm的圆的周长是多少？",
        "计算一个半径为12，高为2cm的圆锥的体积"
    ]

    print("🚀 四 Agent 协同 (Router -> Executor -> Validator -> Responder)")
    print("=" * 60)

    try:
        for q in test_questions:
            answer = await bot.chat(q)
            print(f"🤖 最终回答: {answer}")
            print("=" * 60)
    finally:
        await bot.close()

if __name__ == "__main__":
    asyncio.run(main())

系统特色

智能错误恢复机制

系统实现了智能的错误恢复机制，当工具调用失败时，Validator会提供具体的错误原因，Executor根据反馈自动修正参数：

# 示例：城市名错误修正
用户: 兰州的天气热吗？
Executor 第一次尝试: tool_get_weather(city='勗安')  # 错误参数
Validator 反馈: FAIL: 工具未找到城市 '勗安'。用户问的是 '兰州'，请直接使用 '兰州' 作为参数。
Executor 第二次尝试: tool_get_weather(city='兰州')  # 正确参数

中文参数处理优化

针对中文城市名处理，系统特别强调"逐字复制"原则，避免了翻译、拼音化等常见错误：

EXECUTOR_SYSTEM_PROMPT = (
    "【中文城市名处理】：如果用户问的是中文城市名（如兰州、上海），"
    "必须**逐字复制**用户问题中的城市名作为参数，严禁翻译、拼音化或猜测。"
)

数学表达式规范化

对于计算器工具，系统要求将包含特殊字符的数学表达式转换为纯数字格式：

# 错误示例：'π * 12^2 * 2 / 3'
# 正确示例：'3.14159 * 12 * 12 * 2 / 3'

调用结果示例

🚀 四 Agent 协同 (Router -> Executor -> Validator -> Responder)
============================================================

👤 用户: 上海现在冷不冷？
------------------------------------------------------------
🧠 [Router] 分析意图...
🧠 [Router] 模式: WEATHER
🔧 [Executor] 尝试 1...
🔧 参数: {'city': '北京'}
✅ [Tool Result] 城市: 北京
温度: 5.8°C
体感温度: 2.8°C
湿度: 67%
风速: 6.2 km/h
...
🔍 [Validator] 检查中...
❌ [Validator] 拒绝: 根据用户问题和工具执行结果，可以判断工具调用是否成功且符合用户意图。

用户问题：上海现在冷不冷？
参数：{'city': '北京'}
工具结果：城市: 北京
温度: 5.8°C
体感温度: 2.8°C
湿度: 67%
风速: 6.2 km/h
天气: 小雨

由于用户问题是关于上海的，而工具参数中设置的是北京，且工具执行结果显示的是北京的天气信息，因此可以判断工具调用失败。

输出：FAIL: 工具未找到城市 '上海'。用户问的是 '上海'，请直接使用 '上海' 作为参数。
🔧 [Executor] 尝试 2...
🔧 参数: {'city': '上海'}
✅ [Tool Result] 城市: 上海
温度: 2.0°C
体感温度: -2.9°C
湿度: 54%
风速: 13.3 km/...
🔍 [Validator] 检查中...
✅ [Validator] 通过
🤖 最终回答: 上海现在还蛮冷的，外面只有2摄氏度呢！体感温度甚至是负数，感觉比实际温度还要冷。湿度也不是很高，风速也比较大，所以还是得穿暖和点。天气看起来是晴朗的，应该是干净利落的好天气。
============================================================

👤 用户: 123 * 456 是多少？
------------------------------------------------------------
🧠 [Router] 分析意图...
🧠 [Router] 模式: CALCULATOR
🔧 [Executor] 尝试 1...
🔧 参数: {'expression': '123 * 456'}
✅ [Tool Result] 56088...
🔍 [Validator] 检查中...
✅ [Validator] 通过
🤖 最终回答: 答案是：123乘以456等于56,088。
============================================================

👤 用户: 今天北京天气怎么样？
------------------------------------------------------------
🧠 [Router] 分析意图...
🧠 [Router] 模式: WEATHER
🔧 [Executor] 尝试 1...
🔧 参数: {'city': '北京'}
✅ [Tool Result] 城市: 北京
温度: 5.8°C
体感温度: 2.8°C
湿度: 67%
风速: 6.2 km/h
...
🔍 [Validator] 检查中...
✅ [Validator] 通过
🤖 最终回答: 今天北京的天气不太好，温度只有5.8摄氏度，感觉还要冷点，湿度也蛮高的67%，而且有小雨。风速也不是很大，只有6.2公里每小时。
============================================================

👤 用户: 月亮为什么是圆的
------------------------------------------------------------
🧠 [Router] 分析意图...
🧠 [Router] 模式: CHAT
🤖 最终回答: 这个问题其实很简单，月亮看起来像个圆圈，是因为它是球形的嘛！地球和其他行星都是这样子的，因为它们是在自转和绕太阳公转的过程中保持着平衡和稳定状态，所以才会呈现出圆形的外观。
============================================================

👤 用户: 兰州的天气热吗？
------------------------------------------------------------
🧠 [Router] 分析意图...
🧠 [Router] 模式: WEATHER
🔧 [Executor] 尝试 1...
🔧 参数: {'city': '勗安'}
未找到城市: 勗安
✅ [Tool Result] 无法获取天气信息...
🔍 [Validator] 检查中...
❌ [Validator] 拒绝: FAIL: 工具未找到城市 '勗安'。用户问的是 '兰州'，请直接使用 '兰州' 作为参数。
🔧 [Executor] 尝试 2...
🔧 参数: {'city': '兰州'}
✅ [Tool Result] 城市: 兰州
温度: 4.7°C
体感温度: -0.2°C
湿度: 65%
风速: 18.0 km/...
🔍 [Validator] 检查中...
✅ [Validator] 通过
🤖 最终回答: 兰州的天气不算热，目前温度还比较低，接近4摄氏度，体感温度甚至有负数，湿度也不是很高，风速也相对较大。整体来说，兰州的天气还是比较凉爽的。
============================================================

👤 用户: 随便告诉我一个中国城市的天气，不要是北京，上海，兰州
------------------------------------------------------------
🧠 [Router] 分析意图...
🧠 [Router] 模式: WEATHER
🔧 [Executor] 尝试 1...
🔧 参数: {'city': '北京'}
✅ [Tool Result] 城市: 北京
温度: 5.8°C
体感温度: 2.8°C
湿度: 67%
风速: 6.2 km/h
...
🔍 [Validator] 检查中...
❌ [Validator] 拒绝: FAIL: 工具参数错误，用户要求的城市不是 '北京'。
🔧 [Executor] 尝试 2...
🔧 参数: {'city': '北京'}
✅ [Tool Result] 城市: 北京
温度: 5.8°C
体感温度: 2.8°C
湿度: 67%
风速: 6.2 km/h
...
🔍 [Validator] 检查中...
❌ [Validator] 拒绝: FAIL: 工具参数错误，用户要求的城市不是 '北京'，而是其他城市。
🔧 [Executor] 尝试 3...
🔧 参数: {'city': '上海'}
✅ [Tool Result] 城市: 上海
温度: 2.0°C
体感温度: -2.9°C
湿度: 54%
风速: 13.3 km/...
🔍 [Validator] 检查中...
❌ [Validator] 拒绝: 根据用户的要求和工具的执行结果，我发现以下问题：

1. 用户要求随便告诉一个中国城市的天气，不要是北京，上海，兰州，但是参数中传入了 'city': '上海'，这与用户要求相矛盾。
2. 工具执行结果显示的是上海的天气信息，而不是其他城市。

因此，我判断工具调用失败。
🔧 [Executor] 尝试 4...
🔧 参数: {'city': '广州'}
✅ [Tool Result] 城市: 广州
温度: 11.4°C
体感温度: 7.0°C
湿度: 56%
风速: 19.4 km/...
🔍 [Validator] 检查中...
✅ [Validator] 通过
🤖 最终回答: 你想知道广州的天气啊！现在广州的温度是11.4摄氏度，体感温度是7.0摄氏度，湿度是56%，风速是19.4公里每小时，天气大部分是晴朗的。
============================================================

👤 用户: 计算一个半径为12，高为2cm的圆锥的体积
------------------------------------------------------------
🧠 [Router] 分析意图...
🧠 [Router] 模式: CALCULATOR
🔧 [Executor] 尝试 1...
🔧 参数: {'expression': 'π * 12^2 * 2 / 3'}
✅ [Tool Result] 表达式无效或包含非法操作: 非法的语法结构: Name...
🔍 [Validator] 检查中...
❌ [Validator] 拒绝: 根据示例要求，我将分析用户问题、工具参数和工具执行结果来判断工具调用是否成功且符合用户意图。

**用户问题：** 计算一个半径为12，高为2cm的圆锥的体积

**工具参数：** {'expression': 'π * 12^2 * 2 / 3'}

**工具执行结果：** 表达式无效或包含非法操作: 非法的语法结构: Name

根据用户问题和工具参数，表达式应该是正确的，但实际上出现了错误。让我们分析一下：

* 用户的问题要求计算圆锥的体积。
* 工具参数中的表达式似乎是正确的，但执行结果显示有非法的语法结构。

原因可能是由于特殊字符或符号的使用，例如 `^` 和 `/`，导致工具无法正确解析表达式。正确的表达式应该是纯数字的公式，不使用特殊字符。

因此，我将输出：

FAIL: 表达式非法，转化为纯数字的公式，不使用特殊字符，例如'3.14159 * 12 * 12 * 2 / 3'。
🔧 [Executor] 尝试 2...
🔧 参数: {'expression': '3.14159 * 12 * 12 * 2 / 3'}
✅ [Tool Result] 301.59263999999996...
🔍 [Validator] 检查中...
✅ [Validator] 通过
🤖 最终回答: 这个圆锥的体积是301.59立方厘米。
============================================================

以上便是本文的全部内容，感谢您的阅读，如遇到任何问题，欢迎在评论区留言讨论。

菜单

分享

Agentic AI - 第二课 - 多Agent协同

背景简介

前置信息

详细信息

系统依赖

系统架构设计

系统特色

智能错误恢复机制

中文参数处理优化

数学表达式规范化

调用结果示例

评论

Agentic AI - 第七课 - 优化 Responder Agent

Agentic AI - 第六课 - 优化 Validator Agent

Agentic AI - 第五课 - 优化 Executor Agent

Agentic AI - 第四课 - 优化 Router Agent 实现高效意图识别

Agentic AI - 第三课 - 多Agent协同模块化架构实战

Agentic AI - 第二课 - 多Agent协同

Agentic AI - 第一课 - 从零实现基于 ReAct 范式的智能 Agent

Agentic AI - 附录一 - 模型选择

Python - Sample - pers_logging.py

Python - Sample - config.py