在这里插入图片描述

摘要

Prompt改一个词效果翻车、结构化输出格式漂移JSON解析失败、多轮对话状态丢失、长上下文遗忘关键约束。本文从某电商客服真实复盘切入,剖析Prompt脆弱性、指令遵循漂移、会话状态管理、上下窗挤压四个痛点,给出版本化+回归测试、结构化输出约束+解析兜底、分级持久化、滚动摘要+约束重注入的量化方案。

1. Prompt脆弱性:改一个词效果天差地别

痛点现场

某电商客服团队把Prompt里"请简洁回答"改成"请简短回答",本意是优化用户体验。结果上线2小时拒答率从5%飙到30%——"简短"被模型理解为尽量少说字数,遇到不确定就拒答省字数。团队没做Prompt回归测试,改完直接上线,客服投诉激增才发现。

更典型的是换模型引发的连锁翻车。团队从GPT-3.5切到国产模型,原Prompt里"你是顶级客服专家"在新模型上触发过度自信,编造退货政策细节。某次回答"7天无理由退换"实际政策是"15天",导致用户维权批量投诉,单次事故处理成本8万元。根因是同一Prompt在不同模型上行为差异巨大,团队误以为Prompt可移植。

最隐蔽的是Prompt耦合业务逻辑。Prompt里硬编码了"满99包邮"阈值,运营改促销规则为满79包邮,Prompt没同步更新,模型仍按99回答,线上跑了3天才发现,错答订单量1.2万单。

根因剖析

Prompt脆弱的底层机理是自然语言指令的语义模糊性。"简洁"和"简短"人类觉得同义,模型行为差异大——因为模型训练时这两个词出现的上下文分布不同,"简短"更常出现在"尽量简短"这类省略指令中,模型学到了"省略=拒答"的关联。这是Embedding空间的语义偏移,人无法直觉感知。

换模型翻车的根因是不同模型的指令遵循分布不同。GPT-3.5对"顶级专家"理解为角色扮演,国产模型可能理解为能力声明进而过度自信。Prompt本质是模型分布的采样条件,换模型等于换分布,条件不变但采样结果变。团队把Prompt当"通用配置"而非"绑定特定模型的制品",是认知错位。

Prompt耦合业务逻辑的根因是把Prompt当文档而非代码。文档可以含业务参数,代码应该参数外置。硬编码阈值在Prompt里,运营改规则时不会想到改Prompt,因为Prompt归算法团队管,规则归运营团队管,两个团队的工具链不通。

工程方案:Prompt版本化+回归测试+参数外置

拒答率超容差

关键词命中下降>5%

全通过

外置到配置中心

Prompt变更PR

CI触发回归测试

标准问题集100条

跑当前模型+对比基线

指标对比

拒绝合并

灰度发布5%流量

线上效果采集30min

效果达标

全量发布

自动回滚+告警

业务参数

Prompt模板变量化

核心三件事:Prompt当代码git版本化,变更走CI回归测试,业务参数外置配置中心不写死在Prompt里。回归测试套件含100条标准问题+期望答案关键词+容差阈值,变更必须全通过才能灰度。

// 来源:promptfoo 0.92 + GitLab CI 自研Prompt流水线

import json
import subprocess
from dataclasses import dataclass

@dataclass
class PromptTestCase:
    query: str
    must_contain: list[str]   # 必含关键词
    must_not_contain: list[str]  # 必不含关键词
    refusal_max: float = 0.1  # 拒答率容差

class PromptRegressionSuite:
    """Prompt回归测试套件,promptfoo配置生成+执行+对比基线"""
    def __init__(self, suite_path: str, baseline_model: str):
        self.cases = [PromptTestCase(**c) for c in json.load(open(suite_path))]
        self.baseline_model = baseline_model

    def build_promptfoo_config(self, prompt_version: str, model: str):
        """生成promptfoo yaml配置"""
        tests = []
        for case in self.cases:
            tests.append({
                "assert": [
                    {"type": "icontains", "value": kw}
                    for kw in case.must_contain
                ] + [
                    {"type": "not-icontains", "value": kw}
                    for kw in case.must_not_contain
                ],
            })
        config = {
            "prompts": [prompt_version],
            "providers": [{"id": model}],
            "tests": tests,
        }
        return config

    def run(self, prompt_version: str, model: str) -> tuple[bool, dict]:
        """执行回归测试,对比基线指标"""
        config = self.build_promptfoo_config(prompt_version, model)
        config_path = "/tmp/promptfoo_config.yaml"
        import yaml
        yaml.dump(config, open(config_path, "w"))
        # promptfoo eval执行
        result = subprocess.run(
            ["promptfoo", "eval", "-c", config_path, "--output", "json"],
            capture_output=True, text=True
        )
        metrics = json.loads(result.stdout)
        # 汇总:通过率、拒答率、关键词命中率
        pass_rate = metrics["stats"]["passes"] / metrics["stats"]["total"]
        refusal_rate = self._calc_refusal_rate(metrics)
        # 对比基线容差
        if pass_rate < 0.85:
            return False, {"reason": f"通过率{pass_rate:.2%}<85%"}
        if refusal_rate > self.cases[0].refusal_max:
            return False, {"reason": f"拒答率{refusal_rate:.2%}>{self.cases[0].refusal_max}"}
        return True, metrics

    def _calc_refusal_rate(self, metrics):
        refusal_markers = ["抱歉", "无法", "不清楚", "建议联系"]
        refusal = 0
        for r in metrics["results"]:
            if any(m in r["output"] for m in refusal_markers):
                refusal += 1
        return refusal / len(metrics["results"])

class BusinessParamInjector:
    """业务参数外置,Prompt模板变量化,从配置中心拉取"""
    def __init__(self, config_center):
        self.config = config_center  # Apollo或Nacos

    def render(self, prompt_template: str) -> str:
        """渲染Prompt模板,注入业务参数"""
        params = self.config.get("customer_service_params")
        return prompt_template.format(**params)
        # 模板里写"满{free_shipping_threshold}包邮"
        # 阈值从配置中心读,运营改规则同步更新

量化指标与边界

某电商客服团队落地后,Prompt变更线上事故从月2次降到0,回归测试覆盖100条标准问题通过率85%门槛。换模型时回归测试提前发现3次行为漂移,避免上线翻车。业务参数外置后,运营改促销规则不再需要算法团队改Prompt,协同耗时从2天降到0。

边界与踩坑:回归测试只能覆盖已知case,新Prompt引入的新行为模式可能漏检——需配线上监控补充,对新出现的高拒答query自动入测试套件。测试套件维护成本是主要开销,需持续补充。换模型必须重跑回归,不同模型对同一Prompt响应不同,Prompt不可裸移植。

2. 指令遵循漂移:结构化输出JSON解析失败

痛点现场

某金融报表抽取场景要求模型返回JSON,Prompt明确写"请以JSON格式返回,字段如下"。模型大部分时候遵守,但约3%的调用返回带markdown包裹的json...,或字段名从驼峰变蛇形(amount变amount_value),或字段类型从数字变字符串("123"而非123)。下游解析失败业务中断,日均10万调用意味着3000次解析失败。

更隐蔽的是字段缺失漂移。schema要求8个字段,模型偶尔只返回6个,缺失字段是"非必填"的但下游代码没做空值处理,直接报错。漂移率虽低但绝对量大,故障工单占了该服务30%。团队排查发现漂移集中出现在长输入(超2000 token)时,模型上下窗被输入占满,输出格式遵循能力下降。

最严重的是幻觉字段。模型返回了schema里没有的字段"risk_score",值是编造的,下游没做schema校验直接存库,数据分析时发现该字段全是噪声,污染了报表。根因是模型在训练时见过类似字段名,输出时"顺手"加上了。

根因剖析

格式漂移的底层机理是模型输出是概率采样,格式遵循是"高概率但不绝对"的模式。当上下窗被长输入挤压,模型分配给格式遵循的注意力减少,格式遵循概率下降,漂移率上升。这是Transformer注意力分配的固有问题,输入越长输出格式越不稳。

字段名漂移(驼峰变蛇形)的根因是模型训练数据里同一概念有多种命名风格,采样时可能跳到另一种。schema描述在Prompt里如果不够强调,模型按"最常见"命名风格输出,未必是schema要求的那种。

幻觉字段的根因是模型训练数据里"risk_score"这类字段名频繁出现,模型生成了"看起来完整"的输出倾向,即使schema没要求也会加上。这是模型的补全本能,schema约束如果不严格,模型会自由发挥。

组织割裂体现在:算法团队只管Prompt写得对,不管下游解析;工程团队只管解析代码,不做schema校验;数据团队拿到脏数据才发现问题。三方都对"格式遵循"负责又都不负责,漂移就发生在这些缝隙里。

工程方案:结构化输出约束+解析兜底+schema校验

成功

失败

校验通过

字段缺失/类型错

模型输出

直接JSON解析

pydantic schema校验

剥离markdown重试

解析成功

正则提取JSON片段

解析成功

重试调用+更强约束

业务处理

默认值填充或重试

三层防御:强约束Prompt降低漂移率,解析兜底修复常见漂移,schema校验拦截幻觉字段。强约束用function calling或JSON mode(OpenAI、Anthropic都支持)让模型原生输出结构化,比Prompt约束更稳。兜底解析处理markdown包裹、字段名漂移等常见问题。schema校验用pydantic严格校验类型和字段,幻觉字段直接丢弃。

// 来源:instructor 0.4.0 + pydantic 2.5 + OpenAI function calling

import json
import re
from pydantic import BaseModel, ValidationError
from instructor import from_openai

class FinancialReport(BaseModel):
    """schema定义,严格校验字段类型"""
    company_name: str
    revenue: float
    profit: float
    period: str
    # 必填字段缺失直接校验失败触发重试

class StructuredOutputGuard:
    """结构化输出三层防御:原生约束+解析兜底+schema校验"""
    def __init__(self, client, schema: type[BaseModel]):
        self.client = from_openai(client)  # instructor patch
        self.schema = schema

    def extract(self, user_input: str, max_retries: int = 2):
        """抽取结构化数据,function calling原生约束"""
        for attempt in range(max_retries):
            try:
                # instructor自动用function calling约束输出格式
                # 比纯Prompt约束漂移率低一个数量级
                result = self.client.chat.completions.create(
                    response_model=self.schema,
                    messages=[{"role": "user", "content": user_input}],
                    max_retries=0,  # 自己控制重试
                )
                return result  # 已通过pydantic校验
            except ValidationError as e:
                # schema校验失败,字段缺失或类型错
                log(f"attempt {attempt} schema校验失败: {e}")
                if attempt < max_retries - 1:
                    # 重试时强化约束Prompt
                    user_input = self._strengthen_constraint(user_input, e)
                    continue
                raise
            except Exception as e:
                # function calling不可用时的兜底路径
                log(f"function calling失败,走兜底解析: {e}")
                return self._fallback_parse(user_input)

    def _strengthen_constraint(self, original_input: str, error: ValidationError):
        """schema校验失败时,把错误信息回灌强化约束"""
        missing_fields = [err["loc"][0] for err in error.errors()]
        return f"""{original_input}

请确保返回JSON包含以下必填字段:{missing_fields}
字段类型严格按schema,数字字段不要返回字符串。"""

    def _fallback_parse(self, user_input: str):
        """function calling不可用时的兜底解析路径"""
        raw = self._raw_infer(user_input)
        # 1. 直接解析
        try:
            return self.schema(**json.loads(raw))
        except Exception:
            pass
        # 2. 剥离markdown包裹
        cleaned = re.sub(r'```json\s*', '', raw)
        cleaned = re.sub(r'```\s*', '', cleaned).strip()
        try:
            return self.schema(**json.loads(cleaned))
        except Exception:
            pass
        # 3. 正则提取最外层JSON
        match = re.search(r'\{.*\}', raw, re.DOTALL)
        if match:
            try:
                return self.schema(**json.loads(match.group()))
            except Exception:
                pass
        return None  # 全部失败,上游降级处理

    def _raw_infer(self, user_input: str) -> str:
        """原生模型调用,强化结构化约束Prompt"""
        schema_json = json.dumps(
            self.schema.model_json_schema(), ensure_ascii=False
        )
        return self.client.chat.completions.create(
            messages=[{
                "role": "user",
                "content": f"""严格按以下JSON schema返回,不要任何额外文字或markdown包裹:
schema: {schema_json}

输入: {user_input}

直接输出JSON对象,不要用```json包裹,不要添加schema外的字段:"""
            }],
        ).choices[0].message.content

量化指标与边界

某金融报表抽取团队落地后,JSON解析成功率从97%提到99.8%(function calling原生约束+pydantic校验),兜底解析再救回1.5%,剩余0.2%触发重试。幻觉字段率从0.5%降到0(schema校验直接丢弃非schema字段)。长输入场景漂移率从8%压到1%(function calling约束比Prompt约束更抗长输入干扰)。

边界与踩坑:function calling并非所有模型都支持,国产模型支持参差,需配兜底解析路径。schema严格时字段缺失也失败,需配默认值策略或重试——重试增加延迟和成本,阈值要平衡。instructor库对pydantic v2支持最好,v1有兼容问题。强约束Prompt增加token消耗(schema描述约200 token),高频调用场景成本敏感。

3. 多轮对话状态管理:会话超时状态丢失

痛点现场

某工单助手对话到第5轮收集了用户设备型号、故障现象、已尝试操作,用户等2分钟接电话再回来问"刚才说到哪了",系统返回"请重新描述您的问题"——会话超时状态被清,前5轮上下文全丢,用户重复描述体验极差。

根因是会话状态存内存,超时清理避免内存泄漏,但清理太积极丢用户上下文。团队按"5分钟无交互清会话"配置,工单场景用户中途处理事情2分钟很常见,被误杀。更隐蔽的是内存会话无上限,高峰期5000并发会话占8GB内存,OOM重启丢全部会话,影响所有在线用户。

分级存储缺失也导致恢复断层。团队后来加了Redis持久化,但只存最近1小时,超时用户回来发现"之前的都忘了",因为Redis清理后没留摘要。用户体感是"系统失忆",工单一次解决率从70%降到55%,转人工率上升。

根因剖析

会话状态丢失的底层机理是存储分层与生命周期管理缺失。内存存活跃会话快但易失,Redis存近期会话可恢复但有TTL,DB存历史摘要可长期但查询慢。只用内存会OOM和重启丢失,只加Redis不分级会清理断层,三层分级才完整。

清理太积极的根因是TTL配置按"技术指标"(内存占用)而非"业务指标"(用户回头率)。5分钟清理内存友好,但工单场景用户回头周期可能20分钟,技术TTL和业务周期不匹配。这是配置参数脱离业务场景的典型问题。

OOM重启丢全部会话的根因是会话状态无上限增长,没有反压机制。5000并发会话×每会话4KB上下文=20MB理论值,但实际含历史轮次和元数据膨胀到8GB,说明会话状态没做裁剪,历史轮次全量保留。内存资源无上限配额,高峰期直接OOM。

工程方案:会话状态分级持久化+滚动摘要

5分钟无交互

1小时无交互

L1命中

L2命中

仅L3摘要

会话状态更新

L1内存热存

L2 Redis温存

滚动摘要

L3 DB冷存摘要

释放内存和Redis

用户回来

查找会话

无感恢复

从Redis恢复

用摘要重建上下文

摘要+新一轮重新开始

三级存储:内存存活跃会话(毫秒级访问),Redis存近期会话(秒级恢复),DB存摘要(长期可查)。关键设计是滚动摘要——超过1小时的会话不存全量历史,由模型生成关键事实摘要存DB,用户回来时用摘要重建上下文,既省存储又保留关键信息。

内存会话做配额上限和LRU淘汰,超限淘汰最久未活跃的会话转Redis,避免OOM。每会话状态裁剪到最近3轮+摘要,不全量保留历史。

// 来源:Redis 7.2 + PostgreSQL 16 + 自研会话管理

import redis
import json
import time
import threading
from collections import OrderedDict

class SessionStateManager:
    """三级会话存储:内存LRU+Redis温存+DB摘要冷存"""
    def __init__(self, redis_client, db_client, llm_client):
        self.redis = redis_client
        self.db = db_client
        self.llm = llm_client
        self.memory: OrderedDict = OrderedDict()  # LRU缓存
        self.memory_max = 1000  # 内存会话上限防OOM
        self.memory_ttl = 300   # 5分钟无交互降级Redis
        self.redis_ttl = 3600   # 1小时无交互降级DB摘要
        self.recent_keep = 3    # 保留最近3轮原文

    def get(self, session_id: str) -> dict | None:
        """按L1→L2→L3层级查找会话状态"""
        # L1 内存
        if session_id in self.memory:
            self.memory.move_to_end(session_id)
            return self.memory[session_id]
        # L2 Redis
        data = self.redis.get(f"session:{session_id}")
        if data:
            state = json.loads(data)
            self._admit_memory(session_id, state)
            return state
        # L3 DB摘要
        summary = self.db.get_session_summary(session_id)
        if summary:
            # 用摘要重建会话状态
            state = {
                "history_summary": summary,
                "recent": [],  # 摘要后无原文轮次
                "created_from_summary": True,
            }
            self._admit_memory(session_id, state)
            return state
        return None  # 新会话

    def put(self, session_id: str, state: dict):
        """更新会话状态,裁剪历史+同步Redis"""
        state["last_active"] = time.time()
        # 裁剪:只保留最近3轮原文,更早的进摘要
        state = self._compact(state)
        self._admit_memory(session_id, state)
        # 异步同步Redis
        self.redis.setex(
            f"session:{session_id}",
            self.redis_ttl,
            json.dumps(state, ensure_ascii=False),
        )

    def _compact(self, state: dict) -> dict:
        """裁剪会话状态,超出recent_keep的轮次滚入摘要"""
        recent = state.get("recent", [])
        if len(recent) <= self.recent_keep:
            return state
        # 要滚入摘要的旧轮次
        to_summarize = recent[:-self.recent_keep]
        old_summary = state.get("history_summary", "")
        # 调用LLM生成滚动摘要
        new_summary = self._summarize(old_summary, to_summarize)
        state["history_summary"] = new_summary
        state["recent"] = recent[-self.recent_keep:]
        return state

    def _summarize(self, old_summary: str, old_turns: list) -> str:
        """滚动摘要:旧摘要+新轮次→新摘要,保留关键事实"""
        turns_text = "\n".join(
            f"用户:{t['user']}\n助手:{t['assistant']}" for t in old_turns
        )
        prompt = f"""已有摘要:{old_summary}

新增对话:
{turns_text}

请更新摘要,保留关键事实(设备型号、故障现象、已尝试操作、用户诉求),删除冗余。摘要不超过200字:"""
        return self.llm.infer(prompt)

    def _admit_memory(self, session_id: str, state: dict):
        """会话入内存,LRU淘汰超限会话"""
        if session_id in self.memory:
            self.memory.move_to_end(session_id)
        self.memory[session_id] = state
        # 超上限淘汰最久未活跃的,降级到Redis
        while len(self.memory) > self.memory_max:
            old_sid, old_state = self.memory.popitem(last=False)
            self.redis.setex(
                f"session:{old_sid}",
                self.redis_ttl,
                json.dumps(old_state, ensure_ascii=False),
            )

    def idle_sweep(self):
        """定时清理,超时会话降级存储"""
        now = time.time()
        for sid, state in list(self.memory.items()):
            idle = now - state.get("last_active", 0)
            if idle > self.redis_ttl:
                # 超1小时,摘要存DB释放内存和Redis
                summary = state.get("history_summary", "")
                if not summary:
                    summary = self._summarize("", state.get("recent", []))
                self.db.save_session_summary(sid, summary)
                del self.memory[sid]
                self.redis.delete(f"session:{sid}")

量化指标与边界

某工单助手落地后,会话恢复成功率从60%(内存超时即丢)提到99%(三级持久化),5分钟内回来无感恢复,1小时后用摘要重建保留关键事实。内存占用从无上限8GB变成可控2GB(1000会话上限+裁剪),OOM消失。工单一次解决率从55%回升到72%,转人工率下降。

边界与踩坑:摘要重建丢失细节,重要事实(设备型号、故障代码)需在摘要Prompt中明确要求保留,否则模型可能漏掉。滚动摘要增加LLM调用成本,高频会话场景成本敏感——可配批量摘要或仅在降级时触发。三级存储增加运维复杂度,需监控各级命中率(L1>70%为健康,L3<5%为健康)。分级过期时间需按业务调,客服5分钟合适,工单场景可延长到30分钟。

4. 长上下文遗忘:关键约束在多轮中被稀释

痛点现场

某法律咨询助手Prompt开头写了"始终引用法条编号",用户前3轮回答都规范引用,到第8轮开始遗漏法条只给结论。排查发现模型上下窗虽支持16K,但关键约束在第1轮Prompt里,随着对话轮次增加上下文被新轮次填满,早期约束的注意力权重被稀释,模型"忘了"开头的要求。

更严重的是中段遗忘。用户在第3轮说了"我是未成年人",到第7轮问劳动纠纷,模型按成年人逻辑回答了"可以独立签劳动合同",完全忘了未成年约束。这是Lost in the Middle现象——长上下文中段信息注意力权重低,模型容易忽略。用户投诉"前后矛盾",信任度崩塌。

根因是对话轮次全量塞上下文,8轮对话约4000 token,加上系统Prompt和检索内容,上下窗挤压严重。团队没做上下文压缩,每轮都把历史全量重发,token消耗线性增长,到第10轮单次调用成本是第1轮的5倍。

根因剖析

长上下文遗忘的底层机理是Transformer注意力分配在长序列上不均匀。U形曲线:开头和结尾注意力权重高,中段低。关键约束放Prompt开头,随着对话轮次增加,开头约束相对位置被推远,注意力权重下降,模型"遗忘"。这是模型架构的固有限制,上下窗越大越明显。

中段遗忘的根因是关键信息(如"未成年人")出现在中段轮次,恰好落在注意力低谷。即使上下窗够大,模型也未必"看到"中段信息。这不是上下窗长度问题,是注意力分配问题——扩大上下窗不解决遗忘,反而让中段更稀释。

全量历史重发的根因是团队把上下文当"全量记忆"而非"压缩记忆"。每轮重发全量历史是为了让模型"记住"上下文,但全量重发既贵又加剧遗忘(中段信息更多)。正确做法是滚动摘要+只保留近期原文,既省token又强化近期信息注意力。

工程方案:滚动摘要+约束重注入+关键事实槽

新一轮对话

组装上下文

系统Prompt含关键约束

历史摘要

关键事实槽

最近3轮原文

当前用户输入

模型推理

更新摘要

更新关键事实槽

裁剪历史保留最近3轮

三招对抗遗忘:约束重注入(每轮把关键约束放系统Prompt末尾而非开头,利用结尾注意力高)、关键事实槽(结构化存储关键信息如用户身份、诉求,每轮注入避免依赖中段历史)、滚动摘要(历史压成摘要,原文只留最近3轮,减少上下文长度)。事实槽是结构化数据不参与注意力稀释,模型"必看"。

// 来源:LangGraph 0.0.40 + 自研上下文管理

from dataclasses import dataclass, field
from typing import Any

@dataclass
class FactSlot:
    """关键事实槽,结构化存储避免依赖历史中段"""
    user_identity: dict = field(default_factory=dict)  # 用户身份如未成年
    user_demand: str = ""        # 用户核心诉求
    constraints: list[str] = field(default_factory=list)  # 已知约束
    resolved_facts: list[str] = field(default_factory=list)  # 已确认事实

class ContextManager:
    """上下文管理:约束重注入+事实槽+滚动摘要"""
    def __init__(self, llm_client, system_constraints: list[str]):
        self.llm = llm_client
        self.system_constraints = system_constraints  # 如"始终引用法条编号"

    def build_context(self, session_state: dict, current_input: str) -> list[dict]:
        """组装上下文,约束重注入到系统Prompt末尾利用结尾高注意力"""
        fact_slot: FactSlot = session_state.get("fact_slot", FactSlot())
        messages = []
        # 系统Prompt:角色+摘要+事实槽+约束(约束放末尾)
        system_content = f"""你是法律咨询助手。

历史摘要:{session_state.get('history_summary', '无')}

关键事实:
- 用户身份:{fact_slot.user_identity}
- 用户诉求:{fact_slot.user_demand}
- 已知约束:{fact_slot.constraints}

强制约束(每轮必须遵守):
{chr(10).join(self.system_constraints)}
以上约束在本次回答中必须严格执行。"""
        messages.append({"role": "system", "content": system_content})
        # 最近3轮原文(近期信息注意力高)
        for turn in session_state.get("recent", []):
            messages.append({"role": "user", "content": turn["user"]})
            messages.append({"role": "assistant", "content": turn["assistant"]})
        # 当前输入
        messages.append({"role": "user", "content": current_input})
        return messages

    def update_after_turn(self, session_state: dict, user_input: str, assistant_output: str):
        """每轮后更新摘要、事实槽、裁剪历史"""
        # 加入最近轮次
        recent = session_state.setdefault("recent", [])
        recent.append({"user": user_input, "assistant": assistant_output})
        # 更新事实槽(调LLM抽取关键事实)
        fact_slot = session_state.get("fact_slot", FactSlot())
        fact_slot = self._extract_facts(user_input, assistant_output, fact_slot)
        session_state["fact_slot"] = fact_slot
        # 裁剪历史,超出3轮的滚入摘要
        if len(recent) > 3:
            old_turns = recent[:-3]
            session_state["recent"] = recent[-3:]
            session_state["history_summary"] = self._roll_summary(
                session_state.get("history_summary", ""), old_turns
            )
        return session_state

    def _extract_facts(self, user_input: str, assistant_output: str, slot: FactSlot) -> FactSlot:
        """抽取关键事实到事实槽,结构化存储对抗中段遗忘"""
        prompt = f"""从对话中抽取关键事实,更新到已有事实槽:
已有:身份={slot.user_identity}, 诉求={slot.user_demand}, 约束={slot.constraints}

用户:{user_input}
助手:{assistant_output}

返回JSON:{{"user_identity":{{}}, "user_demand":"", "constraints":[], "resolved_facts":[]}}"""
        result = self.llm.infer_json(prompt)
        # 合并更新,新事实覆盖旧
        slot.user_identity.update(result.get("user_identity", {}))
        if result.get("user_demand"):
            slot.user_demand = result["user_demand"]
        for c in result.get("constraints", []):
            if c not in slot.constraints:
                slot.constraints.append(c)
        for f in result.get("resolved_facts", []):
            if f not in slot.resolved_facts:
                slot.resolved_facts.append(f)
        return slot

    def _roll_summary(self, old_summary: str, old_turns: list) -> str:
        """滚动摘要,保留关键事实压缩历史"""
        turns_text = "\n".join(f"用户:{t['user']}\n助手:{t['assistant']}" for t in old_turns)
        prompt = f"""已有摘要:{old_summary}

新增对话:
{turns_text}

更新摘要,保留关键事实(法条引用、用户身份、已确认结论),不超过200字:"""
        return self.llm.infer(prompt)

量化指标与边界

某法律咨询助手落地后,关键约束遗忘率(如不引用法条)从第8轮的40%降到5%(约束重注入到末尾),中段遗忘(如忽略未成年身份)从30%降到8%(事实槽结构化注入)。单次调用token从第10轮的5倍降到1.5倍(滚动摘要压缩历史),成本下降60%。用户投诉"前后矛盾"从月15件降到1件。

边界与踩坑:约束重注入对结尾注意力高的模型有效,部分模型U形曲线不明显效果打折——需A/B测试验证。事实槽抽取依赖LLM准确性,抽取错误会污染所有后续轮次,需配人工校验或高置信度过滤。滚动摘要丢失细节,重要事实需在摘要Prompt中明确保留。上下文管理增加每轮LLM调用(摘要+事实抽取),延迟和成本上升,高频场景需权衡——可配每N轮才摘要一次。

总结

Prompt与应用层的本质是把自然语言指令当工程制品管理。Prompt版本化+回归测试让变更不再翻车,结构化输出约束+解析兜底让格式漂移可控,会话状态分级持久化让多轮对话不丢上下文,滚动摘要+约束重注入+事实槽让长上下文不遗忘关键约束。四招共同把脆弱的自然语言交互锻造成可控的工程接口。

Logo

电商企业物流数字化转型必备!快递鸟 API 接口,72 小时快速完成物流系统集成。全流程实战1V1指导,营造开放的API技术生态圈。

更多推荐