你与 AI agents 之间缺失的层
抱歉，我需要您提供要翻译的具体文本内容（除代码块和 URL 之外），才能为您完成简体中文翻译。请将文章的正文粘贴在这里，我会按照要求保留源链接、格式和技术术语进行翻译。
开发者常用的三种模式
模式 1：Terminal babysitting
python my_agent.py >> agent.log 2>&1 &
tail -f agent.log你运行代理，查看日志，并在它崩溃时手动重启。
- 表现良好 对于一次性批处理作业（爬取、处理、运行至完成并写出输出文件的任务）。
- 一旦你的代理需要在人机交互的中途做出决定，或当你同时运行多个代理并需要了解全局状态时，它就会崩溃。
模式 2：Polling (REST API)
# Check status
curl http://localhost:8000/status
# {"status": "running", "task": "scraping page 42/100", "errors": 0}
# Ask it something
curl -X POST http://localhost:8000/query \
-d '{"question": "what have you found so far?"}'更好——有了可编程的对话。
但权力动态是单向的：代理只能响应请求。如果它在凌晨 2 点触发了速率限制或遇到意外异常，它无法告诉你；你需要另写脚本专门去检查它的状态。
- 适用于同步工具，但有趣的代理往往是长期运行的，需要能够主动发起联系。
模式 3：Bidirectional messaging
代理在出现状态变化或需要帮助时通过 WebSockets 或 Server‑Sent Events（SSE）主动推送消息。你异步回复。该通道会记录历史，即使容器重启也不会丢失上下文。
# Agent sends you a message when it hits an edge case
import requests
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
# The agent encountered a decision point and escalates
requests.post(
f"{API}/channels/{CHANNEL_ID}/messages",
headers=headers,
json={
"content": (
"Found 3 candidates matching the criteria. "
"Should I contact all three, or just the top one? @human"
)
},
)你收到推送通知，回复后，代理继续工作流。
这就是人类协作的方式。奇怪的是，我们默认并没有以这种方式构建代理编排。
代理通信的架构要求
要为你的代理构建双向 pub/sub 层，需要解决一些非平凡的基础设施需求：
- 持久化对话历史 – 如果代理的唯一记忆只有 token 窗口，进程重启后历史会消失。你需要一个快速、可靠的数据存储（Postgres、Redis 等）来跟踪频道历史。
- 统一认证 – 代理使用 API 密钥；人类使用会话 Cookie 或 OAuth。大多数 Web 框架只擅长处理其中一种，而把另一种硬接上。一个合适的通信层应在同一聊天房间中把两者都视为一等公民。
- 实时传输 – 每隔几秒轮询 REST 接口既脆弱又浪费资源。WebSocket 或 SSE 能让交互感觉像真实的终端会话。
- 低摩擦 – 如果搭建通信层的难度比编写代理本身还大，开发者会回退到
print()语句。
最小可工作示例
下面是一个完整的循环——代理自行引导工作区，获取凭证，并开始发送消息。
步骤 1：引导
curl -X POST http://localhost:8080/api/v1/bootstrap \
-H "Content-Type: application/json" \
-d '{
"owner_email": "admin@local",
"owner_password": "changeme",
"agent_name": "research-bot",
"agent_description": "Handles research tasks"
}'响应
{
"api_key": "au_abc123...",
"channel_id": "ch_xyz789...",
"invite_url": "http://localhost:3001/invite/TOKEN"
}在浏览器中打开 invite_url。现在你和代理共享同一个频道。
步骤 2：代理发送消息
import json, requests, websocket
API = "http://localhost:8080/api/v1"
KEY = "au_abc123..."
CHANNEL = "ch_xyz789..."
headers = {
"Authorization": f"Bearer {KEY}",
"Content-Type": "application/json"
}
# REST: fire‑and‑forget
requests.post(
f"{API}/channels/{CHANNEL}/messages",
headers=headers,
json={"content": "Started research run. Will update with findings."}
)
# WebSocket: real‑time back‑and‑forth
ws = websocket.create_connection(f"ws://localhost:8080/ws?token={KEY}")
# Listen for human replies
while True:
msg = json.loads(ws.recv())
if msg["type"] == "new_message":
sender = msg["message"]["author_name"]
content = msg["message"]["content"]
if sender != "research-bot": # message is from a human
handle_human_reply(content)无需 SDK、无需框架锁定。如果你的代码能够进行 HTTP 调用或打开 WebSocket，这段代码即可运行——Python、Node、Go、Bash，随你选择。
Source: …
多代理模式
一旦拥有了消息层，多代理协同就自然产生。
- 编排器创建任务，将其路由给专职代理，并汇总结果。每个代理拥有自己的 API 密钥；编排器从所有通道读取并处理升级。
- 代理向共享通道发送消息；其他代理读取并作出响应。这种松耦合避免了成为瓶颈的单一中心调度器。
# Agent 1 posts result to shared channel
ws.send(json.dumps({
"type": "send_message",
"channel_id": SHARED_CHANNEL,
"content": "Research complete: found 3 key findings. See attached."
}))
# Agent 2 is subscribed to the same channel
msg = json.loads(ws.recv())
if msg["type"] == "new_message":
if "Research complete" in msg["message"]["content"]:
trigger_analysis_pipeline(msg["message"]["content"])通过一个简单的发布/订阅通道，代理可以协作、升级，并在不需要紧密编排的情况下从故障中恢复。
这只是发布/订阅
区别在于人类可以参与同一频道——你可以观察代理的协作，提问，重新指引他们。协作是可见的。
心态转变
停止把你的代理视为不透明的后端作业。
这种转变会彻底改变你所需的编排层。你不需要笨重的遥测仪表盘。你需要一个快速、透明的消息总线，让你能够准确看到代理的行为，并在它偏离时进行干预。
Agent United
我一直遇到这个摩擦点，于是我写了 Agent United。它是一个开源的自托管聊天平台，负责所有 WebSocket 的底层处理、身份验证分离以及持久化状态，让你只需专注于代理逻辑。只需一条命令即可启动：
docker-compose up如果你在构建需要人工监督的系统，可以在 GitHub 上获取代码，或在 docs.agentunited.ai/docs/agent‑guide 查看 API 示例。
结论
但归根结底，实现方式并不重要。模式才是关键。构建能够反馈的系统。
字数统计: ~1,050 字
目标: Dev.to（完整文章）+ 3 月 9 日 HN 链接帖子 + r/MachineLearning 模式章节仅