ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

AI输出即他人输入:构建可靠的模型输出流水线

AI输出即他人输入:构建可靠的模型输出流水线

1. 这篇文章真正要解决的问题

最近团队里一个同学向我吐槽:他们做的一个 AI 客服系统,明明单轮对话效果不错,但一接到真实业务链条里就频繁出问题。不是回复质量变差,而是下游系统根本没法吃下 AI 的输出——要么 JSON 格式带上了 Markdown 代码块标记,要么字段多了几个值,要么模型把“N/A”写成了“不适用”,结果下游数据分析程序直接把整条记录丢弃。

这其实是一个被很多人低估的问题:我们在调试 AI 应用时,习惯性把目光放在“模型生成结果好不好”上,却很少问一句“这个结果接下来要流向哪里、被谁消费、消费时是否安全”。

文章标题“Your AI Output Is Someone Else's Input”,本质上是一句工程箴言。它想提醒的是:在一个真实的软件系统里,大模型绝不是孤立的终点。它的输出,往往只是另一次请求的输入、另一条流水线的中间数据、另一个微服务的载荷,甚至是无人值守的自动化任务的唯一依据。如果输出不稳定、不规范、不可解释,连锁故障会顺着数据链路一路传导到业务层。

这篇文章会从四个方面展开:

  • 先搞清楚 AI 输出在哪些典型场景里会成为“别人的输入”。
  • 再看为什么模型输出天然不适合直接当数据接口来用。
  • 然后给出工程化接入的思路:约束、校验、清洗、兜底、观测。
  • 最后用一个可运行的最小示例把整个流程串起来。

无论你是正在做 RAG 应用、Agent 编排,还是只写了一个简单的调用 OpenAI 接口的工具脚本,这篇文章都值得看完。因为**“输出即输入”是一个架构问题,不是一个提示词问题**。

2. AI 输出成为他人输入的典型场景

先别急着聊代码。要理解这个问题的严重性,得先看清现实里 AI 输出会流向多少种消费者。

1.1 下游程序直接消费模型输出

这是最常见的一类。典型形态包括:

  • 后端服务调用大模型,要求模型返回 JSON,然后直接通过json.loads()解析。
  • 自动化脚本让模型生成 SQL,然后程序把 SQL 发给数据库执行。
  • 数据管道让模型输出结构化字段,例如判断一段文本的情感是“positive / negative / neutral”,然后直接进入数仓加工。
  • 前端拿到模型生成的 Markdown,直接渲染为用户可见的页面内容。

在这些场景里,大模型本质上是数据生产方。下游程序则成了数据消费方。问题在于,模型本身并不是为这个契约设计的——它天生是一个“文本续写器”,而不是一个“严格类型的数据接口”。

1.2 另一个 AI 模型消费前一个模型的输出

在多智能体(Multi-Agent)架构里,这种“输出即输入”的依赖关系被进一步放大。

比如你设计了一个“翻译 + 审校”的双 Agent 流程:

Agent A(翻译)输出译文文本 ↓ Agent B(审校)消费 A 的输出,判断译文质量 ↓ 返回最终文本

听起来没什么问题,但真正的工程风险藏在细节里:如果 Agent A 在输出末尾多了一句“如果你需要更贴切的表达,我可以继续优化”,Agent B 可能会把这句话也当成译文内容去审校。再比如 Agent A 输出了一条 JSON,但在 JSON 外面包了一层 ```json 代码块标记,Agent B 按原始字符串解析就会失败。

这类问题在单模型调用场景里还能靠人眼发现,但在自动化链路里,错误会在无人干预的情况下被逐级放大。

1.3 人类用户直接面对 AI 输出

还有一种容易被当成“非工程问题”的场景:AI 输出最终展示给终端用户。

很多团队觉得,既然输出是给人看的,那格式乱一点、内容跳跃一点,都只是体验问题。但在企业级系统里,用户看到的内容往往要触发后续操作。一个典型的例子:

  • AI 生成了财务摘要,用户复制到 Excel 里做下一步分析。
  • AI 生成了工单处理建议,客服据此填写工单系统。
  • AI 生成了代码补丁,开发人员直接复制进代码仓库。

如果输出里带了无意义的编号、口语化表述、或者是“大约”“可能”这类模糊词,下游的“人工处理流程”就会被严重拖慢。更糟糕的是,模糊输出会让人类产生错误的信任判断——看起来言之凿凿的内容,其实可能是模型虚构的。

1.4 数据闭环里的“输入污染”

最后一种场景相对隐蔽:模型输出被保存下来,成为后续迭代的训练数据或评测样本。

在 RAG 应用中,很多人会把“问题 - 检索片段 - 模型输出”整条链路记录下来,用来做效果分析。但如果模型输出本身含有幻觉信息、格式错乱、半截句子,这些脏数据就会进入评测集,反过来污染后续的模型优化方向。

这就是“Your AI Output Is Someone Else's Input”最深刻的一层含义:**在数据闭环里,今天的输出就是明天的训练输入。**一旦源头不干净,整个飞轮会越转越偏。

3. 为什么模型输出天然不适合直接当接口数据

要解决“输出即输入”的工程问题,首先得承认一个事实:大模型的输出和传统 API 的数据格式,在本质上存在结构性矛盾。

2.1 概率生成 vs 严格契约

传统 API 定义了一个严格的请求/响应契约:

{ "code": 0, "data": { "name": "张三", "age": 18 } }

调用方可以放心地按字段解析,因为服务端有类型检查、校验逻辑和错误处理机制。数据要么是合法的 JSON,要么返回错误码——不存在“有一半合法 JSON”这种状态。

大模型输出则不一样。你让模型返回一个 JSON,它大概率会返回一个 JSON,但它不保证:

  • 字段名完全一致。
  • 字段值的数据类型正确。
  • 不混入多余字段。
  • 不出现拼写错误。
  • 不包裹 Markdown 标记。

这不是模型“笨”,而是生成式模型的本质决定了它做的是概率化的采样,不是结构化的序列化。你拿到的是一个“看起来像 JSON 的文本”,需要自己去兜底。

2.2 重复性缺失:同一输入,输出不稳定

以前的函数式编程也好,状态机也好,都强调同一输入产生同一输出。大模型恰恰相反——即使温度设为 0,输出也有一定随机性。

这就带来一个严重后果:下游链路很难用传统 API 的“一次性校验通过,永久放心使用”思路。你必须在每次消费输出时做防御性检查。

2.3 模型遵循指令的能力受上下文影响

很多开发者会想:我把字段要求写清楚不就行了?

确实可以缓解一部分问题,但要注意:模型的指令遵循能力会受上下文长度、提示词措辞、样例质量、甚至 token 位置的影响。一个在测试时表现良好的 prompt,换了一批真实业务数据后,可能就开始“自由发挥”。

所以,提示词约束只能降低错误概率,不能消除错误概率。工程上必须假设输出永远可能出错。

2.4 一个容易混淆的概念:不是所有输出问题都该靠 prompt 解决

这里要特别提醒新手:遇到输出格式问题,第一反应往往是“改 prompt”。但实践中,更稳妥的思路是分层防御:

第 1 层:prompt 约束,让模型尽量按指定格式输出。 第 2 层:输出解析,尽量减小格式自由度带来的影响。 第 3 层:数据校验,对解析结果做类型、枚举、正则校验。 第 4 层:兜底策略,校验失败时选择重试、默认值或人工介入。 第 5 层:过程观测,记录每次输出的原始值和校验结果。

只改 prompt,等于把自己全部的安全感压在了一个概率模型上。这在某些 Demo 场景没问题,但在生产链路里风险太集中。

4. 工程化接入的思路与核心手段

理解了问题本质后,我们来看具体手段。下面这些方法,我在实际项目里验证过,能显著降低下游消费侧的事故率。

3.1 尽量使用支持结构化输出的模型能力

如果你用的模型平台提供了 JSON Mode / Structured Output 之类的功能,务必优先使用。

例如在 OpenAI 的 API 中,可以定义响应格式:

# 文件路径:examples/structured_output_demo.py from openai import OpenAI client = OpenAI(api_key="your-api-key") response = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": "你是客服工单分类助手。"}, {"role": "user", "content": "帮我把这句话分类:我的订单已经三天没发货了。"} ], response_format={"type": "json_object"}, ) print(response.choices[0].message.content)

这种能力背后是模型服务方在采样和解析层面做了约束,比单纯在 prompt 里写“请返回 JSON”要可靠得多。但要注意:即便有结构化输出,下游校验依然不能省。因为结构化输出通常只约束了“合法 JSON”这一层,字段语义是否正确,服务方不会替你检查。

3.2 为输出定义明确的 Schema

不止是让模型返回 JSON,你要像设计数据库表一样设计输出的 Schema。

举个例子,假设你希望模型判断工单是不是需要升级处理:

SCHEMA_PROMPT = """ 请从以下三个维度分析用户反馈: 1. severity:问题严重程度,值为 low、medium、high 之一。 2. category:问题分类,从 硬件故障、网络问题、账号问题、其他 四选一。 3. action_required:是否需要人工介入,值为 true 或 false。 只输出 JSON,不要输出多余内容。格式如下: { "severity": "low", "category": "其他", "action_required": false } """

Schema 的价值不只是让模型知道“要长什么样”,更重要的是让下游解析逻辑有了明确预期。后面校验时,你可以针对每一个字段写断言。

3.3 解析与清洗,而不是无脑 loads

如果模型输出里混入了 ```json 标记、说明文字、逗号结尾等常见问题,你需要在解析层做清洗。下面是一个最小处理函数:

# 文件路径:utils/parse_and_clean.py import json import re def extract_json(text: str) -> dict: """ 从模型输出中提取 JSON 对象。 兼容带 ```json 代码块标记、说明文字包裹等情况。 """ if not text: raise ValueError("empty model output") # 尝试直接解析 try: return json.loads(text) except json.JSONDecodeError: pass # 去掉代码块标记 pattern = r"```(?:json)?\s*([\s\S]*?)```" match = re.search(pattern, text) if match: try: return json.loads(match.group(1).strip()) except json.JSONDecodeError: pass # 去掉包裹的说明文字,找到第一个 { 和最后一个 } start = text.find("{") end = text.rfind("}") if start != -1 and end != -1 and end > start: try: return json.loads(text[start:end + 1].strip()) except json.JSONDecodeError: pass raise ValueError(f"cannot extract JSON from: {text}")

注意:这个函数可以处理常见的“模型不守规矩”情况,但它不是万能的。如果模型输出本身就不包含完整 JSON,还是要走兜底策略。

3.4 严格校验,而不是只靠“能解析”

解析出字典后,下一步是校验字段。这里推荐用一个很小的校验函数,而不是把逻辑散落到业务代码里:

# 文件路径:validators/ticket_validator.py from typing import Any ALLOWED_SEVERITY = {"low", "medium", "high"} ALLOWED_CATEGORY = {"硬件故障", "网络问题", "账号问题", "其他"} def validate_ticket_analysis(data: dict[str, Any]) -> list[str]: """ 返回错误信息列表。如果列表为空,说明校验通过。 """ errors = [] if data.get("severity") not in ALLOWED_SEVERITY: errors.append("severity 字段不合法") if data.get("category") not in ALLOWED_CATEGORY: errors.append("category 字段不合法") if not isinstance(data.get("action_required"), bool): errors.append("action_required 字段必须是布尔值") return errors

这一步的价值在于:**把模型输出的“概率可靠性”转化为“确定性的程序判断”。**只要校验逻辑写对了,非法输出就一定会在进入业务逻辑之前被拦截。

3.5 兜底策略:重试、默认值、人工介入

校验失败时,不能直接把异常抛给用户,也不能静默吞掉。应该设计分等级的兜底策略:

失败类型处理方式
解析失败,但重试成本低使用带错误提示的新 prompt 重试 1 到 2 次
字段值不合法如果是低风险场景,使用安全默认值并记录日志
高危场景(支付、工单派发、权限相关)转入人工处理队列,不允许使用默认值
连续重试仍失败触发告警,通知开发人员

这里要特别强调一个原则:**高风险场景宁可挂起,也不要猜。**比如模型输出一个“是否允许退款”的字段,你校验失败后默认返回 false,可能直接导致客服无法处理用户诉求。

3.6 观测与日志,让问题可追踪

最后,所有解析、清洗、校验环节都要留日志。每条调用至少记录四个信息:

  • 模型原始输出全文。
  • 清洗后解析内容。
  • 校验结果(通过或哪些字段失败)。
  • 最终返回给下游的数据。

没有日志,“输出即输入”的链路一旦出问题,排查难度会非常大——因为你根本不知道问题出在模型、解析、校验哪一层。

5. 完整代码示例:构建一个可靠的“输出作为输入”流水线

前面讲了理论和拆解,这一节我们直接落地。下面用一个完整的 Python 示例,演示如何安全地让 AI 输出成为下游处理流程的输入。

4.1 场景定义

假设我们做一个工单处理组件:

  • 用户输入一段问题描述。
  • 大模型分析出严重程度、分类、是否需人工介入。
  • 下游系统根据这三个字段自动分派工单。

因为分派结果会影响实际工单流转,所以这属于“中等风险”场景。我们的流水线必须保证:非法输出不能被静默消费。

4.2 完整代码文件

# 文件路径:pipeline/ticket_pipeline.py import json import logging from typing import Any from openai import OpenAI from utils.parse_and_clean import extract_json from validators.ticket_validator import validate_ticket_analysis logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") logger = logging.getLogger(__name__) SYSTEM_PROMPT = """ 你是客服系统的工单分析助手。请分析用户的反馈,只输出 JSON, 格式如下,不要输出任何额外内容: { "severity": "low", "category": "网络问题", "action_required": false } 规则: - severity 只能取 low、medium、high。 - category 只能取 硬件故障、网络问题、账号问题、其他。 - action_required 只能是 true 或 false。 """ USER_PROMPT_TEMPLATE = """ 用户反馈内容: {user_input} """ # 带重试的调用函数 def analyze_with_retry( client: OpenAI, user_input: str, max_retries: int = 2, ) -> dict[str, Any]: """ 调用模型,最多重试 max_retries 次。每次重试时把前一次的报错信息带回 prompt。 """ last_error = "" for attempt in range(1, max_retries + 2): try: user_content = USER_PROMPT_TEMPLATE.format(user_input=user_input) if last_error: user_content += f"\n\n上一次输出不符合要求,错误信息:{last_error}\n请重新生成。" messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_content}, ] response = client.chat.completions.create( model="gpt-4o-mini", messages=messages, response_format={"type": "json_object"}, ) raw_content = response.choices[0].message.content parsed_data = extract_json(raw_content) errors = validate_ticket_analysis(parsed_data) if not errors: logger.info("attempt %s success, parsed_data=%s", attempt, parsed_data) return parsed_data last_error = "; ".join(errors) logger.warning("attempt %s validation failed: %s", attempt, last_error) except Exception as exc: # 只捕获异常,模型可能偶发连接问题 last_error = f"exception: {exc}" logger.warning("attempt %s exception: %s", attempt, exc) # 重试耗尽,抛给上层处理 raise RuntimeError(f"model analysis failed after {max_retries} retries, last_error={last_error}") # 下游消费示例 def dispatch_ticket(analysis: dict[str, Any]) -> None: """ 模拟下游工单分派逻辑。这里假定 analysis 已经通过校验。 """ severity = analysis["severity"] category = analysis["category"] action_required = analysis["action_required"] print(f"[dispatch] severity={severity}, category={category}, " f"action_required={action_required}") if action_required: print("请输入人工处理队列") else: print("自动处理,不转人工") if __name__ == "__main__": # 注意:实际生产中 API Key 应从环境变量读取,不要硬编码 import os api_key = os.environ.get("OPENAI_API_KEY", "your-api-key") openai_client = OpenAI(api_key=api_key) test_input = "你好,我的路由器一直掉线,已经重启两次了,但还是无法上网。" try: result = analyze_with_retry(openai_client, test_input) dispatch_ticket(result) except RuntimeError as exc: # 真实场景中这里应告警并转人工 logger.error("pipeline failed: %s", exc)

4.3 代码逻辑解释

  • analyze_with_retry封装了“调用模型 → 解析 → 校验 → 重试”的完整闭环。
  • 每次失败,把错误信息注入到 prompt 里,让模型有机会自我纠正。
  • 重试次数耗尽后,直接抛出RuntimeError,由上层决定转人工还是走其他兜底。
  • dispatch_ticket消费已经通过校验的数据,不关心数据是从模型来的还是从数据库来的。

4.4 如何运行

首先安装依赖:

pip install openai

设置环境变量:

export OPENAI_API_KEY="你的 API Key"

运行流水线:

python -m pipeline.ticket_pipeline

正常情况下的输出类似:

INFO:attempt 1 success, parsed_data={'severity': 'medium', 'category': '网络问题', 'action_required': True} [dispatch] severity=medium, category=网络问题, action_required=True 请输入人工处理队列

4.5 如何验证可靠性

跑通之后,你可以刻意构造几个刁钻输入来验证防护层:

# 测试 1:带代码块标记的返回 raw_text = "```json\n{\"severity\": \"high\", \"category\": \"硬件故障\", \"action_required\": true}\n```" print(extract_json(raw_text)) # 测试 2:带说明文字的返回 raw_text2 = "好的,这是分析结果:{\"severity\": \"low\", \"category\": \"账号问题\", \"action_required\": false} 请查收。" print(extract_json(raw_text2)) # 测试 3:字段值非法的返回 invalid_data = {"severity": "urgent", "category": "物流问题", "action_required": "yes"} print(validate_ticket_analysis(invalid_data))

预期输出:

{'severity': 'high', 'category': '硬件故障', 'action_required': True} {'severity': 'low', 'category': '账号问题', 'action_required': False} ['severity 字段不合法', 'category 字段不合法', 'action_required 字段必须是布尔值']

注意,第三个测试里我们用了“urgent”和“物流问题”,它们都是模型可能给出的“看起来合理但超出 Schema”的值。这正说明校验必不可少。

6. 常见问题与排查思路

在真实项目中,很多人第一次踩坑往往是在下面这些位置。整理成一张排查表,方便你按图索骥。

问题现象可能原因排查方式解决方案
解析日志报 JSONDecodeError模型输出带 Markdown 标记或说明文字打印原始输出,观察包裹结构使用extract_json做清洗,而不是直接json.loads
字段值偶尔不在预期枚举内模型对中文枚举理解偏差查看失败样本,统计是哪个字段在 prompt 中增加枚举解释和示例,同时在校验层拦截
重试后仍失败,链路直接中断prompt 约束太弱或任务本身不匹配查看重试时带回的错误信息考虑更换模型或拆分为更小的子任务
下游出现脏数据但日志正常校验规则写得不够全,例如只校验了类型没校验范围复查校验函数覆盖的字段对每个字段写具体断言,覆盖枚举、边界值、空值
模型输出被截断达到 max_tokens 限制查看 API 返回的 finish_reason提高 max_tokens,或要求模型输出更紧凑的结构
调用偶发超时/连接错误网络不稳定或服务限流查看异常类型和响应状态码在调用层增加指数退避重试,时间允许时接入消息队列异步化

这里想单独展开两点。

6.1 不要把所有错误都抛给模型重试

有的同学会让模型无限重试,理由是“总会成功的”。实际经验是:**重试次数过多,成本会失控,而且模型可能在第三次重试后反而开始编造字段值来迎合你。**更合理的策略是限制重试次数(1 到 2 次),失败后降级到人工处理或默认值,核心链路不能被模型调用阻塞。

6.2 日志里的原始输出是排查的关键

很多项目只记录“模型返回的解析结果”,不记录“原始输出”。这在调试时极其痛苦——因为解析结果已经是清洗后的数据,你根本不知道原始输出是什么。建议在所有调用点统一埋点,把原始输出、解析结果、校验结果、最终消费结果全部记录下来。这会极大缩短故障排查时间。

7. 生产环境中的最佳实践与工程建议

如果你准备把“AI 输出即他人输入”的链路放进生产环境,以下建议值得逐条对照。

7.1 统一输出契约,而不是各写各的 prompt

团队里多个模块都在调用大模型时,很容易出现各自定义格式的情况:A 模块用中文枚举,B 模块用英文枚举,C 模块直接用自由文本。这会让下游消费方非常痛苦。

建议在团队内约定一个输出契约管理方式:

# 文件路径:contracts/ticket_analysis.py from dataclasses import dataclass, asdict @dataclass class TicketAnalysis: severity: str category: str action_required: bool def to_dict(self) -> dict: return asdict(self) @classmethod def from_validated_dict(cls, data: dict) -> "TicketAnalysis": # 调用方保证 data 已经通过校验 return cls( severity=data["severity"], category=data["category"], action_required=data["action_required"], )

这样,模型输出经过解析校验后,映射到一个数据类,下游永远只依赖这个契约类型,不直接依赖模型的原始字符串。

7.2 区分“展示型输出”和“数据型输出”

并不是所有模型输出都必须走强校验。需要区分两类:

  • 展示型输出:最终给用户看的摘要、推荐语、文章。重点是流畅、自然、有信息量;格式问题相对宽容,但要注意事实性错误。
  • 数据型输出:要进入程序逻辑、数据库、下游系统的结构化字段。重点是准确、合法、符合预期;必须走严格解析和校验。

很多人把这两类混淆,于是要么在展示型输出上过度约束,导致文案死板;要么在数据型输出上放任自由,导致下游频繁报错。

7.3 为高风险场景设计人工介入闸门

在涉及资金、权限、法律、医疗等领域,AI 输出一定不能直接驱动高影响动作。更稳妥的做法是:

模型输出 → 解析校验 → 生成建议 → 等待人工确认 → 才执行动作

这个“人工确认”不是一个可选项,而是一个安全闸门。AI 的速度优势依然体现在“生成建议草稿”这一步,而不是“替代最终决策”。

7.4 对输出做版本兼容

模型升级、prompt 调整、新模型替换,都会导致输出分布发生变化。这意味着在同一套下游代码里,不能假设上游永远不变。建议在输出契约里增加schema_version字段,下游根据版本号决定如何解析,避免一次模型升级就全线崩盘。

{ "schema_version": "1.0", "severity": "low", "category": "其他", "action_required": false }

7.5 观测不是事后补救,而是上线前的必要条件

至少要做到:

  • 每次调用的延迟、token 消耗。
  • 解析失败率、校验失败率。
  • 各字段的取值分布,尤其是非法值出现频率。
  • 兜底策略触发次数和原因。

这些指标能帮你尽早发现“prompt 退化”或“模型行为漂移”。没有观测,链路上任何一环变差,你都只能在用户投诉时才发现。

8. 从输出为中心到链路为中心的思维转变

最后想聊一个方法论层面的变化。

大多数 AI 应用开发者的第一反应是关注模型本身:这个模型聪明吗?能写代码吗?推理能力强吗?这些当然重要,但在工程化场景里,模型能力只是链路中的一个变量,而不是全部。

真正决定一个 AI 系统能否长期稳定运行的,是整条链路的可靠性。所谓“Your AI Output Is Someone Else's Input”,背后的完整含义是:

  • 你要意识到每个输出都有一个接收方。
  • 你要为接收方定义一个可预期的契约。
  • 你要在接收方消费前设置安全网。
  • 你要把每个环节的失败都变成可观测、可回溯、可控的。

如果用一句话概括,那就是:不要信任模型输出,要验证模型输出。

这种思维不是对模型的否定,而是工程上必需的防御态度。就像你不会信任每一个外部 HTTP 接口返回都满足文档一样,你也不应该因为“大模型看起来很智能”就放弃校验。

如果今天你只带走一个行动项,我希望是:**在你现有的 AI 调用代码里,补上至少一层输出校验。**哪怕只是检查返回的关键字段是否在预期范围内,都能帮你挡住大量的下游事故。

下一步,你可以继续研究:

  • 更复杂的 Agent 编排中如何做输出约定。
  • 如何基于历史输出日志建立评测集,持续追踪模型行为漂移。
  • 如何用流程图和状态机规范多步 AI 链路的异常转移。
  • 如何把校验规则从代码升级为配置,让非开发角色也能参与维护。

这些都是同一个问题的延伸:怎么让 AI 在系统里成为一个可信的组件,而不是一把不受控的文本喷枪。

在实际项目里,这个问题的答案会越来越标准化,甚至会出现专门的“AI 输出网关”类中间件。但在那之前,自己动手补上解析、校验、兜底、观测这四层,永远不会错。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进