## 第一章：Query改写——把口语翻译成检索语言

### 1.1 为什么要改写

用户输入和文档语言之间存在"表达鸿沟"。文档里写的是"主轴驱动模块电源输入电压范围180-264VAC"，用户问的是"这玩意儿电压要多少"。一个是标准化的技术表述，一个是口语化的提问。

向量检索有一定语义泛化能力，但碰到这种差距大的情况常常翻车。我们的实验显示，对query做改写后，Hit@5能提升4-8个百分点。

### 1.2 改写的核心规则

我们设计了一个三层改写策略：

**第一层：规范化**

去除口语词、语气词、重复词，补全省略的主语宾语。

```python
def normalize_query(query: str) -> str:
    # 口语词映射
    mapping = {
        "咋": "怎么",
        "啥": "什么",
        "咋整": "如何处理",
        "好不好": "是否正常",
        "行不行": "是否可以",
        "那": "",  # 指代词单独处理
        "这": "",
        "个": "",
    }
    
    for k, v in mapping.items():
        query = query.replace(k, v)
    
    # 去除重复字符
    query = re.sub(r'(.)\1{2,}', r'\1\1', query)
    
    # 去除句首无意义词
    query = re.sub(r'^(那|这|那个|这个|请问|我想问一下)\s*', '', query)
    
    return query.strip()
```

**第二层：实体补全**

识别query中的实体（设备编号、故障代码、标准名称），确保格式规范化。

```python
import re

def normalize_entities(query: str) -> str:
    # 设备编号：统一大写
    # MC-2023-001、mc2023-001 → MC-2023-001
    pattern = r'([A-Za-z]{2,})[-_]?(\d{4})[-_]?(\d{3})?'
    query = re.sub(pattern, lambda m: m.group(1).upper() + '-' + m.group(2), query)
    
    # 故障代码：确保有"报警"前缀
    # E201 → 报警E201
    pattern = r'\b([A-Z])(\d{3})\b'
    query = re.sub(pattern, r'\1\2报警', query)  # E201报警
    
    # 型号名称标准化
    # 用维护的实体词典替换
    for alias, standard in entity_dict.items():
        query = query.replace(alias, standard)
    
    return query
```

**第三层：语义补全**

结合会话上下文，补全省略的信息。这层最复杂，后面单独讲。

### 1.3 改写的实现方案

初期尝试了纯规则，覆盖了70%的常见模式，但边缘case太多，规则数量爆炸。后来改为**规则+小模型**的混合方案：

```python
class QueryRewriter:
    def __init__(self):
        self.rules = load_rules()
        self.llm = LLMClient(model="qwen2-7b")
    
    def rewrite(self, query: str, context: dict = None) -> str:
        # 第一层：规则快速处理
        normalized = self.apply_rules(query)
        
        # 第二层：判断是否需要LLM改写
        if self.need_llm_rewrite(normalized):
            # 构造改写的prompt
            prompt = self.build_rewrite_prompt(normalized, context)
            rewritten = self.llm.generate(prompt)
            return rewritten
        
        return normalized
    
    def need_llm_rewrite(self, query: str) -> bool:
        # 满足以下条件之一需要LLM介入
        conditions = [
            len(query) < 5,                          # 太短
            self.has_demonstrative(query),          # 有指代词（这、那）
            self.count_nouns(query) < 2,            # 实词太少
            self.is_vague_question(query)           # 模糊问句
        ]
        return any(conditions)
```

LLM改写的prompt：

```
你是一个检索查询改写专家。将用户的口语化问题改写为适合向量检索的技术问句。

规则：
1. 补全省略的设备名称、参数名称
2. 将指代词替换为具体的实体（从上下文中获取）
3. 将模糊的动词替换为精确的技术动作（"弄" → "处理"）
4. 如果问题信息严重不足（少于2个有效实体），返回"INSUFFICIENT"
5. 不要添加用户没有提及的新信息
6. 只输出改写后的问句，不要解释

对话上下文：
{context}

用户当前输入：{query}

改写结果：
```

### 1.4 一些坑和经验

**坑一：过度改写**

初期prompt写得太自由，LLM经常会"补充"一些用户没提到的信息。比如"MC-2023"被改写成"MC-2023数控机床主轴驱动参数"，加了"数控机床"和"主轴驱动"——如果用户其实想问的是这个设备的维修记录呢？改写方向错了，检索直接跑偏。

后来在prompt里加了严格限制，并且对LLM的输出做了二次验证：

```python
def validate_rewrite(original: str, rewritten: str) -> bool:
    # 检查改写结果是否保留了原query的核心实体
    original_entities = extract_entities(original)
    rewritten_entities = extract_entities(rewritten)
    
    # 必须保留至少一个核心实体
    if not original_entities:
        return True  # 原query没有实体，没办法
    
    if not any(e in rewritten for e in original_entities):
        return False
    
    # 改写后长度不能超过原query的3倍
    if len(rewritten) > len(original) * 3:
        return False
    
    return True
```

**坑二：性能问题**

LLM改写耗时80-150ms，对整体延迟影响很大。做了两级优化：

1. **缓存**：同样的query在短期内只改写一次。我们发现32%的查询是重复的。
2. **分级处理**：只有规则无法处理的query才走LLM，规则能解决的直接返回。规则覆盖约65%的query，只有35%走LLM。

**坑三：短query是死穴**

"温度""电压""报警"这种单个词的query，给什么模型都救不了。我们的做法是判断为"信息不足"后，不强行改写，而是返回一个引导式的追问：

```python
def handle_insufficient_query(query: str) -> str:
    prompts = {
        "温度": "请问您想查询什么设备的温度？是正常工作温度范围、还是温度异常报警的处理方法？",
        "电压": "请问您想查询哪个设备的电压参数？输入电压还是输出电压？",
        "报警": "请问您遇到了哪个报警代码？"
    }
    
    for keyword, prompt in prompts.items():
        if keyword in query:
            return prompt
    
    return "您的问题信息较少，能否补充一下具体的设备名称或故障现象？"
```

## 第二章：意图分类——知道用户到底想干嘛

### 2.1 为什么要做意图分类

"MC-2023"这个query，在不同场景下可能指向不同意图：

- 查参数 → 需要检索规格说明书
- 查故障 → 需要检索维修日志
- 查库存 → 需要调API

同一个query，不同意图下最优的检索策略完全不同。如果不做意图分类，只能用通用策略覆盖所有情况，效果必然打折扣。

### 2.2 意图体系设计

我们设计了三级意图体系：

```
一级意图（4类）
├── 知识查询：问"是什么""怎么做"
├── 数据查询：问"是多少""当前状态"
├── 操作引导：问"怎么操作""步骤是什么"
└── 模糊/其他：无法明确归类

二级意图（细化）
├── 知识查询
│   ├── 参数查询
│   ├── 故障处理
│   ├── 操作流程
│   ├── 规范标准
│   └── 原理说明
├── 数据查询
│   ├── 实时数据（调用API）
│   ├── 历史记录
│   └── 统计分析
└── 操作引导
    ├── 设备操作
    ├── 软件操作
    └── 安全规程
```

### 2.3 分类实现

**第一阶段：规则分类**

快速上线，覆盖高频场景：

```python
def classify_by_rules(query: str) -> str:
    # 数字+单位 → 参数查询
    if re.search(r'\d+(\.\d+)?\s*[A-Za-z]', query):
        return 'knowledge/parameter'
    
    # 包含"代码""编号""报警号" → 故障处理
    if any(kw in query for kw in ['代码', '编号', '报警', '故障']):
        return 'knowledge/troubleshooting'
    
    # 包含"多少""当前""现在" → 数据查询
    if any(kw in query for kw in ['多少', '当前', '现在']):
        return 'data/current'
    
    # 包含"怎么""步骤""如何" → 操作引导
    if any(kw in query for kw in ['怎么', '步骤', '如何', '操作']):
        return 'guide/operation'
    
    return 'unknown'
```

规则覆盖了约60%的query，准确率约85%。

**第二阶段：模型分类**

对规则覆盖不了的，用fasttext训练了一个轻量分类器：

```python
import fasttext

def train_intent_classifier(train_data_path: str):
    # 训练数据格式：__label__intent1 文本内容
    model = fasttext.train_supervised(
        train_data_path,
        lr=0.5,
        epoch=25,
        wordNgrams=2,
        dim=100
    )
    return model

# 推理
def classify_intent(query: str) -> str:
    labels, probs = model.predict(query, k=3)
    if probs[0] > 0.7:
        return labels[0].replace('__label__', '')
    else:
        return 'unknown'  # 低置信度交给规则
```

训练数据来源：历史query的人工标注，累积了8000条。fasttext模型大小仅几MB，推理时间<10ms，非常适合线上部署。

模型分类准确率约78%，加上规则兜底，整体准确率能做到88%。

### 2.4 意图驱动的差异化处理

分类后不是简单的打标签，而是驱动后续处理：

```python
def route_by_intent(query: str, intent: str):
    if intent.startswith('knowledge'):
        # 纯RAG检索
        return rag_search(query)
    
    elif intent.startswith('data'):
        # 尝试API调用，失败则降级到RAG
        api_result = call_data_api(query)
        if api_result:
            return api_result
        else:
            return rag_search(query)
    
    elif intent.startswith('guide'):
        # RAG + 偏向操作手册的检索权重
        return rag_search(query, category_boost={'operation_manual': 2.0})
    
    else:
        # 未知意图：通用RAG + 返回追问
        result = rag_search(query)
        result['suggestion'] = "如果没找到答案，请补充更多信息"
        return result
```

## 第三章：指代消解——把"它"变成具体的东西

### 3.1 问题的严重性

我们统计了5000条历史query，发现约12%包含指代词（它、这、那、该设备、上述方法）。这些query如果不做指代消解，检索几乎必然失败。

### 3.2 方案一：简单回溯

用上一轮的实体作为替换：

```python
class CoreferenceResolver:
    def __init__(self):
        self.session_entities = {}  # session_id -> 最近提到的实体列表
    
    def resolve(self, query: str, session_id: str) -> str:
        # 提取当前query中的指代词
        references = self.extract_references(query)
        if not references:
            return query
        
        # 从session中获取最近实体
        entities = self.session_entities.get(session_id, [])
        if not entities:
            return query  # 没有上下文，无法消解
        
        # 替换
        resolved = query
        for ref in references:
            # 用最近的实体替换
            if ref in ['它', '该', '其']:
                replaced = entities[0] if entities else ref
            elif ref == '那个':
                replaced = entities[1] if len(entities) > 1 else entities[0]
            else:
                replaced = entities[0] if entities else ref
            
            resolved = resolved.replace(ref, replaced)
        
        return resolved
    
    def extract_references(self, query: str) -> List[str]:
        # 提取指代词
        patterns = ['它', '该', '其', '那个', '这个', '上述', '上述方法', '该设备']
        return [p for p in patterns if p in query]
    
    def update_session(self, session_id: str, entities: List[str]):
        # 更新session中的实体列表（保留最近3个）
        old = self.session_entities.get(session_id, [])
        # 去重合并，新的放前面
        merged = entities + [e for e in old if e not in entities]
        self.session_entities[session_id] = merged[:3]
```

### 3.3 方案二：LLM联合消解

简单回溯不够用。比如：

> 用户："MC-2023机床的加工精度是多少？"
> 系统："MC-2023的加工精度为±0.01mm"
> 用户："那它的定位精度呢？"

"它"指MC-2023，但"定位精度"和"加工精度"是同一设备的两个参数，需要从文档中补全。

LLM能做更好的消解：

```python
def llm_coreference_resolution(query: str, history: List[dict]) -> str:
    # 构造prompt，把对话历史传进去
    history_text = "\n".join([
        f"用户：{turn['user']}\n系统：{turn['assistant']}" 
        for turn in history[-3:]  # 只保留最近3轮
    ])
    
    prompt = f"""
    以下是对话历史：
    {history_text}
    
    用户当前说：{query}
    
    请将当前问题中的指代词替换为具体的实体，使其成为一个独立可检索的问句。
    只输出替换后的结果，不要解释。
    """
    
    return llm.generate(prompt)
```

LLM消解的准确率比规则高15个百分点，但延迟增加了80ms。权衡后采用混合策略：先试规则，规则的置信度低于0.7才走LLM。

## 第四章：Query扩展——把问号变成网

### 4.1 为什么要扩展

一个query只检索一次，命中的是某个"点"。但很多时候用户的问题有多个侧面，一个方向的答案不够。

比如"主轴温度报警"，可能涉及：
- 报警代码含义
- 报警阈值设定
- 处理方法
- 预防措施

只检索一次只能覆盖一个方向。Query扩展可以生成多个视角的问句，覆盖更全面。

### 4.2 实现方案

```python
def expand_query(query: str) -> List[str]:
    prompt = f"""
    给定一个用户问题，生成4个语义相似但表达不同的检索问句。
    这些问句应该从不同角度覆盖同一个问题的不同侧面。
    
    原始问题：{query}
    
    生成的4个问句（每行一个）：
    """
    
    variants = llm.generate(prompt).strip().split('\n')
    return [query] + variants
```

然后多个变体分别检索，用RRF合并结果：

```python
def rrf_merge(results_lists: List[List[Dict]], k: int = 60) -> List[Dict]:
    scores = {}
    for results in results_lists:
        for rank, doc in enumerate(results):
            score = 1.0 / (k + rank + 1)
            doc_id = doc['id']
            if doc_id not in scores:
                scores[doc_id] = {'score': 0, 'doc': doc}
            scores[doc_id]['score'] += score
    
    sorted_docs = sorted(
        scores.values(),
        key=lambda x: x['score'],
        reverse=True
    )
    
    return [item['doc'] for item in sorted_docs]
```

### 4.3 性能与取舍

Query扩展的代价是检索次数翻倍。我们只对"复杂的、开放性的"问题做扩展，对"明确的、事实性的"问题不做。

判断逻辑：

```python
def need_expansion(query: str) -> bool:
    # 明确事实性问题不做扩展
    if re.search(r'[A-Z]{2,}-\d{4,}', query):  # 有设备编号
        return False
    if re.search(r'\d+\.?\d*\s*[A-Za-z]', query):  # 有具体数值
        return False
    
    # 开放性问题做扩展
    open_words = ['如何', '怎么', '为什么', '有哪些', '什么', '怎么样']
    if any(w in query for w in open_words):
        return True
    
    return False
```

## 第五章：用户意图的长期追踪

### 5.1 用户画像

不同岗位的用户，对同一问题的需求不同：

- 维修工问"主轴温度报警" → 要的是操作步骤
- 工艺工程师问同样的问题 → 要的是原因分析和参数调整建议
- 质量主管问同样的问题 → 要的是历史发生频率和影响范围

我们给每个用户维护一个"意图偏好"：

```python
def update_user_profile(user_id: str, query: str, intent: str):
    profile = get_profile(user_id)
    profile['query_history'].append({
        'query': query,
        'intent': intent,
        'timestamp': datetime.now()
    })
    
    # 分析用户最常用的意图
    intents = [h['intent'] for h in profile['query_history'][-50:]]
    profile['preferred_intent'] = Counter(intents).most_common(1)[0][0]
    
    save_profile(user_id, profile)
```

在意图分类时，如果模型置信度不高，就用用户的历史偏好做bias：

```python
def classify_with_bias(query: str, user_id: str) -> str:
    intent, confidence = model.predict(query)
    
    if confidence < 0.7:
        # 低置信度时用用户偏好
        profile = get_profile(user_id)
        if profile:
            return profile['preferred_intent']
    
    return intent
```

### 5.2 词汇演变

制造业的术语在变，老员工用旧称呼，新员工用新称呼。我们维护一个"术语演变表"：

```python
# 从query日志中自动发现同义表达
def discover_synonyms():
    # 同一个用户在一个session里先后问过：
    # "MC-2023" 和 "新加工中心"
    # 如果后续问"新加工中心"检索到的内容和"MC-2023"一致
    # 就把"新加工中心"加入MC-2023的同义词列表
    pass
```

同义词自动发现是个复杂的任务，目前还在探索中。初步方案是用embedding相似度聚类 + 人工审核。

## 第六章：效果数据

这套Query理解系统上线前后的数据对比：

| 指标 | 上线前 | 上线后 | 提升 |
|------|--------|--------|------|
| Hit@5 | 0.78 | 0.87 | +9% |
| 空回答率 | 14% | 9% | -5% |
| 负反馈率 | 18% | 12% | -6% |
| 平均延迟 | 520ms | 610ms | +90ms |
| Query理解模块贡献 | - | 6个点 | - |

增加90ms延迟换来6个百分点的Hit@5提升，这笔交易是值得的。