在实际企业级应用开发中，PHP 常因其“脚本语言”的刻板印象而被低估，尤其是在人工智能和复杂系统架构领域。然而，现代 PHP（如 PHP 8.x）在性能、类型系统、异步编程和生态工具上已今非昔比。本文将展示如何利用 PHP 构建一个功能完整的“多智能体系统”（Multi-Agent System, MAS），涵盖智能体抽象、消息路由、任务编排与并发执行等核心模块。通过这个实践，你会看到 PHP 不仅能够胜任此类复杂任务，其清晰的语法、强大的 Web 原生支持以及成熟的依赖管理，在某些场景下甚至能带来比 Python 更简洁、更易于部署和维护的实现路径。

这个系统将模拟一个简单的电商订单处理场景，包含多个智能体：OrderAgent（接收订单）、InventoryAgent（检查库存）、PaymentAgent（处理支付）和NotificationAgent（发送通知）。我们将从零开始，逐步实现智能体的定义、消息传递机制、基于事件的协调逻辑，并最终通过一个 Web 接口触发整个流程。文章会详细解释每一步的设计考量、代码实现以及如何排查常见问题。

1. 理解多智能体系统的核心概念与 PHP 的适配性

多智能体系统是由多个自治的、相互交互的智能体组成的计算系统。每个智能体能感知环境（包括其他智能体），并基于自身的目标和规则做出决策，通过协作或竞争来完成复杂任务。其核心组件通常包括：智能体（Agent）、环境（Environment）、消息（Message）和协调机制（Coordination）。

1.1 为什么选择 PHP 来实现？

在技术选型时，我们通常会考虑生态、性能和开发效率。Python 在 AI 领域有丰富的库，但 PHP 在以下方面有其独特优势：

• Web 原生与快速原型：MAS 常需要通过 HTTP、WebSocket 与外部系统交互。PHP 内置了强大的 HTTP 处理能力，配合Swoole或ReactPHP等扩展，可以轻松构建高性能的异步服务，非常适合作为智能体系统的“网关”或“协调器”。
• 清晰的工程结构与依赖管理：Composer是现代 PHP 项目的基石，它能优雅地管理依赖和自动加载。配合 PSR 标准，可以构建出模块清晰、易于测试的智能体组件。
• 强类型与面向对象：PHP 8 引入了联合类型、属性声明、match表达式等特性，使得代码的意图更明确，减少了运行时错误，这对于构建状态复杂、交互频繁的智能体系统至关重要。
• 成熟的部署与运维生态：PHP 应用在 Docker、Kubernetes 以及各类 PaaS 平台上有极其成熟的部署方案和监控工具，便于系统上线和维护。

1.2 系统设计概览

我们将构建的系统采用“中心化协调”与“去中心化执行”相结合的架构。一个中央的AgentCoordinator（协调器）负责接收初始任务、分解工作流，并将消息分发给对应的智能体。每个智能体独立运行其业务逻辑，处理完毕后通过事件或回调通知协调器。消息传递采用简单的基于事件或队列的异步模式。

系统流程如下：

1. 用户通过 HTTP API 提交一个订单请求。
2. OrderAgent接收请求，验证基础信息，并创建一个Order领域对象。
3. AgentCoordinator根据预定义的工作流，向InventoryAgent发送“检查库存”消息。
4. InventoryAgent查询（模拟）库存服务，返回结果。
5. 协调器根据库存结果，决定是触发PaymentAgent（有库存）还是直接结束流程（无库存）。
6. PaymentAgent模拟支付处理。
7. 支付成功后，NotificationAgent被触发，发送订单确认通知。
8. 最终，协调器将处理结果汇总并返回给用户。

2. 环境准备与项目初始化

在开始编码前，需要确保开发环境就绪。本项目基于 PHP 8.2+ 和 Composer。

2.1 环境要求

可以通过命令行检查环境：

php -v # 应输出 PHP 8.2.x 或更高版本 composer --version # 应输出 Composer version 2.x

2.2 初始化项目与目录结构

创建一个新目录并初始化 Composer 项目。

mkdir php-multi-agent-system && cd php-multi-agent-system composer init --name=your-vendor/multi-agent-system --type=project --no-interaction

编辑生成的composer.json，设置 PSR-4 自动加载并添加一个简单的路由库（如nikic/fast-route）用于 Web 接口。

{ "name": "your-vendor/multi-agent-system", "autoload": { "psr-4": { "App\\": "src/" } }, "require": { "php": "^8.2", "nikic/fast-route": "^1.3" }, "require-dev": { "phpunit/phpunit": "^10.0" } }

运行composer install安装依赖。

创建以下目录结构，这是清晰组织代码的关键：

php-multi-agent-system/ ├── composer.json ├── composer.lock ├── public/ │ └── index.php # Web 应用入口 ├── src/ │ ├── Agent/ # 智能体抽象与具体实现 │ ├── Message/ # 消息对象 │ ├── Coordination/ # 协调器 │ ├── Workflow/ # 工作流定义 │ └── Model/ # 领域模型（如 Order） └── tests/ # 单元测试

3. 核心模块实现：从消息、智能体到协调器

我们将自底向上构建系统，首先定义消息格式，然后实现智能体基类，最后构建协调器。

3.1 定义消息体

消息是智能体间通信的载体。我们定义一个简单的Message值对象。

// src/Message/Message.php namespace App\Message; class Message { public function __construct( public readonly string $type, // 消息类型，如 'check_inventory', 'process_payment' public readonly array $payload, // 消息负载，关联数组 public readonly string $senderId, // 发送者ID public readonly ?string $correlationId = null, // 关联ID，用于追踪工作流 public readonly ?string $replyTo = null // 需要回复的队列或主题 ) { } public function toJson(): string { return json_encode([ 'type' => $this->type, 'payload' => $this->payload, 'senderId' => $this->senderId, 'correlationId' => $this->correlationId, 'replyTo' => $this->replyTo, ], JSON_THROW_ON_ERROR); } public static function fromJson(string $json): self { $data = json_decode($json, true, 512, JSON_THROW_ON_ERROR); return new self( $data['type'], $data['payload'], $data['senderId'], $data['correlationId'] ?? null, $data['replyTo'] ?? null ); } }

关键点：

• 使用readonly属性（PHP 8.2+）确保消息的不可变性，这在并发环境中很重要。
• correlationId用于将同一业务流程中的多个消息关联起来，便于日志追踪和调试。
• 提供toJson和fromJson方法，方便消息在网络或队列中序列化传输。

3.2 实现智能体抽象基类

所有具体智能体都应继承自一个抽象基类，它定义了智能体的生命周期和消息处理接口。

// src/Agent/AbstractAgent.php namespace App\Agent; use App\Message\Message; abstract class AbstractAgent { protected string $id; public function __construct(string $id) { $this->id = $id; } public function getId(): string { return $this->id; } // 核心方法：处理接收到的消息 abstract public function handle(Message $message): ?Message; // 智能体启动时的逻辑（如连接数据库、订阅队列） public function boot(): void { // 默认空实现，子类按需覆盖 } // 智能体关闭时的清理逻辑 public function shutdown(): void { // 默认空实现，子类按需覆盖 } }

3.3 实现具体智能体：以 InventoryAgent 为例

现在实现一个检查库存的智能体。它接收包含productId和quantity的消息，并返回库存是否充足。

// src/Agent/InventoryAgent.php namespace App\Agent; use App\Message\Message; class InventoryAgent extends AbstractAgent { // 模拟一个内存中的库存表 private array $inventory = [ 'PROD-001' => 10, 'PROD-002' => 5, 'PROD-003' => 0, ]; public function handle(Message $message): ?Message { // 只处理特定类型的消息 if ($message->type !== 'check_inventory') { return null; // 或者抛出一个异常，取决于错误处理策略 } $productId = $message->payload['productId'] ?? null; $requestedQty = (int) ($message->payload['quantity'] ?? 0); if (!$productId || $requestedQty <= 0) { // 返回一个错误响应消息 return new Message( 'inventory_checked', ['success' => false, 'error' => 'Invalid productId or quantity'], $this->id, $message->correlationId ); } $available = $this->inventory[$productId] ?? 0; $isSufficient = $available >= $requestedQty; // 模拟一个简单的库存扣减（生产环境需要事务） if ($isSufficient) { $this->inventory[$productId] = $available - $requestedQty; } // 返回处理结果消息 return new Message( 'inventory_checked', [ 'success' => true, 'sufficient' => $isSufficient, 'productId' => $productId, 'requested' => $requestedQty, 'remaining' => $this->inventory[$productId] ?? 0, ], $this->id, $message->correlationId ); } }

关键点：

• handle方法是智能体的核心，它决定了智能体能处理哪些消息类型。
• 返回的Message对象包含了处理结果，其correlationId与原消息一致，方便协调器进行匹配。
• 这里使用了内存数组模拟库存，实际项目中应替换为数据库或外部服务调用。

3.4 构建协调器

协调器是系统的大脑，它维护着所有智能体的注册表，并根据预定义的工作流路由消息。

// src/Coordination/AgentCoordinator.php namespace App\Coordination; use App\Agent\AbstractAgent; use App\Message\Message; use RuntimeException; class AgentCoordinator { /** @var array<string, AbstractAgent> 智能体ID到实例的映射 */ private array $agents = []; /** @var array<string, callable> 消息类型到处理工作流的映射 */ private array $workflows = []; public function registerAgent(AbstractAgent $agent): void { $this->agents[$agent->getId()] = $agent; $agent->boot(); } public function registerWorkflow(string $messageType, callable $workflow): void { $this->workflows[$messageType] = $workflow; } /** * 发送消息给指定智能体并获取响应（同步模式） */ public function sendMessage(Message $message, string $agentId): ?Message { if (!isset($this->agents[$agentId])) { throw new RuntimeException("Agent not found: {$agentId}"); } $agent = $this->agents[$agentId]; return $agent->handle($message); } /** * 触发一个工作流（通常由初始消息驱动） */ public function triggerWorkflow(Message $initialMessage): mixed { $messageType = $initialMessage->type; if (!isset($this->workflows[$messageType])) { throw new RuntimeException("No workflow defined for message type: {$messageType}"); } // 执行注册的工作流回调，并将协调器自身传入，以便工作流内可以调用 sendMessage return ($this->workflows[$messageType])($initialMessage, $this); } public function shutdownAll(): void { foreach ($this->agents as $agent) { $agent->shutdown(); } } }

关键点：

• registerWorkflow方法将消息类型与一个可调用的工作流逻辑绑定。这种设计使得工作流可以灵活定义，甚至可以从配置文件加载。
• triggerWorkflow是系统的入口，它根据初始消息的类型找到对应的工作流并执行。
• 当前是同步调用$agent->handle()，这意味着智能体按顺序执行。这对于理解流程是清晰的，但在生产环境中，你可能需要将其改为基于队列的异步模式以提高并发能力。

4. 定义工作流与组装系统

有了智能体和协调器，我们需要定义订单处理的具体工作流，并将所有组件组装起来。

4.1 定义订单处理工作流

工作流定义了消息处理的顺序和逻辑。我们在一个集中的地方（例如一个工厂类或配置文件）定义它。

// src/Workflow/OrderProcessingWorkflow.php namespace App\Workflow; use App\Coordination\AgentCoordinator; use App\Message\Message; class OrderProcessingWorkflow { public static function create(): callable { return function (Message $initialMessage, AgentCoordinator $coordinator) { // 1. 初始消息应为 'new_order' $orderData = $initialMessage->payload; $correlationId = $initialMessage->correlationId ?? uniqid('order_', true); // 2. 发送消息给 InventoryAgent 检查库存 $inventoryCheckMsg = new Message( 'check_inventory', [ 'productId' => $orderData['productId'], 'quantity' => $orderData['quantity'], ], $initialMessage->senderId, $correlationId ); $inventoryResult = $coordinator->sendMessage($inventoryCheckMsg, 'inventory_agent'); if (!$inventoryResult || !($inventoryResult->payload['success'] ?? false)) { return ['success' => false, 'stage' => 'inventory_check_failed', 'details' => $inventoryResult?->payload]; } if (!($inventoryResult->payload['sufficient'] ?? false)) { return ['success' => false, 'stage' => 'out_of_stock', 'details' => $inventoryResult->payload]; } // 3. 库存充足，处理支付 $paymentMsg = new Message( 'process_payment', [ 'orderId' => $correlationId, 'amount' => $orderData['amount'], 'currency' => $orderData['currency'] ?? 'USD', ], $initialMessage->senderId, $correlationId ); $paymentResult = $coordinator->sendMessage($paymentMsg, 'payment_agent'); if (!$paymentResult || !($paymentResult->payload['success'] ?? false)) { // 支付失败，可能需要触发库存补偿逻辑（此处简化） return ['success' => false, 'stage' => 'payment_failed', 'details' => $paymentResult?->payload]; } // 4. 支付成功，发送通知 $notificationMsg = new Message( 'send_notification', [ 'orderId' => $correlationId, 'customerEmail' => $orderData['customerEmail'], 'status' => 'confirmed', ], $initialMessage->senderId, $correlationId ); $coordinator->sendMessage($notificationMsg, 'notification_agent'); // 通知可能异步发送，不阻塞主流程 // 5. 返回最终成功结果 return [ 'success' => true, 'orderId' => $correlationId, 'message' => 'Order processed successfully', 'inventory' => $inventoryResult->payload, 'payment' => $paymentResult->payload, ]; }; } }

关键点：

• 工作流是一个闭包，它接收初始消息和协调器，并编排多个智能体的调用。
• 每个步骤都检查上一步的结果，决定流程是继续还是终止。
• correlationId在整个流程中传递，将所有相关消息串联起来。

4.2 创建其他智能体与引导程序

我们需要创建PaymentAgent和NotificationAgent（OrderAgent的功能已由 Web 控制器和协调器替代）。然后，在一个引导文件（如public/index.php）中初始化整个系统。

PaymentAgent 示例：

// src/Agent/PaymentAgent.php namespace App\Agent; use App\Message\Message; class PaymentAgent extends AbstractAgent { public function handle(Message $message): ?Message { if ($message->type !== 'process_payment') { return null; } // 模拟支付处理，80%成功率 $success = mt_rand(1, 10) <= 8; return new Message( 'payment_processed', [ 'success' => $success, 'transactionId' => $success ? 'TXN-' . bin2hex(random_bytes(8)) : null, 'amount' => $message->payload['amount'], 'orderId' => $message->payload['orderId'], ], $this->id, $message->correlationId ); } }

NotificationAgent 示例：

// src/Agent/NotificationAgent.php namespace App\Agent; use App\Message\Message; class NotificationAgent extends AbstractAgent { public function handle(Message $message): ?Message { if ($message->type !== 'send_notification') { return null; } // 模拟发送邮件或短信 error_log("[NotificationAgent] Sending notification for order: " . ($message->payload['orderId'] ?? 'N/A')); // 在实际项目中，这里会调用邮件服务（如 Symfony Mailer）或消息队列 return new Message( 'notification_sent', ['success' => true, 'orderId' => $message->payload['orderId']], $this->id, $message->correlationId ); } }

系统引导与 Web 入口：

// public/index.php require_once __DIR__ . '/../vendor/autoload.php'; use App\Agent\InventoryAgent; use App\Agent\PaymentAgent; use App\Agent\NotificationAgent; use App\Coordination\AgentCoordinator; use App\Message\Message; use App\Workflow\OrderProcessingWorkflow; use FastRoute\Dispatcher; use FastRoute\RouteCollector; // 1. 初始化协调器并注册智能体 $coordinator = new AgentCoordinator(); $coordinator->registerAgent(new InventoryAgent('inventory_agent')); $coordinator->registerAgent(new PaymentAgent('payment_agent')); $coordinator->registerAgent(new NotificationAgent('notification_agent')); // 2. 注册工作流 $coordinator->registerWorkflow('new_order', OrderProcessingWorkflow::create()); // 3. 设置简单的路由（使用 fast-route） $dispatcher = FastRoute\simpleDispatcher(function(RouteCollector $r) { $r->post('/api/orders', 'handle_order'); }); // 4. 处理 HTTP 请求 $httpMethod = $_SERVER['REQUEST_METHOD']; $uri = $_SERVER['REQUEST_URI']; $routeInfo = $dispatcher->dispatch($httpMethod, $uri); switch ($routeInfo[0]) { case Dispatcher::NOT_FOUND: http_response_code(404); echo json_encode(['error' => 'Not Found']); break; case Dispatcher::METHOD_NOT_ALLOWED: http_response_code(405); echo json_encode(['error' => 'Method Not Allowed']); break; case Dispatcher::FOUND: $handler = $routeInfo[1]; $vars = $routeInfo[2]; if ($handler === 'handle_order') { handleOrder($coordinator); } break; } function handleOrder(AgentCoordinator $coordinator): void { // 简单获取 JSON 输入 $input = json_decode(file_get_contents('php://input'), true); if (json_last_error() !== JSON_ERROR_NONE) { http_response_code(400); echo json_encode(['error' => 'Invalid JSON']); return; } // 验证必要字段 $required = ['productId', 'quantity', 'amount', 'customerEmail']; foreach ($required as $field) { if (empty($input[$field])) { http_response_code(400); echo json_encode(['error' => "Missing field: {$field}"]); return; } } // 创建初始消息，触发工作流 $initialMessage = new Message( 'new_order', $input, 'api_gateway' // 发送者ID ); try { $result = $coordinator->triggerWorkflow($initialMessage); header('Content-Type: application/json'); echo json_encode($result, JSON_PRETTY_PRINT); } catch (Throwable $e) { http_response_code(500); error_log('Order processing failed: ' . $e->getMessage()); echo json_encode(['error' => 'Internal Server Error', 'message' => $e->getMessage()]); } }

5. 运行验证与结果分析

现在，我们可以启动服务并测试整个多智能体系统。

5.1 启动开发服务器

在项目根目录下运行：

php -S localhost:8080 -t public

这将启动一个内置的 PHP Web 服务器，监听 8080 端口。

5.2 发送测试请求

使用curl或 Postman 等工具发送一个 POST 请求。

curl -X POST http://localhost:8080/api/orders \ -H "Content-Type: application/json" \ -d '{ "productId": "PROD-001", "quantity": 2, "amount": 99.98, "customerEmail": "customer@example.com", "currency": "USD" }'

5.3 预期输出与解析

成功响应示例：

{ "success": true, "orderId": "order_6672c3f09a123", "message": "Order processed successfully", "inventory": { "success": true, "sufficient": true, "productId": "PROD-001", "requested": 2, "remaining": 8 }, "payment": { "success": true, "transactionId": "TXN-4a1b2c3d4e5f", "amount": 99.98, "orderId": "order_6672c3f09a123" } }

失败响应示例（库存不足）：

{ "success": false, "stage": "out_of_stock", "details": { "success": true, "sufficient": false, "productId": "PROD-003", "requested": 1, "remaining": 0 } }

失败响应示例（支付失败）：

{ "success": false, "stage": "payment_failed", "details": { "success": false, "transactionId": null, "amount": 99.98, "orderId": "order_6672c3f09a456" } }

流程验证：

1. 日志追踪：观察 PHP 错误日志（或NotificationAgent中的error_log），可以看到通知发送的记录。
2. 状态检查：再次请求PROD-001的库存，会发现remaining已从 10 减少为 8。
3. 工作流中断：尝试订购PROD-003（库存为 0）或模拟支付失败（PaymentAgent有 20% 失败率），流程会在相应阶段停止并返回明确错误。

6. 常见问题排查与优化方向

将系统运行起来只是第一步。在实际开发和部署中，你会遇到各种问题。以下是典型的问题排查路径和优化建议。

6.1 常见问题排查表

6.2 从演示到生产：关键优化方向

当前的实现是同步且单进程的，适合理解概念。要用于生产，需要考虑以下方面：

1. 异步与并发：

   • 问题：同步调用会阻塞，如果PaymentAgent调用外部支付网关耗时 2 秒，整个 API 响应就会延迟 2 秒。
   • 方案：将AgentCoordinator::sendMessage改为异步。可以使用队列系统（如 RabbitMQ、Redis Streams、Kafka）或 PHP 的异步框架（如 Swoole、ReactPHP）。智能体作为队列消费者，协调器只需投递消息，无需等待响应。工作流状态需要持久化。

2. 状态持久化与恢复：

   • 问题：工作流状态（进行到哪一步）和智能体状态（如库存）都在内存中，进程重启即丢失。
   • 方案：使用数据库记录工作流实例和当前步骤。智能体的状态（如库存）应存储在数据库或 Redis 中。实现幂等性处理，防止消息重复消费导致状态错误。

3. 可观测性：

   • 问题：系统复杂后，难以追踪一个请求流经了哪些智能体，耗时多少，是否出错。
   • 方案：在所有消息的payload中注入统一的traceId。在每个智能体的handle方法开始和结束时记录结构化日志（发送到 ELK 或 Loki）。集成 APM 工具（如 OpenTelemetry）来收集链路追踪和指标。

4. 配置化与动态注册：

   • 问题：智能体和工作流硬编码在index.php中，变更需要改代码重启。
   • 方案：将智能体和工作流的定义移到配置文件（如 YAML）或数据库中。协调器启动时读取配置，动态实例化智能体和注册工作流。这支持热更新和更灵活的编排。

5. 错误处理与补偿：

   • 问题：当前流程中，支付失败后，库存已被扣减，但订单未成功，造成了数据不一致。
   • 方案：引入 Saga 分布式事务模式或补偿机制。例如，在支付失败后，向InventoryAgent发送一个compensate_inventory消息，将库存加回。这需要工作流引擎支持回滚或补偿操作。

6.3 引入消息队列的简单改造示例

以下是如何使用 Redis 作为简单队列，将InventoryAgent改为异步消费者的伪代码思路：

协调器投递消息：

// 在 AgentCoordinator 中 private Redis $redis; public function sendMessageAsync(Message $message, string $queueName): void { $this->redis->lPush($queueName, $message->toJson()); }

独立的库存消费者脚本：

// bin/consume_inventory.php require_once __DIR__ . '/../vendor/autoload.php'; $agent = new InventoryAgent('inventory_worker_1'); $redis = new Redis(); $redis->connect('127.0.0.1', 6379); while (true) { $json = $redis->brPop('queue:inventory', 30); // 阻塞获取 if ($json) { $message = Message::fromJson($json[1]); $result = $agent->handle($message); if ($result && $result->replyTo) { // 将处理结果放回指定的回复队列 $redis->lPush($result->replyTo, $result->toJson()); } } }

这样，Web 请求可以立即返回一个“已接收”的响应，而订单处理在后台异步完成。协调器需要监听回复队列来推进工作流。

7. 总结与最佳实践

通过这个项目，我们验证了使用 PHP 构建多智能体系统的可行性。其核心在于良好的抽象（消息、智能体、协调器）和清晰的职责划分。PHP 的面向对象特性和丰富的生态使得实现这些抽象变得直观。

在 PHP 项目中应用多智能体模式的最佳实践：

1. 明确边界：每个智能体应封装一个明确的业务能力（如库存管理、支付），并仅通过定义良好的消息接口进行通信。避免智能体间直接调用方法或共享内存。
2. 消息设计先行：在编写智能体逻辑之前，先设计好消息协议（类型、负载结构）。使用 JSON Schema 或 Protobuf 来定义和验证消息格式，确保兼容性。
3. 幂等性至关重要：由于消息可能重传，智能体的handle方法应设计为幂等的。即，使用相同消息多次调用，产生的系统副作用应一致。可以通过在数据库中记录已处理消息的 ID 来实现。
4. 同步转异步：对于学习原型，同步调用更简单。但对于生产系统，尽早引入消息队列（如 RabbitMQ）或事件总线，将智能体解耦，提高系统的可伸缩性和韧性。
5. 重视可观测性：从第一天起就在消息中注入traceId和spanId，并记录结构化日志。这比事后加装要容易得多。
6. 版本化消息：当需要修改消息格式时，引入版本字段（如type: 'check_inventory.v2'），并让智能体同时支持新旧版本一段时间，实现平滑升级。

这个用 PHP 手撸的多智能体系统，展示了 PHP 在处理复杂业务编排和异步通信方面的能力。它不是一个要取代 Python AI 生态的框架，而是一种利用 PHP 优势（快速开发、Web 集成、稳健部署）来构建分布式、可扩展业务系统的架构模式。当你面临需要将复杂业务流程模块化、解耦和弹性伸缩的场景时，不妨考虑采用智能体模型，而 PHP 完全有能力作为实现它的坚实底座。