用Python构建你的第一个智能体协作网络：从零到一的A2A协议实战指南

想象一下，你正在开发一个复杂的应用，需要同时处理图像识别、文本分析和数据计算。传统做法是写一个庞大的单体程序，但维护和扩展都成了噩梦。有没有一种方法，能让不同的专业模块像团队一样自主协作，各自负责擅长的部分，共同完成任务？这就是智能体（Agent）协作的魅力所在。

近年来，智能体架构已成为AI工程领域的热点。而A2A（Agent-to-Agent）协议，正是为智能体之间高效、标准化通信而设计的“通用语言”。它不关心智能体内部如何实现，只定义它们如何“对话”、如何“分工”、如何“汇报进度”。对于希望构建模块化、可扩展AI系统的开发者而言，掌握A2A意味着能够将复杂的AI能力拆解为一个个可独立部署、自由组合的智能体单元，从而大幅提升系统的灵活性与健壮性。

本文将从零开始，手把手带你使用Python和python_a2a库，构建一个由多个数学计算智能体组成的协作网络。我们将深入探讨智能体的定义、注册、任务路由与状态管理，并提供可直接运行、修改的完整代码。无论你是想了解智能体协作的前沿概念，还是寻求一个可落地的技术方案，这里都有你需要的答案。

1. 理解A2A：智能体协作的“团队宪章”

在动手写代码之前，我们需要先厘清几个核心概念。A2A协议并非凭空出现，它是对分布式系统中服务间通信模式在AI领域的一次精妙适配。

1.1 A2A协议解决了什么问题？

你可以把A2A协议想象成一个高度专业化的团队协作规范。在这个团队里：

• 每个成员（智能体）都是专家：他们只专注于自己最擅长的领域，比如有的精通三角函数计算，有的擅长文本摘要，有的专攻图像分类。
• 需要一个“任务协调员”：当接到一个复杂任务时，协调员需要知道团队里有哪些专家，以及如何将任务分解并派发给最合适的人。
• 成员之间需要一种“通用语言”：专家们必须用同一种方式接收任务、汇报进度和提交结果，否则沟通就会乱套。

A2A协议正是为这个“AI专家团队”制定的协作宪章。它主要定义了以下几件事：

1. 智能体发现与注册：新专家（智能体）如何加入团队并宣告自己的技能？
2. 任务委派与路由：协调员如何将一个复杂任务拆解，并找到能处理每个子任务的最佳专家？
3. 状态管理与进度跟踪：任务派发出去后，协调员如何知道各个子任务的执行状态（进行中、成功、失败）？
4. 数据交换格式：专家们之间传递任务描述和结果时，应该使用什么样的数据格式（通常是结构化的JSON）？

通过这套协议，我们可以构建一个松耦合、高内聚的智能体网络。每个智能体可以独立开发、测试和部署，系统的整体能力通过智能体的组合与协作来体现。

1.2 A2A与其他协议有何不同？

你可能会联想到另一个协议：MCP（Model Context Protocol）。它们之间有何区别与联系？

简单来说，MCP关注的是单个智能体“如何与外部世界交互”。它定义了智能体调用工具（如搜索引擎、数据库、API）的标准方式，可以看作是智能体内部的“工具箱管理协议”。

而A2A关注的是智能体之间“如何相互协作”。它定义了智能体相互发现、对话和分工的规则，更像是智能体间的“团队沟通协议”。

两者的关系是互补的。一个A2A智能体在接收到任务后，完全可以利用MCP协议去调用外部工具来完成自己的那部分工作，然后再通过A2A协议将结果返回给任务发起者或其他智能体。因此，MCP是智能体对内的工具标准，A2A是智能体对外的协作标准。

1.3 A2A的核心技术栈特征

一个典型的A2A实现通常会包含以下技术特征：

技术组件	作用描述	常见实现
通信层	处理智能体间的网络通信	基于HTTP/HTTPS的JSON-RPC或gRPC
序列化	将任务、状态等对象转换为可传输的格式	JSON, Protocol Buffers (Protobuf)
服务发现	让智能体能找到彼此	集中式注册中心、分布式哈希表(DHT)
任务路由	根据任务内容选择最合适的智能体	基于规则的匹配、基于向量相似度的检索、LLM驱动的智能路由
状态管理	跟踪和管理分布式任务的执行状态	状态机、事件溯源、专门的State Manager组件

提示：python_a2a库是Google A2A协议的一个Python实现参考，它封装了底层的通信和序列化细节，让我们可以更专注于智能体本身的业务逻辑。

理解了这些基础，我们就可以开始搭建环境，亲手创建第一个智能体了。

2. 环境搭建与第一个智能体

我们将构建一个简单的“数学专家团队”，包含三个智能体：正弦（Sine）、余弦（Cosine）和正切（Tangent）计算专家。

2.1 项目初始化与依赖安装

首先，确保你的开发环境已经就绪。我们需要Python 3.8或更高版本，以及核心的python_a2a库。

打开终端，创建一个新的项目目录并初始化虚拟环境（推荐）：

# 创建项目目录
mkdir a2a_math_agents
cd a2a_math_agents

# 创建并激活虚拟环境 (以 macOS/Linux 为例)
python3 -m venv venv
source venv/bin/activate

# 在 Windows 上激活命令为：venv\Scripts\activate

接下来，安装python_a2a库。由于它可能还处于早期阶段，我们使用pip从官方源安装：

pip install python_a2a

如果安装成功，你应该能在Python中导入它而不报错。可以快速验证一下：

python -c "import python_a2a; print(python_a2a.__version__)"

2.2 创建正弦计算智能体（Sine Agent）

我们的第一个智能体将负责计算给定数字的正弦值。在项目根目录下创建一个名为sine_agent.py的文件。

# sine_agent.py
import math
import re
from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState

@agent(
    name="SineCalculator",
    description="专门计算给定角度（弧度制）正弦值的智能体",
    version="1.0.0"
)
class SineAgent(A2AServer):
    """
    正弦计算专家。
    接收包含数字的文本请求，返回该数字的正弦值。
    """

    @skill(
        name="compute_sine",
        description="计算一个浮点数的正弦值",
        tags=["math", "trigonometry", "sine", "sin"]
    )
    def get_sine(self, number: float) -> str:
        """核心计算方法"""
        result = math.sin(number)
        return f"sin({number}) = {result:.6f}"  # 格式化输出，保留6位小数

    def handle_task(self, task):
        """
        处理传入任务的核心方法。
        这是A2A智能体必须实现的方法。
        """
        # 1. 从任务消息中提取文本内容
        try:
            input_text = task.message["content"]["text"]
        except KeyError:
            task.status = TaskStatus(
                state=TaskState.FAILED,
                error_message="任务格式错误：未找到文本内容。"
            )
            return task

        # 2. 使用正则表达式从文本中提取数字
        # 匹配整数、小数、正负数
        match = re.search(r"([-+]?\d*\.?\d+)", input_text)
        if not match:
            task.status = TaskStatus(
                state=TaskState.FAILED,
                error_message="请求中未找到有效的数字。请提供如‘计算0.5的正弦’这样的指令。"
            )
            return task

        # 3. 转换并计算
        try:
            number = float(match.group(1))
            print(f"[SineAgent] 收到计算请求，数字: {number}")
            # 调用技能方法进行计算
            response_text = self.get_sine(number)

            # 4. 将结果封装到任务产出物中
            task.artifacts = [{
                "parts": [{
                    "type": "text",
                    "text": response_text
                }]
            }]
            # 5. 标记任务为完成
            task.status = TaskStatus(state=TaskState.COMPLETED)

        except ValueError as e:
            task.status = TaskStatus(
                state=TaskState.FAILED,
                error_message=f"数字转换失败: {str(e)}"
            )
        except Exception as e:
            task.status = TaskStatus(
                state=TaskState.FAILED,
                error_message=f"处理过程中发生未知错误: {str(e)}"
            )

        return task

if __name__ == "__main__":
    # 实例化并运行智能体服务器
    agent_instance = SineAgent()
    print("正弦计算智能体启动，监听端口 8001...")
    run_server(agent_instance, port=8001)

代码解读与关键点：

1. @agent装饰器：这是声明一个类为A2A智能体的关键。你需要提供name（智能体名称）、description（描述）和version（版本）。这些信息会构成该智能体的“名片”（Agent Card），供其他组件发现和了解它。
2. @skill装饰器：用于标记智能体对外提供的具体能力。name是技能的唯一标识，description是详细说明，tags是关键词列表，极大地帮助了后续的任务路由。
3. handle_task方法：这是智能体的“心脏”。所有发给这个智能体的任务都会触发此方法。它接收一个task对象，其中包含了请求的所有信息。我们的逻辑是：
   • 提取任务文本。
   • 用正则表达式找出数字。
   • 调用get_sine方法进行计算。
   • 将结果放入task.artifacts。
   • 更新task.status为完成或失败。
4. run_server函数：启动一个HTTP服务器，让智能体开始监听指定端口（这里是8001），等待任务请求。

运行智能体： 打开一个终端窗口，激活虚拟环境，导航到项目目录，执行：

python sine_agent.py

你会看到输出：正弦计算智能体启动，监听端口 8001...。这个窗口需要保持运行，不要关闭。

2.3 创建余弦与正切智能体

接下来，我们用几乎相同的模式创建另外两个智能体。创建cosine_agent.py：

# cosine_agent.py
import math
import re
from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState

@agent(
    name="CosineCalculator",
    description="专门计算给定角度（弧度制）余弦值的智能体",
    version="1.0.0"
)
class CosineAgent(A2AServer):

    @skill(
        name="compute_cosine",
        description="计算一个浮点数的余弦值",
        tags=["math", "trigonometry", "cosine", "cos"]
    )
    def get_cosine(self, number: float) -> str:
        result = math.cos(number)
        return f"cos({number}) = {result:.6f}"

    def handle_task(self, task):
        # 处理逻辑与sine_agent类似
        try:
            input_text = task.message["content"]["text"]
        except KeyError:
            task.status = TaskStatus(state=TaskState.FAILED, error_message="任务格式错误")
            return task

        match = re.search(r"([-+]?\d*\.?\d+)", input_text)
        if not match:
            task.status = TaskStatus(state=TaskState.FAILED, error_message="未找到有效数字")
            return task

        try:
            number = float(match.group(1))
            print(f"[CosineAgent] 收到计算请求，数字: {number}")
            response_text = self.get_cosine(number)
            task.artifacts = [{"parts": [{"type": "text", "text": response_text}]}]
            task.status = TaskStatus(state=TaskState.COMPLETED)
        except Exception as e:
            task.status = TaskStatus(state=TaskState.FAILED, error_message=f"计算失败: {str(e)}")
        return task

if __name__ == "__main__":
    agent_instance = CosineAgent()
    print("余弦计算智能体启动，监听端口 8002...")
    run_server(agent_instance, port=8002)

再创建tangent_agent.py：

# tangent_agent.py
import math
import re
from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState

@agent(
    name="TangentCalculator",
    description="专门计算给定角度（弧度制）正切值的智能体。注意：对于(π/2 + kπ)附近的值可能溢出。",
    version="1.0.0"
)
class TangentAgent(A2AServer):

    @skill(
        name="compute_tangent",
        description="计算一个浮点数的正切值",
        tags=["math", "trigonometry", "tangent", "tan"]
    )
    def get_tangent(self, number: float) -> str:
        result = math.tan(number)
        return f"tan({number}) = {result:.6f}"

    def handle_task(self, task):
        try:
            input_text = task.message["content"]["text"]
        except KeyError:
            task.status = TaskStatus(state=TaskState.FAILED, error_message="任务格式错误")
            return task

        match = re.search(r"([-+]?\d*\.?\d+)", input_text)
        if not match:
            task.status = TaskStatus(state=TaskState.FAILED, error_message="未找到有效数字")
            return task

        try:
            number = float(match.group(1))
            print(f"[TangentAgent] 收到计算请求，数字: {number}")
            # 简单处理接近奇异点的情况
            if abs(math.cos(number)) < 1e-10:
                task.status = TaskStatus(state=TaskState.FAILED, error_message=f"输入值 {number} 接近正切函数的奇异点，计算溢出。")
                return task
            response_text = self.get_tangent(number)
            task.artifacts = [{"parts": [{"type": "text", "text": response_text}]}]
            task.status = TaskStatus(state=TaskState.COMPLETED)
        except Exception as e:
            task.status = TaskStatus(state=TaskState.FAILED, error_message=f"计算失败: {str(e)}")
        return task

if __name__ == "__main__":
    agent_instance = TangentAgent()
    print("正切计算智能体启动，监听端口 8003...")
    run_server(agent_instance, port=8003)

运行所有智能体： 现在，你需要打开三个独立的终端窗口（或使用tmux/screen等终端多路复用器），分别运行这三个智能体：

• 终端1：python sine_agent.py
• 终端2：python cosine_agent.py
• 终端3：python tangent_agent.py

确保它们都成功启动并监听各自的端口（8001, 8002, 8003）。至此，你的“数学专家团队”已经就位，但他们还不知道彼此的存在，也无法协同工作。接下来，我们需要创建团队的“协调员”。

3. 构建智能体网络与路由中枢

现在我们有三个独立的智能体在运行。但用户不可能记住每个智能体的地址和功能去手动调用。我们需要一个“大脑”来管理它们，并根据用户请求智能地分派任务。这就是**智能体网络（Agent Network）和路由中枢（Router）**的作用。

3.1 创建主协调器与简单路由

创建一个新的文件orchestrator.py，它将作为我们整个系统的入口和调度中心。

# orchestrator.py
from python_a2a import AgentNetwork, A2AClient, TaskStatus, TaskState
import time

class SimpleMathRouter:
    """
    一个简单的基于关键词的路由器。
    在实际项目中，这里可以替换为基于向量检索或LLM的智能路由器。
    """
    def __init__(self):
        # 定义关键词到智能体名称的映射
        self.keyword_to_agent = {
            'sine': 'SineCalculator',
            'sin': 'SineCalculator',
            'cosine': 'CosineCalculator',
            'cos': 'CosineCalculator',
            'tangent': 'TangentCalculator',
            'tan': 'TangentCalculator',
        }

    def route(self, query: str) -> tuple:
        """
        根据查询文本返回最匹配的智能体名称和置信度。
        返回格式: (agent_name, confidence_score)
        """
        query_lower = query.lower()
        best_agent = None
        best_score = 0.0

        for keyword, agent_name in self.keyword_to_agent.items():
            if keyword in query_lower:
                # 简单计分：关键词出现即得基础分，更长的关键词匹配给予更高分
                score = 0.5 + (len(keyword) * 0.1)
                if score > best_score:
                    best_score = score
                    best_agent = agent_name

        return best_agent, best_score

def main():
    print("=== 初始化数学智能体协作网络 ===")

    # 1. 创建智能体网络（注册中心）
    network = AgentNetwork(name="MathAgentNetwork")

    # 2. 将运行的智能体注册到网络中
    # 参数：智能体名称， 智能体服务地址
    network.add("SineCalculator", "http://localhost:8001")
    network.add("CosineCalculator", "http://localhost:8002")
    network.add("TangentCalculator", "http://localhost:8003")

    print(f"网络已注册智能体: {list(network.agents.keys())}")

    # 3. 初始化路由器
    router = SimpleMathRouter()

    # 4. 定义测试查询
    test_queries = [
        "请计算 0.7854 的正弦值",
        "what is the cosine of 1.0472?",
        "帮我算一下tan(0.7854)",
        "sin of pi over 4",
        "计算 45度角的正切，请先转换成弧度", # 这个查询可能无法被简单路由处理
        "1 + 1 等于多少？", # 没有匹配的智能体
    ]

    # 5. 处理每个查询
    for query in test_queries:
        print(f"\n{'='*40}")
        print(f"处理查询: '{query}'")

        # 5.1 路由决策
        agent_name, confidence = router.route(query)
        if not agent_name:
            print(f"  路由结果: 未找到匹配的数学计算智能体。")
            continue

        print(f"  路由决策: 将任务派发给 '{agent_name}' (置信度: {confidence:.2f})")

        # 5.2 获取智能体客户端并发送任务
        try:
            agent_client = network.get_agent(agent_name)
            if not agent_client:
                print(f"  错误: 网络中未找到智能体 '{agent_name}'")
                continue

            # 创建并发送任务
            task_response = agent_client.ask(query)

            # 5.3 处理响应
            if task_response.status.state == TaskState.COMPLETED:
                # 从artifacts中提取文本结果
                result_text = ""
                for artifact in task_response.artifacts:
                    for part in artifact.get("parts", []):
                        if part.get("type") == "text":
                            result_text += part.get("text", "") + "\n"
                print(f"  智能体响应: {result_text.strip()}")
            elif task_response.status.state == TaskState.FAILED:
                print(f"  任务执行失败: {task_response.status.error_message}")
            else:
                print(f"  任务状态: {task_response.status.state}")

        except Exception as e:
            print(f"  与智能体 '{agent_name}' 通信时发生错误: {e}")

        # 短暂暂停，避免请求过快
        time.sleep(0.5)

    print(f"\n{'='*40}")
    print("所有查询处理完毕。")

if __name__ == "__main__":
    main()

运行协调器： 打开第四个终端窗口，确保前三个智能体仍在运行，然后执行：

python orchestrator.py

你将看到类似以下的输出，展示了路由决策和智能体返回的结果：

=== 初始化数学智能体协作网络 ===
网络已注册智能体: ['SineCalculator', 'CosineCalculator', 'TangentCalculator']

========================================
处理查询: '请计算 0.7854 的正弦值'
  路由决策: 将任务派发给 'SineCalculator' (置信度: 0.60)
  智能体响应: sin(0.7854) = 0.707077

========================================
处理查询: 'what is the cosine of 1.0472?'
  路由决策: 将任务派发给 'CosineCalculator' (置信度: 0.60)
  智能体响应: cos(1.0472) = 0.500171
...

同时，在运行sine_agent.py等智能体的终端里，你也会看到它们收到请求的日志，例如[SineAgent] 收到计算请求，数字: 0.7854。

3.2 深入解析：网络与路由如何工作

我们的orchestrator.py虽然简单，但完整演示了一个A2A协作网络的核心流程：

1. 网络注册（AgentNetwork）：AgentNetwork充当了一个简单的服务注册中心。它维护着一个智能体名称到其网络地址（URL）的映射表。当我们需要与某个智能体通信时，就通过这个名字来查找地址。
2. 路由决策（SimpleMathRouter）：这是系统的“调度员”。它接收用户查询，分析内容，并决定将任务派发给哪个智能体。我们实现了一个基于关键词的简单路由器。在实际生产系统中，这里通常会使用更复杂的技术：
   • 基于向量检索：将智能体的技能描述（Agent Card）和用户查询都转换为向量，通过相似度匹配找到最相关的智能体。
   • LLM驱动路由：使用大语言模型来理解查询的深层意图，甚至动态规划任务分解步骤，选择一系列智能体来协作完成。
3. 任务执行与结果聚合：路由器选定智能体后，通过agent_client.ask(query)发送任务。智能体处理完毕后，结果被封装在task_response.artifacts中返回。协调器负责提取和呈现这个结果。

注意：当前示例中，每个查询只路由给一个智能体。对于更复杂的任务（如“计算sin(x)+cos(x)”），协调器需要先将任务分解，分别调用正弦和余弦智能体，再将结果聚合。这引出了下一个重要话题：工作流编排。

4. 进阶：实现智能工作流与任务编排

简单的关键词路由只能处理单一、明确的请求。现实世界的任务往往是复合型的。我们需要引入**工作流编排（Workflow Orchestration）**的概念。

4.1 构建一个复合计算工作流

假设用户请求是：“计算 sin(π/4) + cos(π/4)”。这个任务需要两个智能体协作完成。我们来创建一个高级的编排器advanced_orchestrator.py。

# advanced_orchestrator.py
from python_a2a import AgentNetwork, A2AClient, TaskStatus, TaskState
import re
import math

class AdvancedMathOrchestrator:
    def __init__(self, network):
        self.network = network
        # 更健壮的路由逻辑
        self.agent_skills = {
            'SineCalculator': ['sine', 'sin'],
            'CosineCalculator': ['cosine', 'cos'],
            'TangentCalculator': ['tangent', 'tan'],
        }

    def parse_expression(self, expression: str):
        """
        简单解析数学表达式，识别需要调用的函数。
        这是一个简化版，实际应用可能需要完整的语法解析器。
        """
        expression = expression.lower().replace('pi', str(math.pi))
        # 查找类似 sin(0.5), cos(1.2) 的模式
        patterns = {
            'SineCalculator': r'sin\(([^)]+)\)',
            'CosineCalculator': r'cos\(([^)]+)\)',
            'TangentCalculator': r'tan\(([^)]+)\)',
        }
        tasks = []
        for agent_name, pattern in patterns.items():
            matches = re.findall(pattern, expression)
            for match in matches:
                try:
                    # 尝试计算表达式中的值，这里简单处理为直接转换
                    # 实际中可能需要一个表达式求值器，如 `eval()`（需谨慎使用）或 `ast.literal_eval`
                    value = float(match)
                    tasks.append({
                        'agent': agent_name,
                        'operation': agent_name.replace('Calculator', '').lower(),
                        'argument': value,
                        'original_match': match
                    })
                except ValueError:
                    # 如果无法直接转换为数字，可能是更复杂的表达式，这里先跳过
                    print(f"警告: 无法解析参数 '{match}'，跳过该任务。")
                    continue
        return tasks

    def execute_workflow(self, query: str):
        print(f"\n开始执行工作流，查询: '{query}'")
        # 1. 解析表达式，分解任务
        subtasks = self.parse_expression(query)
        if not subtasks:
            print("  无法从查询中解析出有效的数学函数任务。")
            return None

        print(f"  解析出 {len(subtasks)} 个子任务:")
        for task in subtasks:
            print(f"    - 调用 {task['agent']} 计算 {task['operation']}({task['argument']})")

        results = {}
        # 2. 顺序执行每个子任务（实际可并行化）
        for task in subtasks:
            agent_name = task['agent']
            arg = task['argument']
            # 构造一个简单的查询文本
            sub_query = f"calculate {task['operation']} of {arg}"

            try:
                agent_client = self.network.get_agent(agent_name)
                if not agent_client:
                    print(f"  错误: 智能体 {agent_name} 未找到")
                    results[agent_name] = {'error': 'Agent not found'}
                    continue

                print(f"  正在调用 {agent_name}...")
                task_response = agent_client.ask(sub_query)

                if task_response.status.state == TaskState.COMPLETED:
                    result_text = ""
                    for artifact in task_response.artifacts:
                        for part in artifact.get("parts", []):
                            if part.get("type") == "text":
                                result_text += part.get("text", "")
                    # 从结果文本中提取数值，例如从 "sin(0.7854) = 0.707077" 提取 0.707077
                    match_val = re.search(r'=\s*([-+]?\d*\.?\d+)', result_text)
                    if match_val:
                        results[agent_name] = float(match_val.group(1))
                        print(f"    -> 结果: {results[agent_name]}")
                    else:
                        results[agent_name] = result_text
                else:
                    error_msg = task_response.status.error_message or "Unknown error"
                    print(f"    -> 任务失败: {error_msg}")
                    results[agent_name] = {'error': error_msg}

            except Exception as e:
                print(f"  调用 {agent_name} 时发生错误: {e}")
                results[agent_name] = {'error': str(e)}

        # 3. 聚合结果（这里简单打印，实际可根据表达式进行运算）
        print(f"\n  所有子任务完成。原始结果集: {results}")
        # 示例：如果查询是 sin(a)+cos(b)，这里可以执行实际的加法
        # 本例中我们只展示聚合的概念
        return results

def main():
    # 初始化网络（同前）
    network = AgentNetwork(name="AdvancedMathNetwork")
    network.add("SineCalculator", "http://localhost:8001")
    network.add("CosineCalculator", "http://localhost:8002")
    network.add("TangentCalculator", "http://localhost:8003")

    orchestrator = AdvancedMathOrchestrator(network)

    # 测试复合查询
    complex_queries = [
        "计算 sin(0.7854) + cos(0.7854)",
        "sin(1.0) 和 cos(0.5) 分别是多少？",
        "tan(0.5)",
    ]

    for query in complex_queries:
        final_result = orchestrator.execute_workflow(query)
        # 这里可以添加更复杂的逻辑来基于原始表达式（如sin(a)+cos(b)）计算结果
        # 例如，如果结果都是数字，可以尝试计算表达式
        print("-" * 50)

if __name__ == "__main__":
    main()

这个高级编排器做了几件关键的事：

• 表达式解析：parse_expression方法尝试从自然语言查询中识别出数学函数调用（如sin(...)，cos(...)）。这里使用了简单的正则表达式，在真实场景中可能需要更强大的自然语言处理（NLP）或语法解析器。
• 任务分解：将识别出的每个函数调用转化为一个独立的子任务（Subtask），并指定负责的智能体。
• 顺序执行与结果收集：目前是顺序调用各个智能体，收集它们返回的结果。在生产系统中，这完全可以并行化以提高效率。
• 结果聚合：收集所有子任务的结果后，可以根据原始表达式的语义进行聚合计算（例如，将sin和cos的结果相加）。本例中只展示了收集步骤，实际的表达式求值需要更复杂的逻辑。

运行这个高级编排器（同样需要智能体在运行）：

python advanced_orchestrator.py

你会看到它成功解析了复合表达式，并分别调用了对应的智能体。

4.2 设计模式：编排（Orchestration）与协同（Choreography）

在构建多智能体系统时，有两种主流的协作模式：

模式	描述	优点	缺点	适用场景
编排 (Orchestration)	一个中心化的“指挥者”（如我们的AdvancedMathOrchestrator）负责接收任务，将其分解，指挥各个智能体工作，并汇总结果。	控制流清晰，易于监控和调试，适合复杂、有严格顺序的工作流。	中心节点可能成为性能和单点故障的瓶颈。	需要严格事务性、步骤间有依赖的复杂业务流程。
协同 (Choreography)	没有中心指挥者。智能体之间通过事件或消息直接通信。每个智能体完成自己的工作后，可能会触发下一个智能体开始工作。	去中心化，扩展性好，更灵活，避免了单点故障。	控制流分散，整体行为难以理解和调试，错误处理复杂。	事件驱动、松耦合、智能体间交互模式相对固定的场景。

在我们的数学计算例子中，由于任务逻辑简单且需要聚合结果，使用编排模式更为合适。对于更动态、更去中心化的场景（例如一个智能体发现数据问题后自动触发另一个清理数据的智能体），则可以考虑协同模式。

5. 生产级考量与最佳实践

将实验性的智能体网络推向生产环境，需要考虑更多因素。以下是一些关键的最佳实践和扩展思路。

5.1 健壮性与错误处理

我们的示例代码错误处理还很基础。一个健壮的系统需要：

• 智能体健康检查：编排器应定期检查智能体是否在线和响应正常。
• 任务超时与重试：为每个任务设置超时，并在失败时进行有限次数的重试。
• 降级策略：当某个智能体不可用时，是否有备用方案？例如，如果正切计算智能体失败，是否可以尝试用正弦和余弦智能体的结果来计算（tan = sin/cos）？
• 结果验证：对智能体返回的结果进行合理性检查（例如，正弦、余弦值应在[-1,1]之间）。

下面是一个增加了超时和重试的客户端调用示例：

import requests
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class ResilientAgentClient:
    def __init__(self, agent_url):
        self.agent_url = agent_url

    @retry(
        stop=stop_after_attempt(3), # 最多重试3次
        wait=wait_exponential(multiplier=1, min=1, max=10), # 指数退避等待
        retry=retry_if_exception_type((requests.ConnectionError, requests.Timeout))
    )
    def ask_with_retry(self, query, timeout=5.0):
        """带重试机制的任务请求"""
        # 这里简化了，实际应使用A2A客户端库
        # 假设通过HTTP POST发送请求
        payload = {"message": {"content": {"text": query}}}
        response = requests.post(f"{self.agent_url}/ask", json=payload, timeout=timeout)
        response.raise_for_status()
        return response.json()

5.2 可观测性与监控

当你有数十上百个智能体协同工作时， visibility 至关重要。

• 集中式日志：所有智能体和编排器应将日志发送到统一的平台（如ELK Stack, Loki）。
• 分布式追踪：为每个用户请求生成一个唯一的trace_id，并在流经的每个智能体中传递。这样可以在日志中完整追踪一个请求的生命周期。可以使用OpenTelemetry等标准。
• 指标收集：监控每个智能体的请求量、成功率、延迟等指标。这对于容量规划和故障排查至关重要。

# 伪代码：在智能体处理任务时添加追踪信息
def handle_task(self, task):
    trace_id = task.context.get('trace_id', 'unknown')
    print(f"[{trace_id}] 开始处理任务: {task.id}")
    # ... 处理逻辑 ...
    print(f"[{trace_id}] 任务处理完成，状态: {task.status.state}")

5.3 安全与权限控制

在生产环境中，不能允许任意智能体相互调用。

• 身份认证：每个智能体应有自己的身份标识（如API Key, mTLS证书）。
• 授权：定义哪些智能体可以调用另一些智能体的哪些技能。例如，一个“数据清洗”智能体可能被授权调用“数据库查询”智能体，但不能调用“发送邮件”智能体。
• 输入验证与净化：智能体在处理输入前，必须进行严格的验证，防止注入攻击或恶意输入。

5.4 动态扩展与自动发现

一个理想的A2A系统应该能够动态适应变化。

• 服务发现：使用像Consul、Etcd或ZooKeeper这样的服务发现工具，而不是在代码中硬编码智能体地址。新智能体启动时可以自动注册，下线时自动注销。
• 负载均衡：如果某个智能体（如“图像识别”）有多个实例，路由器和负载均衡器应能将请求分发到最空闲的实例上。
• 自动缩放：根据监控指标（如队列长度、CPU使用率），自动创建或销毁智能体实例。

构建这样一个系统并非一日之功，但遵循模块化、关注点分离的原则，从一个小而精的智能体网络开始，逐步迭代扩展，是通往成熟AI架构的可靠路径。