8.0
热点
AI SCORE
技术实践2026-08-14 02:00
AI Agent生产落地：从Prompt链式调用到基础设施架构
dev.to · AI#AI Agent#基础设施#架构
Editor brief · 编辑速览
企业级Agent系统面临三大致命问题：非确定性控制流、安全隔离边界缺失、不透明所有权，demo能跑的生产跑不了。
最初发表于 tamiz.pro。
对 AI 智能体的朴素想象是一个每轮对话都变得更聪明的聊天机器人。生产环境的现实要严苛得多：智能体跨服务协调、遵守安全边界、遵循标准化通信契约，并在无上限的请求量下存活。从提示词到基础设施的转变不是哲学层面的——这是一个将演示与已部署系统区分开来的硬核工程转型。
早期 AI 应用依赖提示链：由人类可读的叙述连接的一系列 LLM 调用。这种方式在需要五个智能体处理 10,000 个并发请求时就会失效——每个请求调用多个下游工具，延迟预算以秒计算，且合规需要审计追踪。
提示词层面的系统面临三个致命的规模化问题：
非确定性控制流 —— LLM 输出是概率性的，因此构建在提示词中的路由逻辑在边缘情况下会崩溃。
缺乏隔离边界 —— 被攻破或恶意的提示词可能泄露密钥、覆写状态或窃取工具输出。
所有权不透明 —— 当一个单一的巨大提示词编排所有操作时，调试需要重新阅读数十行指令文本，而不是检查结构化的状态转换。
替代方案是将智能体视为基础设施——具有明确契约、有界执行上下文和标准化智能体间通信的服务。
A2A（Agent-to-Agent，智能体间通信）指的是新兴的一类协议，允许智能体相互通信而无需中央编排器做每一个决策。可以把它想象成智能体的 RPC：结构化、类型化、版本化且可观测。
没有 A2A，智能体生态系统是这样的：
有了 A2A，智能体使用通用协议通信。每个智能体发布一个能力契约——一种机器可读的描述，说明它接受什么输入、产生什么输出，以及可能有什么副作用。其他智能体通过类型化接口发现和调用能力，而不是靠猜测 JSON 结构。
一个设计良好的 A2A 系统建立在三个原语之上：
有限基数的消息。每个智能体间消息都有明确的生命周期：发送、确认、完成或失败。与即发即忘的 HTTP 不同，A2A 消息携带序列号和关联 ID，使智能体能够重建对话历史。
类型化能力描述符。能力不是声明为 POST /agent/process，而是声明为：
{
"type": "tool",
"name": "database_query",
"version": "1.2.0",
"input_schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"required": ["query", "connection_id"],
"properties": {
"query": { "type": "string", "maxLength": 4096 },
"connection_id": { "type": "string", "pattern": "^conn-[a-f0-9]+$" }
}
},
"output_schema": {
"type": "object",
"properties": {
"rows": { "type": "array", "maxItems": 1000 },
"truncated": { "type": "boolean" }
}
},
"error_schemas": [
{ "code": "QUERY_TOO_LARGE", "message": "Query exceeds 4096 characters" },
{ "code": "CONNECTION_UNAVAILABLE", "message": "Connection pool exhausted" }
]
}
这个模式不是装饰性的。它在调用点启用编译时验证，在能力边界启用运行时强制，并在每次契约变更时自动生成测试。
信任边界调用上下文。每个 A2A 调用都携带一个调用上下文，回答以下问题：谁在调用？他们可以访问哪些工具？预算是多少？A2A 协议将其作为签名令牌或策略包嵌入，而不是作为因为你构建了客户端就信任的头部。
A2A 和沙箱是互补的。A2A 定义契约——可以问什么以及必须返回什么。沙箱定义执行边界——智能体在哪里以及如何运行其逻辑。它们共同创建一个系统，使智能体能够在不信任彼此内部实现的情况下协作。
智能体沙箱是一个受控环境，智能体代码在明确的资源限制、网络限制和输出过滤下运行。沙箱不是奢侈品——它是使多智能体系统安全运行的机制。
思考一下智能体在无沙箱环境下运行会出什么问题：
这些场景中的每一个都可以在基础设施层面解决，而不是寄希望于下一次提示词改进能 catch 到它。
生产级智能体沙箱实现三个层次：
计算沙箱。具有 CPU、内存和超时预算的隔离进程执行。工具作为子进程或 WebAssembly 模块运行，而不是在智能体进程内运行任意 Python。超出其配额的工具会被终止并报告为错误，而不是挂起编排器。
网络沙箱。智能体只能访问其被授权的网络。只读智能体不能调用 POST https://webhook.example.com/keys。出口流量由宿主机强制执行，而不是由智能体代码执行。入口流量根据入站工具响应的允许列表进行过滤。
数据沙箱。密钥、凭据和 PII 位于智能体执行上下文之外。工具收到令牌，而不是完整凭据。输出流在离开沙箱边界前扫描敏感模式。
以下是一个最小化但面向生产的沙箱封装，围绕工具调用：
import asyncio
import json
import time
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import Any, Optional
@dataclass
class SandboxConfig:
max_memory_mb: int = 256
timeout_seconds: float = 30.0
allowed_networks: list[str] | None = None
max_output_bytes: int = 65536
secret_prefixes: list[str] | None = None
class SandboxError(Exception):
pass
class SandboxTooMuchMemory(SandboxError):
pass
class SandboxTimeoutError(SandboxError):
pass
class SandboxNetworkBlocked(SandboxError):
pass
@asynccontextmanager
async def run_tool_in_sandbox(
tool_code: str,
arguments: dict[str, Any],
config: SandboxConfig,
):
start = time.monotonic()
# 1. 将调用编码为自包含的脚本负载
payload = json.dumps({
"args": arguments,
"limits": {
"memory_mb": config.max_memory_mb,
"timeout_s": config.timeout_seconds,
"max_output_bytes": config.max_output_bytes,
},
"allowed_networks": config.allowed_networks,
"secret_prefixes": config.secret_prefixes or [],
}).encode()
# 2. 生成一个沙箱化子进程。在实践中这会使用
# gVisor、Firecracker 或 WASI——这里我们展示的是契约。
proc = await asyncio.create_subprocess_exec(
"sandbox-runner",
"--tool-code", "-",
"--payload", "-",
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
try:
stdout, stderr = await asyncio.wait_for(
proc.communicate(input=payload),
timeout=config.timeout_seconds,
)
except asyncio.TimeoutError:
proc.kill()
raise SandboxTimeoutError(
f"Tool exceeded {config.timeout_seconds}s budget"
)
if proc.returncode != 0:
error = stderr.decode(errors="replace")
if "memory" in error.lower():
raise SandboxTooMuchMemory(error)
if "network" in error.lower():
raise SandboxNetworkBlocked(error)
raise SandboxError(f"Tool exited {proc.returncode}: {error}")
result = json.loads(stdout.decode())
# 3. 执行后输出过滤
if result.get("output_bytes", 0) > config.max_output_bytes:
raise SandboxError("Output exceeds max_output_bytes")
yield result
注意这段代码里没有的东西：认证、授权、日志记录或重试逻辑。那些属于编排层。沙箱只回答一个问题：这个工具是否在其声明的预算内完成，且输出在结构上是否有效？
最古老的智能体架构是中央编排器：一个智能体读取提示词，决定调用哪些工具，调用它们，然后返回结果。这种模式在两种条件下会崩溃——高并发和异构工具所有权。
想象一下，十个团队各自维护一套工具。没有团队愿意暴露一个通用的 REST 端点，让其他团队的智能体可以用任意载荷调用它。他们需要：
中央编排器无法在不成为瓶颈和单点故障的情况下满足这些需求。
替代方案是能力注册中心——一个智能体发布能力和发现能力的服务中心。每个智能体注册自己的能力。其他智能体查询注册中心以找到匹配其需求的能力。当某个智能体调用能力时，注册中心解析目标并注入调用上下文。
# capability-registry.example.yaml
capabilities:
- agent_id: payments-agent
version: 2.1.0
capabilities:
- name: charge
input: ChargeRequest
output: ChargeResult
error_schemas: [InsufficientFunds, CardDeclined, NetworkTimeout]
rate_limit:
calls_per_minute: 100
burst: 20
trust_policy:
required_roles: [payments-service]
allowed_origins:
- order-agent
- refund-agent
- agent_id: research-agent
version: 1.4.0
capabilities:
- name: search
input: SearchRequest
output: SearchResultList
error_schemas: [RateLimited, QueryTooLong]
rate_limit:
calls_per_minute: 30
burst: 5
trust_policy:
required_roles: [research-service]
allowed_origins: [user-agent]
这个文件不是配置转储，而是一份机器可读的契约。注册中心在运行时通过检查调用方身份、应用速率限制，并在无效调用到达智能体之前将其短路来强制执行契约。
A2A 消息携带 capability_id 字段而非地址。注册中心使用信任策略和速率限制状态将该标识符解析为实际调用目标。消费者永远不需要知道——也不应该关心——哪个部署宿主提供了该能力。这种解耦使得智能体可以水平扩展而不需要脆弱的路由表。
构建可信智能体最难的部分是控制状态。一个修改共享变量、发送侧通道消息或静默重试失败的智能体，在测试中看起来正确，在生产中却会灾难性失败。
每个生产级智能体都应该将其状态公开为带有显式转换的有限状态机。考虑一个处理用户请求的任务智能体：
┌──────────┐
user_request ──▶ │ PENDING │
└────┬─────┘
│ plan_generated
┌────▼─────┐
│ PLANNING │
└────┬─────┘
│ plan_approved
┌────▼─────┐
┌─────│ EXECUTING│─────┐
│ └────┬─────┘ │
│ │ tool_failed tool_completed
┌─────▼─────┐ ┌─▼────────┐ ┌──▼──────────┐
│ RETRYING │ │COMPLETED │ │FAILED_MAX │
└─────┬─────┘ └──────────┘ └─────────────┘
│
│ retry_budget_exhausted
└───────────────────────▶ FAILED_MAX
这个图不是装饰。它驱动四个工程决策：
序列化 —— 转换是唯一的写入路径。如果每个处理程序都应用于不可变快照并原子性地写回，并发 bug 就不可能存在。
可观测性 —— 状态转换即事件。将它们发送到持久日志。重放是免费的。
恢复 —— 重启时，智能体从事件日志重建状态，而不是信任易失性内存。
测试 —— 每个转换都是一个单元测试。你验证的是行为，而非运气。
一个以不同结果两次调用同一工具的智能体，是一个在自身状态上撒谎的智能体。每个工具处理器必须是幂等的，或者明确是非幂等的并带有补偿逻辑。
interface ToolHandler {
id: string;
execute(ctx: InvocationContext, params: unknown): Promise<ToolResult>;
/** True if re-invoking with the same params must return the same result */
idempotent: boolean;
/** Optional cancellation hook */
cancel?(ctx: InvocationContext): Promise<void>;
}
当一个工具是非幂等的——写入数据库、向 API 发送 POST——智能体必须为每次调用关联一个唯一的调用 ID，并在失败后重新调用之前检查是否已先前完成。没有这个，重试风暴会导致成本翻倍和状态损坏。
智能体默认是不透明的。一次糟糕的迭代，你无法判断故障在于提示词、工具、LLM 还是路由逻辑。可观测性不是附加组件——它是使调试成为可能的透镜。
生产级智能体系统必须一致地发出四个信号：
结构化调用日志。每个 A2A 调用都记录关联 ID、调用方、目标、能力、输入哈希、开始时间戳、结束时间戳和结果。输入被哈希用于去重，但从不以明文形式记录。
工具执行追踪。对于每个工具调用，记录：工具 ID、沙箱边界信息、资源使用、网络出口和输出大小。这让你能够检测沙箱逃逸和资源滥用。
LLM 成本和延迟分解。追踪每个模型调用消耗的 token 数、延迟和每个智能体的成本。当单独看起来便宜的智能体，在合计重试、回退调用和长上下文窗口时会变得昂贵。
状态转换审计日志。每个 FSM 转换都与触发它的事件、触发者和结果状态一起记录。这个日志是事后分析的真实数据来源。
{
"trace_id": "7f3a9b2c-4e1d-4f8b-b5a6-9c8d7e6f5a4b",
"span_id": "01a2b3c4d5e6f7a8",
"timestamp_ms": 1718496234567,
"caller_agent_id": "order-agent",
"target_agent_id": "payments-agent",
"capability_id": "charge",
"capability_version": "2.1.0",
"input_hash": "sha256:e3b0c44298fc1c149afbf4c8996fb924",
"start_ms": 1718496234567,
"end_ms": 1718496234789,
"duration_ms": 222,
"outcome": "success",
"sandbox": {
"cpu_ms": 45,
"memory_peak_mb": 87,
"network_egress_bytes": 1024,
"timeout_budget_ms": 30000,
"timeout_used_ms": 0
},
"llm_usage": {
"model": "claude-sonnet-4",
"input_tokens": 1240,
"output_tokens": 89,
"cache_read_tokens": 320
},
"cost_usd": 0.00142,
"state_transition": {
"from": "EXECUTING",
"to": "EXECUTING",
"event": "tool_completed"
},
"retry_count": 0,
"idempotency_key": "charge-7f3a9b2c-order-88291"
}
这个日志告诉你事后分析需要的一切，无需访问生产内存或重放对话。
单智能体系统的架构模式不能干净地转移到多智能体系统。以下是真正有效的模式。
当一个任务分解为独立子任务时，扇出到多个智能体并聚合结果。聚合器等待法定人数或最佳响应策略：
这种模式随着智能体数量线性降低延迟，并提供天然的容错能力。
当任务有顺序依赖时，用显式背压管道化智能体。每个智能体持有一个有界队列。当队列满时，上游智能体阻塞，而不是丢弃工作或溢出到磁盘。这使系统在负载下保持稳定。
# High-level backpressure-controlled pipeline
async def run_pipeline(task: Task, agents: list[Agent]) -> Result:
queue = asyncio.Queue(maxsize=100) # bounded
async def producer():
await queue.put(task)
async def consumer(agent: Agent):
while True:
item = await queue.get() # blocks when empty
try:
result = await agent.process(item)
await queue.put(result) # blocks when full
finally:
queue.task_done()
await asyncio.gather(
producer(),
*(consumer(agent) for agent in agents),
)
maxsize=100 的队列大小并非随意设定。它是一个内存边界，用于防止下游代理减速时队列无限增长。
代理应声明降级层次结构。如果主代理不可用或返回错误，系统会尝试下一个能力版本，或者使用功能更少、更简单的代理：
primary-agent:1.0 ──failure──▶ primary-agent:1.0-fallback ──failure──▶ read-only-agent:2.3
每个降级代理都有文档化的能力削减说明。编排器向调用方公开这一信息，以便 UI 可以适应——显示降级响应而不是硬报错。
并非所有调用都值得使用最具能力的模型。按复杂度进行路由：
成本感知路由器会检查调用上下文——输入大小、所需推理深度和输出结构复杂度——并选择满足 SLA 的最小模型。在数百万次调用中，节省会不断累积。
代理系统中的安全性不在于边界防御。它在于假设每个边界都可能被攻破，并据此进行设计。
假设每个代理都可能被攻破。将每个 A2A 调用视为可能被欺骗的调用。强制执行以下措施：
每个能力声明其所需的最小资源。沙盒强制执行此规则。仅从数据库读取的能力永远不应该能够写入。这在能力注册级别强制执行，而非在代码中。
像对待不可信网络一样对待用户输入：它是对抗性的。在沙盒边界验证和清理输入。使用结构化解析（JSON schema）而非自然语言过滤。自然语言过滤在对抗性提示面前会失败；schema 验证则不会。
代理绝不能持有长期密钥。使用由密钥提供者颁发的短期令牌。在每次调用时或按固定 schedule 轮换，以先到者为准。将密钥存储在保险库中，而非环境变量或配置文件。
提示工程指标——token 数量、每次调用延迟、每个提示的成功率——是必要的但不够充分。生产环境代理系统需要基础设施指标：
问：如果只有一个代理，我需要 A2A 协议吗？
不需要。A2A 的价值在于当你有多个代理需要协调、共享能力或独立演进时。对于具有固定工具集的单一代理，单体编排器更简单且足够。当协调复杂度超过单个编排器所能管理的范围时，再引入 A2A。
问：我可以从中心化编排器开始，然后迁移到 A2A 吗？
可以，但请将编排器设计为无状态的，并将能力设计为可独立寻址的。如果从第一天起每个工具调用都包装在能力合约中，迁移只是配置变更。如果工具调用在编排器中是硬编码的，迁移就是重写。
问：如果我要构建代理状态机，如何处理非确定性？
非确定性存在于 LLM 层，而非代理层。LLM 返回概率性计划；代理基于该计划执行确定性状态机。记录计划，根据 schema 验证计划，将其作为状态转换应用，并将任何偏差视为错误。这种关注点分离使代理即使在输入不确定时也保持可预测。
从提示到基础设施的转变并非要放弃提示工程——而是要认识到提示是更大系统中的一个组件。A2A 协议、代理沙盒、能力注册中心和状态机设计构成了骨架。提示是神经系统。把骨架构建好，神经系统就有稳定的东西可以控制。