123xiao | 无名键客
返回

《从 Prompt 到 Pipeline：中级开发者实战构建可迭代优化的 AI 应用工作流》

字数: 0 阅读时长: 1 分钟

从 Prompt 到 Pipeline：中级开发者实战构建可迭代优化的 AI 应用工作流

很多团队在接入大模型时，第一步往往都差不多：先写一个 Prompt，调通一个接口，看到模型能“回答”，然后很快把它塞进产品里。

问题也通常从这里开始。

一开始只是一个 Prompt，后来变成 3 个 Prompt；再后来要加检索、重试、内容审核、日志、缓存、评估、回滚。到了这个阶段，如果还把 AI 应用当作“一次模型调用”，系统就会越来越脆弱：结果不稳定、成本不可控、定位问题困难、优化没有抓手。

这篇文章我想带你从**“会写 Prompt”，走到“会搭 Pipeline”。目标不是做一个炫技 Demo，而是做一个可迭代优化**的 AI 工作流：能跑、能看、能改、能评估。

背景与问题

为什么单个 Prompt 很快会失控

中级开发者最常见的误区，不是不会调模型，而是把 Prompt 当成最终产品形态。

比如你要做一个“用户反馈总结器”，最开始的实现可能像这样：

  1. 把用户反馈文本拼进 Prompt
  2. 调用模型
  3. 返回总结结果

看上去没毛病，但实际线上会出现几个典型问题：

  • 输入脏数据多：空文本、超长文本、乱码、混杂多语言
  • 输出格式不稳定：有时是 JSON，有时是自然语言，有时缺字段
  • 结果质量波动：同样任务，不同批次输出差异明显
  • 不可观测：你只知道“结果不好”，但不知道是 Prompt、模型、检索还是解析出了问题
  • 难以优化：每次改 Prompt 都像玄学，没有基线，没有回归测试

从“模型调用”到“工作流”的思路转变

更稳妥的方式是把 AI 应用拆成一条流水线：

  • 输入预处理
  • Prompt 组装
  • 模型调用
  • 输出校验
  • 失败重试
  • 结果评估
  • 日志与监控
  • 缓存与成本控制

这其实很像我们熟悉的后端工程：
Prompt 只是业务规则的一部分，Pipeline 才是可维护系统的主体。

前置知识与环境准备

适合谁读

如果你已经具备以下基础，这篇文章会比较合适：

  • 会写 Python
  • 用过至少一个大模型 API
  • 理解基本的 HTTP 调用和 JSON
  • 对日志、重试、缓存这些工程概念不陌生

环境准备

下面的示例使用 Python，尽量保持可运行、可替换。

安装依赖：

pip install flask requests pydantic

环境变量：

export LLM_API_KEY="your_api_key"
export LLM_API_URL="https://your-llm-endpoint/v1/chat/completions"

如果你接的是其他模型平台，只需要改一下请求格式即可。本文重点不在某个厂商 SDK，而在 Pipeline 的组织方式。

核心原理

先给出一句核心结论：

可迭代优化的 AI 应用，不是把 Prompt 写得多花，而是把“输入、推理、输出、评估、回滚”串成可观测、可替换、可验证的 Pipeline。

一个实用的 Pipeline 分层

我通常会把 AI 工作流拆成这几层：

  1. Input Layer：清洗输入、限制长度、做基础校验
  2. Context Layer：补充上下文，比如知识库检索、模板选择、用户画像
  3. Prompt Layer：构造系统提示词、用户提示词、输出格式要求
  4. Inference Layer：调用模型，处理超时、重试、fallback
  5. Guardrail Layer：输出解析、结构化校验、敏感内容过滤
  6. Evaluation Layer：记录质量指标，为后续优化提供依据

流程图：从请求到可用结果

flowchart TD
    A[用户请求] --> B[输入校验与清洗]
    B --> C[上下文构建/检索]
    C --> D[Prompt组装]
    D --> E[模型调用]
    E --> F[输出解析]
    F --> G{结构是否合法?}
    G -- 是 --> H[业务后处理]
    G -- 否 --> I[重试/降级/回退]
    I --> D
    H --> J[记录日志与评估]

为什么“结构化输出”比“自然语言输出”更重要

很多教程喜欢展示模型生成“像人一样的回答”，但在业务系统里，更有价值的是模型能否输出稳定结构。

比如你要的是：

  • summary
  • sentiment
  • priority
  • actions

那就不要让模型自由发挥一大段散文，而要明确要求 JSON 输出，并且在代码层做校验。

这一步非常关键，因为它把“模型行为”转成了“工程可处理的数据”。

时序图：一次完整调用包含什么

sequenceDiagram
    participant U as User
    participant S as Service
    participant R as Retriever
    participant L as LLM
    participant V as Validator
    participant M as Metrics

    U->>S: 提交原始文本
    S->>S: 输入清洗/截断/去噪
    S->>R: 查询相关上下文
    R-->>S: 返回补充信息
    S->>L: 发送结构化Prompt
    L-->>S: 返回模型输出
    S->>V: 解析并校验JSON
    V-->>S: 校验结果
    S->>M: 记录耗时/成本/成功率
    S-->>U: 返回结构化结果

实战：构建一个可迭代优化的反馈总结 Pipeline

为了聚焦工程方法，我们实现一个简单但很常见的场景：

输入：一段用户反馈
输出：结构化分析结果，包括摘要、情绪、优先级、建议动作

第一步：定义目标输出结构

先别急着写 Prompt。先把“我们到底要什么结果”写清楚。

from pydantic import BaseModel, Field
from typing import List

class FeedbackAnalysis(BaseModel):
    summary: str = Field(..., description="用户反馈摘要")
    sentiment: str = Field(..., description="positive/neutral/negative")
    priority: str = Field(..., description="low/medium/high")
    actions: List[str] = Field(..., description="建议的后续动作列表")

这一步的价值在于：

  • 给模型明确目标
  • 给程序明确校验标准
  • 给后续评估明确字段基线

第二步：输入清洗

真实业务里，输入预处理会直接影响模型质量。我踩过一个坑：明明 Prompt 写得没问题，结果线上摘要总是很怪，最后发现是前端把 HTML 标签和追踪参数一起传进来了。

import re

def clean_text(text: str, max_length: int = 2000) -> str:
    text = text.strip()
    text = re.sub(r"<[^>]+>", " ", text)  # 移除简单HTML标签
    text = re.sub(r"\s+", " ", text)      # 合并空白
    return text[:max_length]

建议至少做这些处理：

  • 去空白和无意义字符
  • 截断超长文本
  • 清除 HTML / Markdown 噪声
  • 明确空输入兜底逻辑

第三步：设计 Prompt 模板

这里有个经验：Prompt 不要又长又散，要把任务、约束、格式拆清楚。

def build_prompt(feedback_text: str) -> list:
    system_prompt = """
你是一个企业SaaS产品的客户反馈分析助手。
你的任务是分析用户反馈，并输出严格的JSON。
要求：
1. 只输出JSON，不要输出额外说明
2. 字段必须包含：summary, sentiment, priority, actions
3. sentiment 只能是 positive/neutral/negative
4. priority 只能是 low/medium/high
5. actions 必须是字符串数组
""".strip()

    user_prompt = f"""
请分析下面的用户反馈：

{feedback_text}

请返回如下JSON格式：
{{
  "summary": "简要总结",
  "sentiment": "positive|neutral|negative",
  "priority": "low|medium|high",
  "actions": ["动作1", "动作2"]
}}
""".strip()

    return [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_prompt},
    ]

注意这里的设计点：

  • 角色分层：system 放规则，user 放任务数据
  • 输出约束显式化：不要让模型猜字段名
  • 枚举值限定：便于程序校验和统计

第四步：封装模型调用

这里我们用通用 HTTP 方式调用，避免绑定某个 SDK。

import os
import requests

LLM_API_KEY = os.getenv("LLM_API_KEY")
LLM_API_URL = os.getenv("LLM_API_URL")

def call_llm(messages, model="default-model", timeout=20):
    headers = {
        "Authorization": f"Bearer {LLM_API_KEY}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": model,
        "messages": messages,
        "temperature": 0.2
    }

    response = requests.post(
        LLM_API_URL,
        headers=headers,
        json=payload,
        timeout=timeout
    )
    response.raise_for_status()
    data = response.json()

    return data["choices"][0]["message"]["content"]

为什么 temperature 这里设成 0.2？

因为这个任务是结构化抽取和总结，目标是稳定，不是创意写作。
中级开发者做业务系统时，先追求稳定性，通常比“回答更有灵气”重要得多。

第五步：输出解析与校验

别信任模型输出。哪怕你已经要求 JSON，它还是可能夹带解释文字，或者字段拼错。

import json
from pydantic import ValidationError

def parse_output(raw_output: str) -> FeedbackAnalysis:
    raw_output = raw_output.strip()

    start = raw_output.find("{")
    end = raw_output.rfind("}")
    if start == -1 or end == -1:
        raise ValueError("未找到有效JSON对象")

    json_str = raw_output[start:end+1]
    data = json.loads(json_str)
    return FeedbackAnalysis(**data)

这段逻辑看着朴素，但非常实用：

  • 允许模型前后带少量噪声
  • 强制结构校验
  • 失败时能明确知道是“解析失败”还是“字段不合法”

第六步：加入重试与降级

一条可用的 Pipeline，不能假设模型每次都乖乖工作。最少要考虑：

  • 网络超时
  • 接口 429 / 5xx
  • 输出格式异常
  • 单次模型结果不稳定
import time

def analyze_feedback(text: str, retries: int = 2) -> FeedbackAnalysis:
    cleaned = clean_text(text)
    if not cleaned:
        raise ValueError("输入内容为空")

    last_error = None

    for attempt in range(retries + 1):
        try:
            messages = build_prompt(cleaned)
            raw_output = call_llm(messages)
            result = parse_output(raw_output)
            return result
        except (requests.RequestException, ValueError, ValidationError, json.JSONDecodeError) as e:
            last_error = e
            time.sleep(1 + attempt)

    raise RuntimeError(f"分析失败，重试后仍无法完成: {last_error}")

这里先做的是最基础的重试。再往上走，你可以增加：

  • 不同模型的 fallback
  • 不同 Prompt 版本切换
  • 失败时返回保守默认值

第七步：暴露成一个可运行服务

下面给一个最小可运行的 Flask 服务。

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/analyze", methods=["POST"])
def analyze():
    body = request.get_json(force=True)
    text = body.get("text", "")

    try:
        result = analyze_feedback(text)
        return jsonify({
            "success": True,
            "data": result.dict()
        })
    except Exception as e:
        return jsonify({
            "success": False,
            "error": str(e)
        }), 400

if __name__ == "__main__":
    app.run(debug=True, port=5001)

启动后测试：

curl -X POST http://127.0.0.1:5001/analyze \
  -H "Content-Type: application/json" \
  -d '{"text":"你们新版本加载太慢了，尤其是报表页面，客户演示时很尴尬，希望尽快优化。"}'

预期返回：

{
  "success": true,
  "data": {
    "summary": "用户反馈新版本报表页面加载较慢，影响客户演示体验，希望尽快优化。",
    "sentiment": "negative",
    "priority": "high",
    "actions": [
      "排查报表页面性能瓶颈",
      "优先优化客户演示相关场景",
      "向用户同步处理进展"
    ]
  }
}

让 Pipeline 真正“可迭代”：加上日志与评估

很多团队的问题不在“不会调 Prompt”，而在“调完不知道有没有变好”。

最小化可观测字段

建议至少记录：

  • 请求 ID
  • 输入长度
  • Prompt 版本
  • 模型版本
  • 响应耗时
  • 是否解析成功
  • 重试次数
  • token 成本（如果平台能提供）
  • 人工评分或规则评分

一个简单的日志结构

import time
import uuid

def analyze_feedback_with_metrics(text: str):
    request_id = str(uuid.uuid4())
    start = time.time()
    prompt_version = "feedback_v1"

    try:
        result = analyze_feedback(text)
        duration = round((time.time() - start) * 1000, 2)

        log = {
            "request_id": request_id,
            "prompt_version": prompt_version,
            "input_length": len(text),
            "duration_ms": duration,
            "success": True
        }
        print(log)
        return result
    except Exception as e:
        duration = round((time.time() - start) * 1000, 2)
        log = {
            "request_id": request_id,
            "prompt_version": prompt_version,
            "input_length": len(text),
            "duration_ms": duration,
            "success": False,
            "error": str(e)
        }
        print(log)
        raise

状态图：一次任务在 Pipeline 中的生命周期

stateDiagram-v2
    [*] --> Received
    Received --> Cleaned
    Cleaned --> PromptBuilt
    PromptBuilt --> Inferencing
    Inferencing --> Parsed: 输出有效
    Inferencing --> RetryPending: 超时/格式错误
    RetryPending --> Inferencing
    RetryPending --> Failed: 超过重试上限
    Parsed --> Evaluated
    Evaluated --> [*]
    Failed --> [*]

逐步验证清单

如果你准备把这套方法搬进自己的项目，我建议按这个顺序验证，不要一上来就堆复杂能力。

V1：先跑通最小闭环

  • 输入能正常进入服务
  • Prompt 能稳定返回结果
  • 输出能被 JSON 解析
  • Pydantic 校验通过

V2：再增强稳定性

  • 超时和接口异常可重试
  • 空输入、超长输入有明确处理
  • 输出非法时能识别并报错
  • 关键日志能落盘或入监控系统

V3：最后做优化闭环

  • 每次 Prompt 改动有版本号
  • 有固定测试样本集
  • 能对比不同 Prompt / 模型效果
  • 有人工抽检或自动规则评分

常见坑与排查

这一部分我尽量说一些真实会踩到的坑，而不是只讲理想方案。

坑 1：Prompt 改完感觉“更聪明了”，但线上反而更差

这是最常见的问题。原因通常是：

  • 新 Prompt 对少数样本更好，但整体稳定性变差
  • 输出更详细了，但格式更容易漂移
  • 增加了过多自然语言约束，模型抓不到重点

排查方法：

  1. 固定一组测试样本
  2. 比较改动前后的结构化通过率
  3. 比较字段缺失率、平均耗时、失败率
  4. 不只看“个别惊艳案例”

一句话：不要凭感觉优化 Prompt，要用样本和指标优化。

坑 2：模型明明说了返回 JSON，结果还是不合法

这也非常普遍。原因包括：

  • Prompt 中 JSON 示例不够明确
  • 输出字段要求和自然语言说明冲突
  • 温度过高
  • 输入太复杂，模型注意力分散

排查方法：

  • 降低 temperature
  • 缩短 Prompt，减少模糊描述
  • 明确“只输出 JSON”
  • 做解析兜底，而不是直接 json.loads(raw)

坑 3：检索增强后，效果没提升反而更乱

很多人一加 RAG，就觉得该变强。实际上未必。

可能的问题：

  • 检索内容不相关
  • 上下文太长，把核心任务淹没了
  • 没区分“用户原始输入”和“外部知识”
  • Prompt 没告诉模型该如何使用检索内容

建议：

  • 先验证“无检索版本”的基线效果
  • 只注入最相关的少量上下文
  • 在 Prompt 中明确上下文的用途和优先级

坑 4：线上成本失控

一个常见原因是：
开发阶段只盯效果，没限制输入长度，也没做缓存和分级策略。

排查方向：

  • 平均输入 token 是否过高
  • 是否对重复请求做了缓存
  • 是否所有请求都用了同一高价模型
  • 是否把不需要 AI 的简单规则场景也丢给模型了

安全/性能最佳实践

AI Pipeline 一旦进生产，安全和性能就不是“加分项”，而是“上线门槛”。

1. 不要把敏感信息直接拼进 Prompt

例如：

  • 用户手机号
  • 身份证号
  • 银行卡号
  • 内部密钥
  • 未脱敏业务数据

至少要做脱敏或最小化传输。

def mask_sensitive(text: str) -> str:
    text = re.sub(r"\b1[3-9]\d{9}\b", "[PHONE]", text)
    text = re.sub(r"\b\d{15,18}[\dXx]?\b", "[ID_CARD]", text)
    return text

2. 给模型调用设置超时和并发保护

如果没有超时，模型服务抖动时会把你的线程池拖死。
如果没有并发限制，高峰期会把下游 API 打爆。

基本建议：

  • HTTP 超时：10~30 秒
  • 使用连接池
  • 对下游模型接口做限流
  • 使用队列解耦长任务

3. 结构化结果要做白名单校验

不要直接相信模型输出的任意字段。
比如 priority 只能是 low/medium/high，那就必须校验，不在枚举里的直接判失败。

这不仅是稳定性问题，也是安全边界问题。

4. 对 Prompt 做版本化管理

别把 Prompt 散落在代码里到处复制。建议：

  • 每个 Prompt 有版本号
  • 每次改动有变更记录
  • 能快速回滚到上一个稳定版本

一个简单做法是把 Prompt 模板放到独立文件或配置中心中。

5. 用缓存减少重复调用

对于“相同输入、短期内结果变化不大”的任务，缓存非常划算。

from functools import lru_cache

@lru_cache(maxsize=256)
def cached_analyze(text: str):
    return analyze_feedback(text).dict()

当然，生产环境里更推荐 Redis 之类的外部缓存，而不是进程内缓存。

6. 把 AI 失败当作常态来设计

这点非常重要。
传统接口失败，通常是异常；AI 系统里，部分失败、弱失败、语义失败都很常见。

所以你要提前定义：

  • 失败时是报错、降级还是回退
  • 哪些场景允许空结果
  • 哪些场景必须人工兜底
  • 如何记录失败样本用于后续修复

一种更工程化的组织方式

如果你的项目会继续扩展，我建议把代码拆成类似结构：

ai_app/
  app.py
  pipeline/
    input_cleaner.py
    prompt_builder.py
    llm_client.py
    output_parser.py
    evaluator.py
  schemas/
    feedback.py
  prompts/
    feedback_v1.txt
    feedback_v2.txt
  tests/
    test_pipeline.py

这样做的好处是：

  • Prompt、模型调用、解析逻辑解耦
  • 更容易做 A/B 测试
  • 更容易替换模型供应商
  • 测试边界更清晰

总结

从 Prompt 到 Pipeline，本质上是一次思维升级：

  • Prompt 解决“模型说什么”
  • Pipeline 解决“系统如何稳定地产生可用结果”

如果你已经是中级开发者，我非常建议你把 AI 应用当作一条完整的数据处理流水线，而不是一个会说话的黑盒。真正能持续优化的系统，通常具备这几个特征：

  1. 输入可控：有清洗、有截断、有脱敏
  2. 输出可验：有结构化格式、有严格校验
  3. 过程可观测：有日志、有耗时、有错误分类
  4. 行为可回归：有测试样本、有版本管理、有对比基线
  5. 异常可兜底：有重试、有降级、有 fallback

如果你准备今天就动手，我建议按这个最小路径开始：

  1. 先把当前 Prompt 输出改成结构化 JSON
  2. 给输出加上程序级校验
  3. 补上日志、重试和版本号
  4. 准备一组固定样本做回归测试

做到这一步，你的 AI 应用就已经不再是“碰运气的调用”，而是一条真正能演进的工程化 Pipeline。

如果只记住一句话，我希望是这句：

不要优化一个 Prompt，要优化一条可验证、可迭代、可回滚的 AI 工作流。

分享到：
Share this post via WhatsApp Share this post on Facebook Share this post on X Share this post via Telegram Share this post on Pinterest Share this post via email
上一篇
《微服务架构中分布式事务的实战落地：基于 Saga 模式的设计、补偿与故障排查》
下一篇
《微服务架构中的分布式事务实战：基于 Saga 模式的订单系统一致性设计与落地》