从提示编写到系统架构：如何成为创建灵活提示内容管理系统的高手提示工程架构师
关键词
提示工程（Prompt Engineering）、提示内容管理系统（PCMS）、Prompt Lifecycle、变量模板、多模态提示、可观测性、场景路由
摘要
当你还在为“调Prompt”熬夜时，优秀的提示工程架构师已经在搭建提示内容管理系统（Prompt Content Management System, PCMS）——一套能让提示像“数字资产”一样被高效管理、灵活适配、快速迭代的基础设施。
本文将带你从“提示编写者”升级为“系统架构师”：
用“剧本管理”的生活化比喻拆解PCMS的核心逻辑；掌握PCMS的7大核心组件（模板引擎、变量系统、场景路由等）；通过“电商客服PCMS”的真实案例，学会从0到1搭建灵活系统；理解“可观测性”“多模态”等前沿技术如何让PCMS更智能。
读完本文，你将具备设计**“适配千变万化场景、支撑百万级提示调用”**的PCMS的能力，成为大模型时代稀缺的“提示架构高手”。
一、为什么你需要从“写Prompt”转向“做PCMS”？
1.1 大模型时代的“提示危机”
3年前，大模型还只是实验室工具，提示是程序员手里的“小纸条”——比如用
"写一首关于春天的诗"
电商客服需要10套提示适配“订单查询”“退货申请”“售后投诉”；教育APP需要20套提示对应“小学低年级”“高中奥数”“成人职场”；企业知识库需要50套提示匹配“技术文档”“产品手册”“客户案例”。
如果还在用“记事本存Prompt”“复制粘贴改参数”，你会遇到3个致命问题：
一致性差：不同客服用不同提示，导致回复风格混乱；维护成本高：修改一个通用规则（比如“禁止说‘不知道’”），要改100个Prompt；迭代困难：不知道哪个Prompt效果好，也不知道用户讨厌哪种回复。
1.2 PCMS：提示的“数字资产管理系统”
PCMS的本质，是把“零散的Prompt”变成“可复用、可配置、可迭代的模板”。就像电商的商品管理系统——你不用手动写每个商品的描述，而是用“模板+变量”生成：
商品模板：
"{{品牌}} {{型号}} 手机，搭载{{处理器}}，电池容量{{电池}}mAh"
{{品牌}}=苹果
{{型号}}=iPhone 15
{{处理器}}=A17 Pro
PCMS做的事，就是把“商品”换成“提示”——用“模板+变量”管理提示，让提示能适配不同场景、自动填充数据、快速迭代优化。
1.3 目标读者与核心挑战
目标读者：提示工程师、大模型应用开发者、想转型“提示架构师”的技术人；核心挑战：
如何让提示“灵活”——适配100个场景而不用写100个Prompt？如何让提示“好管”——改一个规则就能同步所有相关Prompt？如何让提示“智能”——根据用户上下文自动调整内容？
二、PCMS的核心概念：用“剧本 analogy”讲清楚
为了让你快速理解PCMS的逻辑，我们用**“拍电影”**做类比——大模型是“演员”，提示是“剧本”，PCMS是“剧本管理系统”。
2.1 核心概念1：Prompt Lifecycle（提示的“生命周期”）
拍电影要经历“写剧本→选演员→排练→上映→复盘”，提示也有完整的生命周期：
需求分析：搞清楚“这个提示要解决什么问题”（比如“让客服回复更共情”）；模板设计：写“通用剧本”（比如“先共情，再解答，最后引导”）；变量定义：确定“需要填充的空”（比如
{{用户问题}}
{{订单状态}}
2.2 核心概念2：变量模板（提示的“填空游戏”）
你肯定用过邮件模板：
"亲爱的{{姓名}}，您的快递{{快递单号}}已发出"
示例：电商客服的提示模板
你是友好的电商客服小蜜，用户的问题是：{{user_question}}
当前订单{{order_id}}的状态是：{{order_status}}（枚举：待支付/已发货/已签收）
请按照以下规则回复：
1. 第一句共情（比如“我理解您想了解订单进度的心情”）；
2. 第二句用大白话解答订单状态；
3. 第三句问“还有什么可以帮您的吗？”
禁止说“不知道”“不清楚”，禁止使用技术术语。
变量的价值：把“固定内容”和“动态数据”分开——你不用写100个“订单状态”的Prompt，只要改
{{order_status}}
2.3 核心概念3：场景路由（提示的“智能分诊台”）
医院的分诊台会把“发烧病人”分到内科、“骨折病人”分到骨科。PCMS的场景路由，就是提示的“分诊台”——根据用户场景匹配对应的模板：
场景标签：
["电商", "售后", "订单查询"]
就像外卖APP的“地址匹配”——你输入“北京市朝阳区”，APP会自动推荐“附近的奶茶店”，场景路由做的是“输入场景标签，推荐对应的提示模板”。
2.4 核心概念4：多模态提示（提示的“图文菜谱”）
早期的提示只有文字，但今天的大模型支持多模态（文字+图片+语音），比如GPT-4V能看图片、Claude 3能听语音。PCMS需要管理多模态提示，就像“菜谱”——文字说明+步骤图+语音讲解：
示例：教育APP的数学题提示
你是小学数学老师，要讲解{{题目}}：“小明有5个苹果，给了小红2个，还剩几个？”
请结合以下资源：
1. 图片：{{示例图}}（苹果分配的示意图）；
2. 语音：{{讲解语音}}（用儿童能听懂的语气读题目）；
3. 公式：{{公式}}（5-2=3）。
讲解要慢，每步都问“小朋友，你懂了吗？”
多模态提示的价值，是让大模型的回复更“直观”——就像教孩子做饭，光说“打鸡蛋”不如“看图片+听声音”。
三、PCMS的核心组件：从0到1搭建的“方法论”
要做一个灵活的PCMS，你需要掌握7个核心组件。我们用“搭建电商客服PCMS”为例，一步步拆解每个组件的设计逻辑。
3.1 组件1：模板引擎——提示的“语法编译器”
模板引擎是PCMS的“心脏”，负责把“模板+变量”渲染成“最终的Prompt”。就像Word的邮件合并功能——你写好模板，选好数据源，一键生成100封邮件。
3.1.1 模板引擎的核心要求
支持变量替换：比如
{{user_question}}
{% if order_status == "已发货" %}提醒物流信息{% endif %}
<|FunctionCallBegin|>{"name":"get_order_status","parameters":{"order_id":"123"}}<|FunctionCallEnd|>
3.1.2 技术实现：用Python+Jinja2做模板引擎
Jinja2是Python生态最流行的模板引擎，完全满足PCMS的需求。我们用它写一个电商客服的模板：
from jinja2 import Environment, BaseLoader
# 1. 定义提示模板
prompt_template = """
你是电商客服小蜜，用户的问题是：{{ user_question }}
当前订单{{ order_id }}的状态是：{{ order_status }}
{% if order_status == "已发货" %}
物流信息：{{ logistics_info }}（预计{{ delivery_time }}送达）
{% elif order_status == "退货中" %}
退货地址：{{ return_address }}（请在3天内寄回）
{% else %}
请等待商家处理，有进展会通知你。
{% endif %}
回复要友好，最后问“还有什么可以帮你吗？”
"""
# 2. 定义变量
variables = {
"user_question": "我的订单怎么还没到？",
"order_id": "12345",
"order_status": "已发货",
"logistics_info": "顺丰快递（单号：SF123456789）",
"delivery_time": "2024-05-20"
}
# 3. 渲染Prompt
env = Environment(loader=BaseLoader())
template = env.from_string(prompt_template)
final_prompt = template.render(**variables)
print(final_prompt)
输出结果：
你是电商客服小蜜，用户的问题是：我的订单怎么还没到？
当前订单12345的状态是：已发货
物流信息：顺丰快递（单号：SF123456789）（预计2024-05-20送达）
回复要友好，最后问“还有什么可以帮你吗？”
3.1.3 进阶技巧：支持多模态变量
如果要加入图片变量（比如
{{ product_img }}
# 多模态模板
multimodal_template = """
你是电商产品顾问，用户问的是：{{ user_question }}
产品图片：<image>{{ product_img }}</image>
产品参数：{{ product_spec }}
请结合图片讲解产品功能，比如“图片里的摄像头是{{ camera_pixel }}像素”。
"""
# 变量（product_img是OSS的图片URL）
variables = {
"user_question": "这个手机的摄像头怎么样？",
"product_img": "https://oss.example.com/iphone15-camera.jpg",
"product_spec": "iPhone 15，6.1英寸屏幕，A17 Pro处理器",
"camera_pixel": "4800万"
}
3.2 组件2：变量管理系统——提示的“数据源”
变量是模板的“原料”，变量管理系统要解决3个问题：变量从哪来？变量是什么类型？变量怎么验证？
3.2.1 变量的3种来源
用户输入：比如
{{ user_question }}
{{ order_status }}
{{ product_img }}
我们用“电商客服”的变量来源举例子：
| 变量名 | 来源 | 说明 |
|---|---|---|
| user_question | 用户输入 | 直接取用户的问题 |
| order_id | 用户输入 | 用户提供的订单号 |
| order_status | 订单系统API | 调用 |
| logistics_info | 物流系统API | 调用 |
| return_address | 售后系统数据库 | 从 |
3.2.2 变量的4种类型
字符串：比如
{{ user_question }}
{{ order_status }}
{{ delivery_days }}
{{ product_img }}
{{ guide_audio }}
3.2.3 变量的验证机制
为了避免“变量不存在”导致的渲染错误，我们需要给变量加校验规则：
必填校验：比如
order_id
delivery_days
order_status
["待支付", "已发货", "已签收", "退货中"]
技术实现：用Pydantic做变量校验（Pydantic是Python的“数据校验库”，支持类型提示和规则定义）：
from pydantic import BaseModel, Field, HttpUrl
from enum import Enum
# 1. 定义订单状态的枚举
class OrderStatus(str, Enum):
PENDING = "待支付"
SHIPPED = "已发货"
DELIVERED = "已签收"
RETURNING = "退货中"
# 2. 定义变量的校验模型
class CustomerServiceVariables(BaseModel):
user_question: str = Field(..., description="用户的问题，必填")
order_id: str = Field(..., description="订单号，必填，格式：数字+字母")
order_status: OrderStatus = Field(..., description="订单状态，必填")
logistics_info: str = Field(None, description="物流信息，可选")
return_address: str = Field(None, description="退货地址，可选")
product_img: HttpUrl = Field(None, description="产品图片URL，必须是HTTP/HTTPS链接")
# 3. 验证变量
variables = {
"user_question": "我的订单怎么还没到？",
"order_id": "12345",
"order_status": "已发货", # 符合Enum的取值
"product_img": "https://oss.example.com/iphone15.jpg" # 符合HttpUrl规则
}
# 校验通过（无异常）
validated_vars = CustomerServiceVariables(**variables)
# 校验失败的例子（order_status填了"未知"）
invalid_vars = {
"user_question": "我的订单怎么还没到？",
"order_id": "12345",
"order_status": "未知" # 不在Enum的取值里
}
# 会抛出ValidationError：Value error, '未知' is not a valid OrderStatus
3.3 组件3：场景路由——提示的“智能分诊台”
场景路由是PCMS的“大脑”，负责根据“用户场景”匹配对应的“提示模板”。就像导航软件的“路线规划”——你输入“从A到B”，软件会根据“实时路况”推荐最快的路线。
3.3.1 场景路由的2种实现方式
规则引擎：用“如果-那么”（If-Then）规则匹配场景，适合简单场景；机器学习：用分类模型（比如BERT）判断用户意图，适合复杂场景。
我们先讲规则引擎的实现——用Python的
rules-engine
from rule_engine import Rule, Context
# 1. 定义提示模板和对应的规则
templates = [
{
"id": "order_query",
"name": "订单查询模板",
"tags": ["电商", "售后", "订单查询"],
"rule": "order_status in ['已发货', '已签收'] and user_question contains '订单' or '物流'",
"content": "订单查询的提示模板..."
},
{
"id": "return_apply",
"name": "退货申请模板",
"tags": ["电商", "售后", "退货"],
"rule": "order_status == '已签收' and user_question contains '退货' or '退款'",
"content": "退货申请的提示模板..."
},
{
"id": "complain",
"name": "投诉模板",
"tags": ["电商", "售后", "投诉"],
"rule": "user_question contains '投诉' or '差评' or '不满意'",
"content": "投诉的提示模板..."
}
]
# 2. 定义场景路由函数
def match_template(user_vars):
# 把变量转成规则引擎的上下文
context = Context(user_vars)
for template in templates:
# 解析规则
rule = Rule(template["rule"])
# 匹配规则
if rule.matches(context):
return template
return None # 没有匹配的模板
# 3. 测试场景匹配
user_vars = {
"user_question": "我的订单12345已经发货了，什么时候到？",
"order_status": "已发货"
}
# 匹配到"订单查询模板"（规则满足：order_status是已发货，user_question包含"订单"）
matched_template = match_template(user_vars)
print(matched_template["name"]) # 输出：订单查询模板
3.3.2 进阶：用机器学习做意图分类
如果场景很多（比如100个），规则引擎会变得“臃肿”，这时候需要用意图分类模型——比如用BERT训练一个模型，输入用户的问题，输出“订单查询”“退货申请”等意图。
技术实现步骤：
数据标注：收集1000条用户问题，标注意图（比如“我的订单怎么没到？”→“订单查询”）；训练模型：用Hugging Face的
transformers
3.4 组件4：版本控制——提示的“时光机”
你肯定遇到过“改坏了Prompt，想回退到之前的版本”的情况。版本控制就是PCMS的“时光机”，负责记录每个模板的“修改历史”，支持回退、对比、审计。
3.4.1 版本控制的核心功能
版本记录：每个模板的修改都生成一个版本（比如v1.0、v1.1、v2.0）；版本对比：查看v1.0和v2.0的差异（比如新增了“物流信息”变量）；版本回退：如果v2.0效果不好，一键回退到v1.0；版本审计：记录谁修改了模板，修改时间是什么。
3.4.2 技术实现：用Git做版本控制
Git是最流行的版本控制系统，完全可以用来管理提示模板。我们用“电商客服模板”举例子：
初始化仓库：在PCMS的模板目录下运行
git init
git add . && git commit -m "v1.0 初始版本"
git commit -m "v1.1 新增物流信息"
git checkout v1.0
git diff v1.0 v1.1
3.4.3 进阶：结合数据库做版本管理
如果模板存在数据库里（比如PostgreSQL），可以加一个
template_versions
| 字段名 | 类型 | 说明 |
|---|---|---|
| version_id | UUID | 版本ID |
| template_id | UUID | 关联的模板ID |
| content | TEXT | 模板内容 |
| change_log | TEXT | 修改说明（比如“新增物流信息变量”） |
| created_by | VARCHAR(50) | 修改人 |
| created_at | TIMESTAMP | 修改时间 |
这样，你可以通过
template_id
version_id
3.5 组件5：可观测性——提示的“体检报告”
可观测性是PCMS的“眼睛”，负责监控提示的“效果”——比如哪个模板调用最多？哪个模板用户最讨厌？哪个模板消耗的tokens最多？
3.5.1 可观测性的3大指标
调用指标：调用量（PV）、QPS（每秒请求数）、响应时间；效果指标：有效率（符合预期的回复占比）、满意度（用户好评占比）、投诉率（用户投诉占比）；成本指标：tokens消耗（每调用的tokens数）、API费用（大模型的调用成本）。
我们用“电商客服”的可观测性指标举例子：
| 模板ID | 调用量 | 有效率 | 满意度 | tokens消耗 |
|---|---|---|---|---|
| order_query | 1000 | 90% | 85% | 120 |
| return_apply | 500 | 80% | 75% | 150 |
| complain | 200 | 70% | 60% | 200 |
从表中可以看出：
order_query效果最好（有效率90%、满意度85%），可以推广；complain效果最差（有效率70%、满意度60%），需要优化；return_apply的tokens消耗比order_query高（150 vs 120），可能是模板太长，需要精简。
3.5.2 技术实现：用OpenTelemetry做可观测性
OpenTelemetry（OTel）是云原生时代的“可观测性标准”，支持追踪（Trace）、指标（Metric）、日志（Log）。我们用它来监控提示的调用流程：
安装依赖：
pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
# 1. 初始化Tracer
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)
otlp_exporter = OTLPSpanExporter(endpoint="http://jaeger-collector:4318/v1/traces")
span_processor = BatchSpanProcessor(otlp_exporter)
trace.get_tracer_provider().add_span_processor(span_processor)
# 2. 在模板渲染函数中加追踪
def render_prompt(template, variables):
with tracer.start_as_current_span("render_prompt") as span:
# 记录模板ID和变量
span.set_attribute("template.id", template["id"])
span.set_attribute("variables.user_question", variables["user_question"])
# 渲染模板
env = Environment(loader=BaseLoader())
rendered = env.from_string(template["content"]).render(**variables)
# 记录渲染结果的长度
span.set_attribute("rendered.length", len(rendered))
return rendered
# 3. 在大模型调用函数中加追踪
def call_llm(prompt):
with tracer.start_as_current_span("call_llm") as span:
span.set_attribute("prompt.length", len(prompt))
# 调用大模型API（比如OpenAI的GPT-4）
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
# 记录大模型的回复和tokens消耗
span.set_attribute("llm.response", response.choices[0].message.content)
span.set_attribute("llm.tokens.prompt", response.usage.prompt_tokens)
span.set_attribute("llm.tokens.completion", response.usage.completion_tokens)
return response
3.5.3 可视化：用Grafana做Dashboard
把OTel的数据导出到Prometheus，再用Grafana做可视化 Dashboard，你可以看到：
实时调用量趋势（比如晚上8点是客服高峰）；每个模板的有效率变化（比如v2.0比v1.0提高了20%）；大模型的tokens消耗（比如complain模板消耗最多，需要优化）。
3.6 组件6：多模态支持——提示的“多媒体工具箱”
随着GPT-4V、Claude 3等多模态大模型的普及，PCMS必须支持文字+图片+语音+视频的提示管理。
3.6.1 多模态提示的结构
多模态提示的核心，是把不同类型的资源“嵌入”到Prompt中。不同大模型的语法不同，比如：
GPT-4V：用
<image>
"这是一张图片：<image>https://example.com/img.jpg</image>"
[[image://https://example.com/img.jpg]]
content
{"type": "image_url", "image_url": {"url": "https://example.com/img.jpg"}}
PCMS需要适配不同大模型的语法，让用户不用关心底层细节——比如用户上传一张图片，PCMS自动生成对应大模型的语法。
3.6.2 技术实现：多模态资源管理
多模态资源（图片、语音、视频）需要存储在**对象存储（OSS）**中（比如阿里云OSS、AWS S3），PCMS负责：
上传：用户上传图片，PCMS把图片存到OSS，返回URL；关联：把图片URL和提示模板关联（比如
{{ product_img }} = OSS URL
我们用“教育APP的数学题提示”举例子：
上传图片：用户上传“苹果分配示意图”到OSS，得到URL：
https://oss.example.com/apple-division.jpg
你是小学数学老师，要讲解题目：{{ question }}
请结合图片讲解：<image>{{ image_url }}</image>
步骤要慢，每步都问“小朋友，你懂了吗？”
渲染Prompt：变量
{{ question }} = "小明有5个苹果，给了小红2个，还剩几个？"
{{ image_url }} = "https://oss.example.com/apple-division.jpg"
你是小学数学老师，要讲解题目：小明有5个苹果，给了小红2个，还剩几个？
请结合图片讲解：<image>https://oss.example.com/apple-division.jpg</image>
步骤要慢，每步都问“小朋友，你懂了吗？”
3.7 组件7：权限管理——提示的“安全闸门”
PCMS中的提示是“业务核心资产”，必须保证只有授权的人才能修改模板。权限管理要解决2个问题：谁能访问？谁能修改？
3.7.1 权限的2种类型
只读权限：比如客服人员只能看模板，不能修改；读写权限：比如提示工程师可以修改模板，发布新版本；** admin权限**：比如架构师可以管理用户权限，审计版本历史。
3.7.2 技术实现：用RBAC做权限管理
RBAC（Role-Based Access Control，基于角色的访问控制）是最流行的权限模型，核心逻辑是“用户→角色→权限”：
用户：比如“张三”“李四”；角色：比如“提示工程师”“客服人员”“admin”；权限：比如“查看模板”“修改模板”“发布版本”。
我们用“电商客服PCMS”的RBAC举例子：
| 角色 | 权限 |
|---|---|
| 客服人员 | 查看模板、调用模板 |
| 提示工程师 | 查看模板、修改模板、发布版本 |
| admin | 所有权限（包括管理用户、审计历史） |
技术实现可以用FastAPI的OAuth2或者Keycloak（开源的身份管理系统），比如用FastAPI做权限校验：
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# 模拟用户角色数据库
users = {
"zhangsan": {"password": "123456", "role": "prompt_engineer"},
"lisi": {"password": "654321", "role": "customer_service"},
"admin": {"password": "admin123", "role": "admin"}
}
# 权限校验函数
def get_current_user_role(token: str = Depends(oauth2_scheme)):
# 模拟从token中解析用户（实际用JWT）
username = token # 简化处理，实际要解析JWT
if username not in users:
raise HTTPException(status_code=401, detail="无效的token")
return users[username]["role"]
# 需要“提示工程师”权限的接口
@app.put("/api/templates/{template_id}")
def update_template(template_id: str, role: str = Depends(get_current_user_role)):
if role != "prompt_engineer" and role != "admin":
raise HTTPException(status_code=403, detail="没有修改权限")
# 执行修改模板的逻辑
return {"message": "模板修改成功"}
# 需要“客服人员”权限的接口
@app.post("/api/templates/{template_id}/render")
def render_template(template_id: str, role: str = Depends(get_current_user_role)):
if role not in ["customer_service", "prompt_engineer", "admin"]:
raise HTTPException(status_code=403, detail="没有调用权限")
# 执行渲染模板的逻辑
return {"rendered_prompt": "你是电商客服小蜜..."}
四、实战：用PCMS搭建“电商客服系统”
现在，我们把前面的组件整合起来，用Python+FastAPI+Jinja2+OTel搭建一个完整的电商客服PCMS。
4.1 系统架构图
graph TD
A[用户] --> B[电商客服前端]
B --> C[PCMS API]
C --> D[场景路由]
D --> E[模板引擎]
E --> F[变量管理系统]
F --> G[内部系统API]
F --> H[OSS（多模态资源）]
E --> I[大模型API（GPT-4）]
C --> J[可观测性（OTel+Grafana）]
C --> K[版本控制（Git）]
C --> L[权限管理（RBAC）]
4.2 分步实现
4.2.1 步骤1：初始化项目
创建虚拟环境：
python -m venv pcms-env
source pcms-env/bin/activate
pcms-envScriptsactivate
pip install fastapi uvicorn jinja2 pydantic openai opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp
4.2.2 步骤2：定义数据模型
用Pydantic定义模板、变量、场景的模型：
# models.py
from pydantic import BaseModel, Field, HttpUrl
from enum import Enum
from typing import List, Optional
# 订单状态枚举
class OrderStatus(str, Enum):
PENDING = "待支付"
SHIPPED = "已发货"
DELIVERED = "已签收"
RETURNING = "退货中"
# 提示模板模型
class PromptTemplate(BaseModel):
id: str = Field(..., description="模板ID")
name: str = Field(..., description="模板名称")
tags: List[str] = Field(..., description="场景标签")
content: str = Field(..., description="模板内容")
variables: List[str] = Field(..., description="变量列表")
llm: str = Field(..., description="适配的大模型（比如gpt-4、claude-3）")
# 变量模型
class CustomerServiceVariables(BaseModel):
user_question: str = Field(..., description="用户问题")
order_id: str = Field(..., description="订单号")
order_status: OrderStatus = Field(..., description="订单状态")
logistics_info: Optional[str] = Field(None, description="物流信息")
return_address: Optional[str] = Field(None, description="退货地址")
product_img: Optional[HttpUrl] = Field(None, description="产品图片URL")
4.2.3 步骤3：实现核心功能
用FastAPI实现模板渲染、场景匹配、大模型调用：
# main.py
from fastapi import FastAPI, Depends, HTTPException
from jinja2 import Environment, BaseLoader
from pydantic import ValidationError
import openai
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from models import PromptTemplate, CustomerServiceVariables, OrderStatus
# 初始化FastAPI
app = FastAPI(title="电商客服PCMS", version="1.0")
# 初始化OpenAI（替换成你的API key）
openai.api_key = "your-openai-api-key"
# 初始化OTel
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)
otlp_exporter = OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces")
span_processor = BatchSpanProcessor(otlp_exporter)
trace.get_tracer_provider().add_span_processor(span_processor)
# 模拟模板数据库
templates = [
PromptTemplate(
id="order_query",
name="订单查询模板",
tags=["电商", "售后", "订单查询"],
content="""
你是电商客服小蜜，用户的问题是：{{ user_question }}
当前订单{{ order_id }}的状态是：{{ order_status }}
{% if order_status == "已发货" %}
物流信息：{{ logistics_info }}（预计{{ delivery_time }}送达）
{% endif %}
请共情用户，然后解答，最后问“还有什么可以帮您的吗？”
""",
variables=["user_question", "order_id", "order_status", "logistics_info", "delivery_time"],
llm="gpt-4"
),
PromptTemplate(
id="return_apply",
name="退货申请模板",
tags=["电商", "售后", "退货"],
content="""
你是电商客服小蜜，用户的问题是：{{ user_question }}
当前订单{{ order_id }}的状态是：{{ order_status }}
{% if order_status == "已签收" %}
退货地址：{{ return_address }}（请在3天内寄回，邮费到付）
{% endif %}
请引导用户提供退货原因，最后问“需要帮你预约上门取件吗？”
""",
variables=["user_question", "order_id", "order_status", "return_address"],
llm="gpt-4"
)
]
# 场景匹配函数
def match_template(tags: List[str]) -> PromptTemplate:
max_intersection = 0
matched = None
for template in templates:
intersection = len(set(template.tags) & set(tags))
if intersection > max_intersection:
max_intersection = intersection
matched = template
if not matched:
raise HTTPException(status_code=404, detail="没有匹配的模板")
return matched
# 模板渲染函数
def render_template(template: PromptTemplate, variables: dict) -> str:
with tracer.start_as_current_span("render_template") as span:
span.set_attribute("template.id", template.id)
span.set_attribute("template.name", template.name)
env = Environment(loader=BaseLoader())
try:
rendered = env.from_string(template.content).render(**variables)
except Exception as e:
raise HTTPException(status_code=500, detail=f"模板渲染错误：{str(e)}")
span.set_attribute("rendered.prompt", rendered)
return rendered
# 大模型调用函数
def call_llm(prompt: str, llm: str) -> str:
with tracer.start_as_current_span("call_llm") as span:
span.set_attribute("llm.model", llm)
span.set_attribute("prompt.length", len(prompt))
try:
if llm == "gpt-4":
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
else:
raise HTTPException(status_code=400, detail=f"不支持的大模型：{llm}")
except Exception as e:
raise HTTPException(status_code=500, detail=f"大模型调用错误：{str(e)}")
span.set_attribute("llm.response", response.choices[0].message.content)
span.set_attribute("llm.tokens.prompt", response.usage.prompt_tokens)
span.set_attribute("llm.tokens.completion", response.usage.completion_tokens)
return response.choices[0].message.content
# 核心API：处理客服请求
@app.post("/api/customer-service")
def handle_customer_service(vars: CustomerServiceVariables):
with tracer.start_as_current_span("handle_customer_service") as span:
# 1. 验证变量
try:
validated_vars = CustomerServiceVariables(**vars.dict())
except ValidationError as e:
raise HTTPException(status_code=422, detail=e.errors())
span.set_attribute("user.question", validated_vars.user_question)
span.set_attribute("order.id", validated_vars.order_id)
span.set_attribute("order.status", validated_vars.order_status)
# 2. 场景匹配
# 根据订单状态和用户问题生成场景标签
scenario_tags = ["电商", "售后"]
if "订单" in validated_vars.user_question or "物流" in validated_vars.user_question:
scenario_tags.append("订单查询")
elif "退货" in validated_vars.user_question or "退款" in validated_vars.user_question:
scenario_tags.append("退货申请")
matched_template = match_template(scenario_tags)
span.set_attribute("matched.template.id", matched_template.id)
span.set_attribute("matched.template.name", matched_template.name)
# 3. 渲染模板
# 把validated_vars转成字典，过滤掉None的变量
var_dict = {k: v for k, v in validated_vars.dict().items() if v is not None}
rendered_prompt = render_template(matched_template, var_dict)
span.set_attribute("rendered.prompt", rendered_prompt)
# 4. 调用大模型
llm_response = call_llm(rendered_prompt, matched_template.llm)
span.set_attribute("llm.response", llm_response)
# 5. 返回结果
return {
"template_id": matched_template.id,
"template_name": matched_template.name,
"rendered_prompt": rendered_prompt,
"llm_response": llm_response
}
4.2.4 步骤4：运行与测试
启动OTel Collector：OTel Collector是接收OTel数据的中间件，你可以用Docker运行：
docker run -p 4318:4318 otel/opentelemetry-collector-contrib:latest
启动PCMS：
uvicorn main:app --reload
http://localhost:8000/api/customer-service
{
"user_question": "我的订单12345怎么还没到？",
"order_id": "12345",
"order_status": "已发货",
"logistics_info": "顺丰快递（单号：SF123456789）",
"delivery_time": "2024-05-20"
}
查看结果：你会得到渲染后的Prompt和大模型的回复，同时OTel会把数据发送到Collector，用Grafana可以看到可视化 Dashboard。
4.3 常见问题与解决方案
© 版权声明
文章版权归作者所有，未经允许请勿转载。
相关文章
暂无评论...
