大模型实战：提示工程与RAG架构应用指南

1. 大模型实战：从提示工程到应用落地的系统指南

作为一名长期奋战在AI应用开发一线的工程师，我深知将大模型从"玩具"变成"工具"的挑战。本章将分享我在实际项目中积累的提示工程、函数调用和RAG应用的核心经验，这些方法已经帮助多个企业级项目成功落地。

1.1 为什么提示工程是AI应用的第一道门槛

记得我第一次调用大模型时，简单写了句"写个Python排序函数"，结果得到的代码简直惨不忍睹——没有注释、边界条件处理缺失、甚至用了危险的eval()。这让我意识到：大模型的输出质量与提示质量直接相关。

典型问题场景分析：

• 模糊需求导致模糊输出：当提示过于宽泛时，模型会返回"仅供参考"的内容
• 缺乏约束引发安全隐患：未限制编码规范可能导致不符合团队标准的代码
• 结构缺失增加集成成本：自然语言输出需要人工解析，无法直接用于生产环境

关键认知：提示工程不是"魔法咒语"，而是明确的人机协作协议。好的提示应该像PRD文档一样清晰具体。

1.2 CRIS+原则：工程级提示设计框架

基于数十个项目的实践，我总结出CRIS+框架（比原书CRIS新增Schema要求），这是保证提示有效性的方法论基础：

1.2.1 Clear（清晰性）

• 问题：使用"优化"、"处理"等模糊动词
• 解决方案：采用"动词+宾语+约束条件"结构

python复制# 不良示例
"优化这段代码"

# 良好示例
"将以下Python代码的时间复杂度从O(n²)优化到O(n log n)，仅使用标准库"

1.2.2 Role（角色设定)

• 误区：简单声明"你是程序员"
• 正确做法：定义具体的专业背景和能力

python复制# 不良示例
"你是程序员"

# 良好示例
"你是5年经验的前端工程师，精通React和TypeScript，熟悉ESLint规范"

1.2.3 Instruction Format（指令格式）

• 关键点：明确输入输出的数据结构

python复制# 不良示例
"分析用户数据"

# 良好示例
"""
输入：JSON数组（每个元素含user_id、age、gender）
输出：JSON对象（含age均值、男女比例）
"""

1.2.4 Safety & Scope（安全与范围）

• 必须包含：技术栈限制、危险操作禁止条款

python复制"禁止使用eval/exec、os.system等危险函数；仅使用Python 3.9+标准库"

1.2.5 +Schema（结构化输出）

• 工程价值：使输出可直接被程序解析

python复制"必须输出valid JSON，符合以下Schema：
{'code': int, 'msg': string, 'data': {'result': string}}"

1.3 提示模板工厂：可复用的代码实现

在实际项目中，我开发了以下提示生成器，确保团队保持统一的提示标准：

python复制def build_prompt(role, instruction, input_data, output_schema, safety_rules):
    """
    生成符合CRIS+原则的提示
    :param role: 模型角色
    :param instruction: 具体指令
    :param input_data: 输入数据（字符串/JSON）
    :param output_schema: 输出Schema描述
    :param safety_rules: 安全规则列表
    :return: 完整提示字符串
    """
    safety_text = "- " + "\n- ".join(safety_rules) if safety_rules else "无"
    return f"""
你是{role}。

# 指令
{instruction}

# 输入数据
{input_data}

# 输出要求
1. 严格按照以下Schema输出JSON格式：
{output_schema}
2. 字段类型必须匹配，必填项不可缺失
3. 仅返回JSON，不要添加额外解释

# 安全规则
{safety_text}
"""

# 使用示例
prompt = build_prompt(
    role="数据分析师",
    instruction="从用户评论中提取产品缺陷（最多5个），按出现频率排序",
    input_data='{"comments": ["电池不耐用", "屏幕易碎", "电池不耐用", "系统卡顿"]}',
    output_schema='''{
        "defects": [{"name": string, "count": int}],  # 缺陷列表，按count降序
        "total": int  # 缺陷总数
    }''',
    safety_rules=["仅提取客观缺陷，不添加主观评价", "缺陷名称统一为名词短语"]
)

实战技巧：

1. 在提示末尾重复输出格式要求，合规率可提升30%+
2. 对复杂Schema，提供1个示例输出
3. 使用"必须"、"严格"等强约束词

2. 函数调用：安全接入业务系统的桥梁

2.1 为什么需要函数调用？

在电商客服项目中，我们曾遇到模型虚构天气信息的尴尬。函数调用技术从根本上解决了这个问题：

四大核心价值：

1. 数据真实性：天气数据来自真实API而非模型记忆
2. 操作可控性：所有外部调用需经业务系统授权
3. 流程可审计：完整日志记录每次函数调用
4. 集成便捷性：结构化输出直接对接现有系统

2.2 完整工作流程解析

以查询天气为例，典型交互流程如下：

1. 用户询问："北京明天天气如何？"
2. 应用发送：用户消息+可用函数列表（含get_weather描述）
3. 模型返回：函数调用请求（非自然语言）

json复制{"name": "get_weather", "parameters": {"city": "北京", "date": "2023-11-20"}}

4. 业务系统：执行真实API调用
5. 模型生成：基于API结果组织自然语言回复

异常处理机制：

• 参数校验失败 → 要求模型重新生成
• API调用超时 → 触发重试机制
• 结果不符合预期 → 人工兜底流程

2.3 Qwen-7B本地部署实战

以下是基于Qwen-7B-Chat的完整实现：

python复制from transformers import AutoModelForCausalLM, AutoTokenizer
import json

# 1. 模型加载
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen-7B-Chat", trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
    "Qwen/Qwen-7B-Chat", 
    trust_remote_code=True,
    device_map="auto",
    torch_dtype="auto"
).eval()

# 2. 业务函数定义
def get_weather(city: str, date: str = None) -> dict:
    """获取天气（模拟实现）"""
    mock_data = {
        "北京": {"2023-11-20": {"temp": 8, "desc": "晴", "wind": "3级"}},
        "上海": {"2023-11-20": {"temp": 12, "desc": "多云", "wind": "2级"}}
    }
    date = date or "2023-11-20"
    return mock_data.get(city, {}).get(date, {"error": "无数据"})

# 3. 系统提示设计
system_prompt = """
你是智能助手，可调用以下函数：
1. get_weather(city: str, date: str = None)
  - 功能：查询城市天气
  - 调用条件：用户询问天气时

响应规则：
- 需要调用函数时，返回JSON：{"name":"函数名","parameters":{"参数":"值"}}
- 参数缺失时，询问用户补充
"""

# 4. 交互处理
def process_query(user_query):
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_query}
    ]
    
    # 首次生成
    inputs = tokenizer.apply_chat_template(messages, return_tensors="pt").to("cuda")
    outputs = model.generate(inputs, max_new_tokens=200)
    response = tokenizer.decode(outputs[0], skip_special_tokens=True)
    
    try:
        # 解析函数调用
        call_info = json.loads(response)
        if call_info["name"] == "get_weather":
            weather = get_weather(**call_info["parameters"])
            
            # 二次生成
            messages.append({"role": "assistant", "content": response})
            messages.append({"role": "user", "content": f"函数结果：{weather}"})
            final_output = model.generate(
                tokenizer.apply_chat_template(messages, return_tensors="pt").to("cuda"),
                max_new_tokens=200
            )
            return tokenizer.decode(final_output[0], skip_special_tokens=True)
    except:
        return response  # 直接返回自然语言回复

避坑指南：

1. 函数描述必须精确到参数类型
2. 生产环境必须添加参数校验层
3. 设置调用超时（建议3秒）
4. 重要操作需记录审计日志

3. RAG架构：解决知识时效性难题

3.1 为什么传统方案会失败？

在金融知识库项目中，我们测试了三种方案：

RAG的胜利在于：将知识检索与文本生成解耦。

3.2 生产级RAG架构设计

经过多个项目迭代，我们的稳定架构如下：

code复制[文档来源]
  ↓
[文档加载] → PDF/Word/HTML解析
  ↓
[文本分块] → 语义分块（非固定长度）
  ↓
[向量化] → BGE-large-zh中文模型
  ↓
[向量存储] → Milvus集群
  ↓
[查询流程] → 混合检索（向量+BM25）→ 结果重排 → 提示注入

关键组件选型建议：

3.3 LangChain实现示例

python复制from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma

# 1. 文档处理
loader = PyPDFLoader("企业规范.pdf")
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。"]
)
docs = text_splitter.split_documents(loader.load())

# 2. 向量化
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-large-zh")
db = Chroma.from_documents(docs, embeddings, persist_directory="./chroma_db")

# 3. 检索增强
retriever = db.as_retriever(search_type="mmr", search_kwargs={"k": 3})

def rag_query(question):
    docs = retriever.get_relevant_documents(question)
    context = "\n\n".join(d.page_content for d in docs)
    
    prompt = f"""基于以下上下文回答问题：
{context}

问题：{question}
要求：
1. 严格基于上下文
2. 未提及的内容回答"不清楚"
"""
    return llm(prompt)

性能优化技巧：

1. 对PDF使用OCR预处理（解决扫描件问题）
2. 启用Milvus的IVF_PQ索引（节省70%内存）
3. 使用Cross-Encoder进行结果重排（提升15%准确率）
4. 实现检索缓存（减少重复计算）

4. 生产环境可靠性保障

4.1 输出校验体系

我们在医疗项目中建立了四级校验：

1. 格式校验（JSON Schema）

python复制class DiagnosisOutput(BaseModel):
    disease: str = Field(pattern=r"^[A-Za-z\u4e00-\u9fa5]+$")
    confidence: float = Field(ge=0, le=1)
    treatments: List[str] = Field(min_items=1)

2. 逻辑校验

python复制def validate_treatment(disease: str, treatments: list):
    known_patterns = {
        "感冒": ["休息", "喝水"],
        "骨折": ["固定", "静养"]
    }
    return all(t in known_patterns.get(disease, []) for t in treatments)

3. 安全过滤

python复制sensitive_words = ["自杀", "暴力"]
if any(word in output for word in sensitive_words):
    raise ContentSafetyError

4. 人工复核队列

python复制if confidence < 0.7 or disease in high_risk_list:
    send_to_human_review(output)

4.2 监控指标设计

我们部署的监控看板包含：

4.3 降级方案

当检测到异常时，系统自动切换：

1. 主模型异常 → 降级到较小的本地模型
2. RAG检索超时 → 仅用模型已有知识回答（标注限制）
3. 函数调用失败 → 转为人工处理流程

5. 框架选型实战建议

5.1 LangChain vs LlamaIndex

根据我们的压力测试结果（基于AWS c5.2xlarge）：

选型决策树：

1. 验证阶段：使用LangChain快速原型开发
2. RAG密集型：切换到LlamaIndex
3. 生产部署：针对核心链路自研轻量实现

5.2 技术债务预防

我们项目中采用的解耦设计：

code复制[业务逻辑] ←→ [适配层] ←→ [AI框架]

这样当需要替换LangChain时，只需重写适配层。实际案例中，我们将核心服务从LangChain迁移到自研实现，只花了2人日。

6. 企业级AI助手完整实现

6.1 系统架构

code复制[前端]
  ↑↓ HTTP
[API服务] ←→ [向量库] ←→ [文档库]
  ↑↓           ↑
[LLM]       [SQL数据库]

6.2 核心代码结构

python复制class EnterpriseAssistant:
    def __init__(self):
        self.llm = load_qwen()
        self.vector_db = init_milvus()
        self.sql_conn = init_postgres()
        
    def handle_query(self, user_input):
        # 意图识别
        intent = classify_intent(user_input)
        
        if intent == "policy_query":
            return self.rag_query(user_input)
        elif intent == "data_query":
            return self.sql_query(user_input)
        else:
            return self.llm(user_input)
    
    def rag_query(self, question):
        # 检索增强流程
        docs = retrieve_docs(question)
        prompt = build_rag_prompt(question, docs)
        return generate_with_validation(prompt)
    
    def sql_query(self, query):
        # 语义转SQL
        sql = self.llm(f"将问题转为SQL：{query}")
        if not validate_sql(sql):
            raise SQLInjectionError
        return execute_sql(sql)

6.3 部署优化方案

1. 模型部署：使用vLLM加速推理（提升3倍吞吐）
2. 缓存策略：
   • 模型输出缓存（TTL=1h）
   • 检索结果缓存（TTL=24h）
3. 负载均衡：
   • 按功能拆分endpoint
   • 设置单用户速率限制

7. 经验总结与避坑指南

7.1 典型失败案例

案例1：未做输出校验

• 现象：模型推荐了未批准的药品
• 解决方案：增加药品白名单校验

案例2：过度依赖框架

• 现象：LangChain版本升级导致API变更
• 解决方案：封装核心功能，锁定版本

案例3：检索质量差

• 现象：返回无关文档片段
• 解决方案：引入HyDE查询改写

7.2 效能提升技巧

1. 提示压缩：移除冗余空格（节省15%token）
2. 异步处理：对耗时操作实现async/await
3. 批处理：合并多个小请求（提升吞吐量）
4. 量化部署：使用8bit量化模型（减少40%内存）

在电商客服项目中，这些优化使单机QPS从12提升到35，同时将延迟从2.3s降至1.4s。