AI Agent 的提示词传输、流式输出与异常处理

本文从工程实践出发，介绍 AI Agent 系统中的三个关键环节：提示词组装与传输、模型响应的流式输出，以及异常分类、重试与降级处理。

目录

• 一、提示词传输：把意图与上下文可靠地交给模型
• 二、流式输出：让模型的生成过程实时可见
• 三、异常处理：构建稳健的 Agent 系统
• 四、总结

引言

近年来，随着 GPT、Claude、DeepSeek 等大语言模型（LLM）能力的提升，AI Agent 已经从概念逐步走向实际应用。无论是智能客服、代码助手，还是自动化办公与数据分析平台，AI Agent 都扮演着“连接用户意图与模型能力”的关键角色。

然而，一个真正可靠的 AI Agent，远不止“调用一次模型 API”那么简单。从用户输入到最终回答，中间涉及提示词的组装与传输、响应的流式返回，以及各种异常场景的容错处理。这三个部分构成了 Agent 系统的工程底座，直接影响系统的响应速度、用户体验与稳定性。

本文将从工程实践出发，系统梳理这三个环节的实现原理与常见方案，并给出可直接参考的代码示例。

一、提示词传输：把意图与上下文可靠地交给模型

1.1 提示词的本质与结构

提示词（Prompt）是 AI Agent 与模型交互的核心载体。在工程实现中，它通常不是一段简单的纯文本，而是一个结构化的消息序列（Messages）。

主流对话模型 API 通常采用消息列表作为输入，每条消息包含角色（role）与内容（content）。常见角色包括：

• system：设定 Agent 的人设、能力边界与行为规范；
• user：用户的实际输入；
• assistant：历史对话中模型的回复，用于维持多轮上下文；
• tool：工具调用结果，即 Agent 执行外部操作后返回给模型的信息。

一个完整的提示词，通常由“系统提示词 + 历史对话 + 当前用户输入 + 工具执行结果”组成。系统提示词的质量会影响 Agent 的行为是否符合预期；历史消息的顺序和角色则决定模型能否正确理解对话脉络。

1.2 传输协议的选择

提示词需要通过网络传输到模型服务端，常见协议包括 HTTP 与 gRPC。

• HTTP POST：最常见的方式。请求体通常为 JSON，包含模型名称、消息列表和采样参数，例如 temperature、top_p、max_tokens 等。它实现简单、调试方便，适合大多数业务场景。
• gRPC：基于 HTTP/2 和二进制序列化，在部分低延迟、高吞吐或双向流式通信场景中更具优势，通常用于内部服务调用。

一个典型的 HTTP 请求体如下：

{
  "model": "deepseek-chat",
  "messages": [
    {
      "role": "system",
      "content": "你是一个专业的技术助手。"
    },
    {
      "role": "user",
      "content": "请解释什么是流式输出？"
    }
  ],
  "temperature": 0.7,
  "stream": true
}

提示词在传输前需要进行序列化。对于包含大量历史上下文的场景，消息列表可能非常庞大。因此，应避免传输无关字段，并尽量减少重复信息。

1.3 上下文管理与 Token 控制

提示词长度受模型上下文窗口（Context Window）限制。当历史对话累积过多时，可能触发 Token 超限错误，例如 context_length_exceeded。因此，Agent 必须进行上下文管理：

• 截断策略：保留最近 N 轮对话，或者按照 Token 数量截断；
• 摘要策略：将较早的对话压缩为摘要，只保留关键信息；
• 记忆抽取：将长期信息保存到数据库或外部记忆系统中，使用时再按需检索；
• RAG 检索：从知识库中只召回与当前问题有关的内容，避免把全部资料放入上下文。

下面是一个简化的消息截断示例：

def trim_messages(messages, max_tokens=8000):
    """从后向前保留消息，直到预计 Token 数达到上限。"""
    kept = []
    total = 0

    for message in reversed(messages):
        cost = estimate_tokens(message["content"])
        if total + cost > max_tokens:
            break

        kept.append(message)
        total += cost

    return list(reversed(kept))

注意：生产环境应使用与目标模型匹配的 Tokenizer 计算 Token，而不是仅按字符数估算。同时，应优先保留系统提示词和当前用户消息。

合理的上下文管理不仅能避免请求超出模型限制，也能降低响应时间和调用成本。

二、流式输出：让模型的生成过程实时可见

2.1 为什么需要流式输出

大语言模型通常逐步生成 Token。如果采用一次性返回，用户可能需要等待数秒甚至数十秒才能看到完整结果。

流式输出（Streaming）允许服务端在模型生成内容的同时不断返回增量数据。前端收到数据后立即渲染，可以显著缩短用户感知到的等待时间。

2.2 基于 SSE 的流式实现

服务端到客户端的流式推送经常采用 SSE（Server-Sent Events）。SSE 基于 HTTP，响应的媒体类型通常设置为：

Content-Type: text/event-stream

服务端可以将模型的增量输出封装成 SSE 事件：

data: {"choices":[{"delta":{"content":"你"}}]}

data: {"choices":[{"delta":{"content":"好"}}]}

data: [DONE]

如果请求需要使用 POST 并携带 JSON 请求体，前端通常使用 fetch 和 ReadableStream 读取响应，而不是原生 EventSource：

const response = await fetch("/api/chat", {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
  },
  body: JSON.stringify({ messages }),
});

if (!response.ok || !response.body) {
  throw new Error(`请求失败：${response.status}`);
}

const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
let fullText = "";

while (true) {
  const { value, done } = await reader.read();
  if (done) break;

  buffer += decoder.decode(value, { stream: true });
  const lines = buffer.split("\n");
  buffer = lines.pop() ?? "";

  for (const line of lines) {
    if (!line.startsWith("data: ")) continue;

    const payload = line.slice(6);
    if (payload === "[DONE]") continue;

    const data = JSON.parse(payload);
    fullText += data.choices?.[0]?.delta?.content ?? "";
    render(fullText);
  }
}

这里额外使用了 buffer 保存不完整的数据行，避免网络分片恰好把一个 JSON 对象截断后导致解析失败。

2.3 基于 WebSocket 的流式实现

如果应用需要双向实时通信，或者需要让客户端主动发送“停止生成”“确认工具调用”等控制指令，可以考虑 WebSocket。

WebSocket 建立长连接后，服务端可以持续推送增量内容，客户端也能随时发送控制消息：

const socket = new WebSocket("wss://api.example.com/chat");
let fullText = "";

socket.onmessage = (event) => {
  const data = JSON.parse(event.data);

  if (data.done) {
    socket.close();
    return;
  }

  fullText += data.delta;
  render(fullText);
};

function stopGeneration() {
  socket.send(JSON.stringify({ type: "stop" }));
}

相比 SSE，WebSocket 的连接管理更加复杂，但它更适合交互复杂、需要双向控制的 Agent 应用。

2.4 增量渲染与状态管理

流式输出要求前端维护一个“累积文本”状态，每收到一个增量片段，就将其追加到已有内容中。

const [text, setText] = useState("");

function appendDelta(delta) {
  setText((previous) => previous + delta);
}

对于 Markdown 和代码高亮，还要处理“语法尚未闭合”的中间状态。例如，代码块还没有收到结尾标记时，可以在渲染阶段临时补全：

function safeRender(raw) {
  const fenceCount = (raw.match(/```/g) || []).length;
  return fenceCount % 2 === 1 ? `${raw}\n\`\`\`` : raw;
}

实际项目还应考虑：

• 合并高频增量，避免每个 Token 都触发一次完整渲染；
• 使用 AbortController 支持用户中止请求；
• 区分正常结束、用户取消、网络断开和服务端异常；
• 在组件销毁时关闭连接并释放资源；
• 流中断后清除“生成中”状态，并向用户提供重试入口。

三、异常处理：构建稳健的 Agent 系统

3.1 常见错误类型

AI Agent 在运行过程中可能遇到多种异常。先识别和分类错误，才能制定合适的处理策略。

一个重要原则是：只有具备瞬时性且重复执行安全的操作才适合自动重试。 对支付、发券、发送消息等有副作用的工具调用，应先保证幂等性。

3.2 超时与重试机制

网络请求必须设置合理的超时时间，避免调用方无限等待。对于网络抖动、部分 5xx 错误和限流，可以采用指数退避（Exponential Backoff）并加入随机抖动（Jitter），防止大量请求同时重试形成“惊群”。

import random
import time


def call_with_retry(func, max_retries=3, base_delay=1.0):
    for attempt in range(max_retries):
        try:
            return func()
        except RetryableError:
            if attempt == max_retries - 1:
                raise

            delay = base_delay * (2**attempt) + random.uniform(0, 0.5)
            time.sleep(delay)

对于限流错误，应优先尊重服务端返回的 Retry-After 响应头。此外，还可以采用并发限制、请求队列和熔断机制，防止上游故障扩散。

3.3 降级与兜底策略

当主模型不可用时，Agent 可以采用不同层级的降级策略：

• 模型降级：切换到能力和接口兼容的备用模型；
• 功能降级：临时关闭联网搜索等非核心能力，保留基础问答；
• 只读降级：禁用存在副作用的写入工具，仅允许查询；
• 缓存兜底：对适合缓存且不过时的高频问题返回历史结果；
• 人工接管：将无法自动处理的任务转交给人工。

def chat_with_fallback(prompt):
    for model in ["main-model", "backup-model"]:
        try:
            return call_model(model, prompt)
        except ServiceUnavailable:
            continue

    cached = cache.get(hash(prompt))
    return cached or "服务暂时不可用，请稍后再试。"

模型降级前应确认能力兼容性，例如上下文窗口、工具调用、结构化输出和多模态能力是否满足当前任务需求。

3.4 用户友好的错误反馈

错误处理不仅要技术正确，也要保证用户体验。面向用户时，应避免暴露堆栈信息、密钥、内部地址或晦涩的上游错误，而应返回清晰、可操作的提示。

{
  "request_id": "req_9f3a7c",
  "code": "UPSTREAM_BUSY",
  "message": "服务暂时繁忙，请稍后重试。",
  "retryable": true
}

服务端应记录用于排查问题的详细信息，例如：

• 请求 ID 与会话 ID；
• 模型与工具名称；
• 请求耗时与重试次数；
• 标准化错误码；
• 脱敏后的请求参数；
• 完整异常堆栈。

前端展示 request_id，可以帮助用户反馈问题，也方便开发人员快速定位对应日志。

3.5 流式响应中的异常收尾

流式接口一旦开始发送响应，通常无法再改写 HTTP 状态码。因此，建议在流中定义明确的事件类型：

event: delta
data: {"content":"正在生成的内容"}

event: error
data: {"code":"UPSTREAM_TIMEOUT","message":"生成超时，请重试。"}

event: done
data: {"finish_reason":"error"}

客户端需要保证无论收到 done、error，还是连接异常，都能够：

1. 停止加载动画；
2. 保留已经生成的内容；
3. 标记回答未完整生成；
4. 提供重新生成入口；
5. 释放网络连接和页面资源。

四、总结

提示词传输、流式输出与异常处理共同构成了 AI Agent 的工程基础：

• 提示词传输决定模型能否正确理解用户意图和上下文；
• 流式输出影响系统的感知速度与交互体验；
• 异常处理决定系统在真实复杂环境中的稳定性。

这三个环节并不是彼此孤立的。上下文管理会影响请求大小和传输效率；流式中断本身就是一种异常场景；异常处理又需要结合流式协议进行收尾。

只有把提示词、流式协议和异常治理作为一个完整链路统一设计，才能构建出可靠、可维护且具有良好用户体验的 AI Agent。