多Agent通信配置怎么落地，从传输方式到消息路由的实操步骤
多个 Agent 之间通信配置的核心，是先确定它们运行在哪、用什么协议说话、消息长什么样。
局域网内的小规模协作，用 HTTP 直连最省事；
跨机器、需要削峰或广播的场景，加一层消息队列更稳。
下面按从简到繁的顺序，把两种配置方式都过一遍。
先确认通信范围和传输方式
动手之前先回答三个问题，答案直接决定配置写法：
- Agent 是否在同一台机器或同一局域网？ 同机或同局域网可以用 HTTP 直连，不需要额外中间件。
- 消息是否需要排队、重试或广播？ 需要的话引入 Redis 或 RabbitMQ 这类消息队列，通信双方不直接连接。
- 是否要求同步返回结果？ 需要立即拿到结果的用 HTTP；只要求最终送达的用队列。
举个例子：一个负责抓取网页的 Agent 和一个负责总结的 Agent，在同一台服务器上，总结 Agent 需要等抓取结果才能继续，这种就用 HTTP 直连。
如果抓取 Agent 会突然产生大量任务，总结 Agent 处理不过来，就加消息队列缓冲。
HTTP 直连方式的最小配置
假设两个 Agent 都用 Python 编写，运行在同一台机器的不同端口。
接收方 Agent 监听 127.0.0.1:8001，发送方通过 POST 请求把消息发过去。
接收方用标准库起一个简单服务，保存为 agent_receiver.py：
from http.server import BaseHTTPRequestHandler, HTTPServer
import json
class Handler(BaseHTTPRequestHandler):
def do_POST(self):
length = int(self.headers.get('Content-Length', 0))
body = self.rfile.read(length)
msg = json.loads(body)
print('收到消息:', msg)
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(json.dumps({'status': 'ok'}).encode())
if __name__ == '__main__':
server = HTTPServer(('127.0.0.1', 8001), Handler)
print('接收 Agent 已启动，监听 8001')
server.serve_forever()发送方 Agent 用 requests 库发消息，保存为 agent_sender.py：
import requests
url = 'http://127.0.0.1:8001/message'
payload = {'from': 'agent_a', 'to': 'agent_b', 'content': '请总结这篇文章'}
resp = requests.post(url, json=payload, timeout=5)
print(resp.status_code, resp.json())先在终端运行 python agent_receiver.py，看到启动提示后，另开一个终端运行 python agent_sender.py。
发送方输出 200 {'status': 'ok'}，接收方终端打印出消息内容，说明通信配置成功。
这里有两个容易忽略的点：Content-Length 必须按实际字节数读取，否则中文内容会截断；
发送方要设置 timeout，避免接收方卡死时发送方一直等待。
用消息队列解耦多个 Agent
当 Agent 数量增加，或者发送方和接收方启动时间不一致时，直连的失败率会上升。
这时用 Redis 的列表结构做简单队列，发送方 lpush，接收方 brpop 阻塞等待。
安装 Redis 客户端库：
pip install redis发送方把消息写入队列 agent_tasks：
import redis, json
r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)
msg = {'from': 'agent_a', 'content': '抓取并总结'}
r.lpush('agent_tasks', json.dumps(msg, ensure_ascii=False))
print('已投递')接收方循环阻塞读取：
import redis, json
r = redis.Redis(host='127.0.0.1', port=6379, decode_responses=True)
while True:
_, raw = r.brpop('agent_tasks', timeout=0)
msg = json.loads(raw)
print('处理任务:', msg)先启动接收方，再运行发送方。
接收方会立即打印出任务内容。decode_responses=True 不能漏，否则拿到的是字节串，json.loads 会报类型错误。
消息队列方式下，发送方和接收方不需要知道对方地址，只要约定队列名和消息格式即可。
代价是多了一个 Redis 服务要维护，需要确认 redis-server 已启动且端口 6379 可访问。
联调时值得检查的几处配置
通信跑通不代表配置正确，下面几点建议逐项核对：
- 端口与防火墙：HTTP 方式下，如果 Agent 不在同一台机器，接收方监听的地址不能是
127.0.0.1，要改成0.0.0.0或内网 IP，并放行对应端口。 - 消息格式统一：发送方和接收方对字段名要有一致约定，建议固定
from、to、content这类字段，避免一边叫msg一边叫content。 - 错误处理：HTTP 发送方要判断返回码，非 200 时记录日志并决定是否重试；队列接收方要捕获
json.loads异常，避免一条坏消息让循环退出。 - 超时设置：
requests.post的timeout和brpop的timeout含义不同，前者是等待响应时间，后者是阻塞等待新消息的时间，按业务需要分别设置。
如果发送方报 Connection refused，先确认接收方进程是否在运行、端口是否写对；
如果接收方打印乱码或解析失败，检查发送方是否用了 json= 参数以及 ensure_ascii 设置。
验证通信是否真正生效
配置完成后，用下面三个检查确认链路正常：
- 在接收方终端确认启动日志出现，没有端口占用报错。
- 发送方执行一次请求或投递，观察返回码和接收方输出是否一致。
- 停掉接收方再发送一次，确认发送方能在超时或连接失败时给出明确错误，而不是静默卡住。
第三点常被跳过，但它决定了故障时你能不能快速定位。
多 Agent 通信配置不是配完就结束，把失败路径也验证一遍，后续增加 Agent 时才不容易互相拖垮。
