Anthropic提示工程交互式课程：从零构建高效对话系统的实战指南

1次阅读

共计 2331 个字符，预计需要花费 6 分钟才能阅读完成。

刚接触对话系统开发时，我踩过不少坑。最典型的问题是 意图识别漂移 （Intent Detection Drift）——用户说 ” 我想订明天的机票 ”，系统却识别成 ” 查询航班动态 ”。另一个头疼的是 上下文丢失（Context Loss），比如用户问 ” 那家餐厅人均多少 ” 后接着说 ” 订个位子 ”，系统却忘记前文指的哪家餐厅。

传统解决方案需要维护庞大的规则库：

正则表达式匹配规则随业务增长呈指数级膨胀
基于关键词的 FAQ 库无法处理语义相近但表述不同的请求
硬编码的对话流程让系统显得机械呆板

维度	传统规则引擎 /NLP 管道	Anthropic 提示工程
响应延迟	50-200ms（依赖规则复杂度）	300-800ms（模型推理时间）
维护成本	需要持续添加新规则 / 语料	通过 few-shot 学习自动泛化
上下文处理	需手动设计状态机	自动保持多轮对话记忆
冷启动难度	需收集大量标注数据	5-10 个示例即可初见成效

关键差异在于：传统方案像编程，提示工程像教学。前者需要精确描述每个逻辑分支，后者则是通过示例教会模型理解需求。

from typing import Dict, Any
from enum import Enum, auto

class DialogState(Enum):
    GREETING = auto()
    COLLECT_INFO = auto()
    CONFIRMATION = auto()
    COMPLETION = auto()

class DialogManager:
    def __init__(self):
        self.state = DialogState.GREETING
        self.context: Dict[str, Any] = {}

    def process_input(self, user_input: str) -> str:
        try:
            if self.state == DialogState.GREETING:
                self.context['user_name'] = extract_name(user_input)
                self.state = DialogState.COLLECT_INFO
                return f"您好{self.context['user_name']}，请问需要什么帮助？"

            # 其他状态处理...

        except Exception as e:
            self.state = DialogState.GREETING  # 出错时重置对话
            return "抱歉，我有点困惑，我们重新开始好吗？"

# 示例调用
dialog = DialogManager()
print(dialog.process_input("我是张三"))  # 输出：您好张三，请问需要什么帮助？

使用 CLAUDE 模型的对话 API 时，关键是将历史对话作为上下文传入：

import anthropic

client = anthropic.Client(api_key="your_api_key")

def chat_with_context(history: list[str], new_query: str) -> str:
    prompt = "\n\n".join(history + [new_query])
    response = client.completion(prompt=f"{anthropic.HUMAN_PROMPT}{prompt}{anthropic.AI_PROMPT}",
        max_tokens_to_sample=1000
    )
    return response.completion

# 使用示例
history = [
    "用户：推荐一家上海本帮菜",
    "AI：推荐老吉士酒家，米其林一星"
]
new_query = "用户：人均消费多少？"
print(chat_with_context(history, new_query))  # 会关联前文回答老吉士的价格

为每个对话会话生成唯一 session_id
对相同输入 + 上下文组合缓存响应结果
用户说 ” 没听清 ” 时能重复最后回答而不会推进状态

from fastapi import FastAPI, Request
import re

app = FastAPI()

banned_words = [...]  # 从数据库加载

def filter_sensitive(text: str) -> str:
    for word in banned_words:
        text = re.sub(rf"(?i){re.escape(word)}", "***", text)
    return text

@app.post("/chat")
async def chat_endpoint(request: Request):
    data = await request.json()
    cleaned_input = filter_sensitive(data["query"])
    # ... 处理逻辑