智能客服与对话系统

# 对话系统基础架构
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import List, Optional
from enum import Enum

class DialogueAct(Enum):
    INFORM = "inform"
    REQUEST = "request"
    CONFIRM = "confirm"
    DENY = "deny"
    THANK = "thank"
    CLOSE = "close"

@dataclass
class DialogueState:
    intent: str = ""
    slots: dict = field(default_factory=dict)
    history: List[dict] = field(default_factory=list)
    confidence: float = 0.0
    turn_count: int = 0

@dataclass
class UserUtterance:
    text: str
    timestamp: float
    channel: str = "web"

@dataclass
class SystemResponse:
    text: str
    actions: List[str] = field(default_factory=list)
    suggestions: List[str] = field(default_factory=list)

class DialogueSystem(ABC):
    @abstractmethod
    def nlu(self, utterance: UserUtterance) -> dict:
        pass

    @abstractmethod
    def dialogue_management(self, state: DialogueState, nlu_result: dict) -> DialogueState:
        pass

    @abstractmethod
    def nlg(self, state: DialogueState) -> SystemResponse:
        pass

    def process(self, utterance: UserUtterance, state: DialogueState) -> SystemResponse:
        nlu_result = self.nlu(utterance)
        new_state = self.dialogue_management(state, nlu_result)
        return self.nlg(new_state)

# 对话系统评估指标
import numpy as np
from collections import defaultdict

def task_success_rate(completed_tasks: int, total_tasks: int) -> float:
    """任务完成率：成功完成的任务占总任务的比例"""
    return completed_tasks / total_tasks if total_tasks > 0 else 0.0

def slot_filling_f1(predicted: dict, ground_truth: dict) -> dict:
    """槽位填充的精确率、召回率和 F1 分数"""
    tp = sum(1 for k, v in predicted.items()
             if k in ground_truth and ground_truth[k] == v)
    fp = sum(1 for k, v in predicted.items()
             if k not in ground_truth or ground_truth[k] != v)
    fn = sum(1 for k, v in ground_truth.items()
             if k not in predicted or predicted[k] != v)

    precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0
    recall = tp / (tp + fn) if (tp + fn) > 0 else 0.0
    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0.0
    return {"precision": precision, "recall": recall, "f1": f1}

def dialogue_turn_efficiency(turns: List[int]) -> dict:
    """对话效率指标：完成任务的平均轮数和方差"""
    arr = np.array(turns)
    return {"mean_turns": float(arr.mean()), "std_turns": float(arr.std()), "median": float(np.median(arr))}

# 规则引擎意图识别
import re
from typing import Tuple, Optional

class RuleIntentClassifier:
    def __init__(self):
        self.rules = []

    def add_rule(self, intent: str, patterns: list, priority: int = 0):
        compiled = [re.compile(p, re.IGNORECASE) for p in patterns]
        self.rules.append((intent, compiled, priority))
        self.rules.sort(key=lambda x: -x[2])

    def classify(self, text: str) -> Tuple[str, float]:
        for intent, patterns, priority in self.rules:
            for pattern in patterns:
                if pattern.search(text):
                    confidence = min(0.95, 0.5 + priority * 0.1)
                    return intent, confidence
        return "unknown", 0.0

    def extract_slots(self, text: str, slot_patterns: dict) -> dict:
        """从文本中提取槽位信息"""
        slots = {}
        for slot_name, pattern in slot_patterns.items():
            match = re.search(pattern, text)
            if match:
                slots[slot_name] = match.group(1) if match.groups() else match.group(0)
        return slots

# 使用示例
classifier = RuleIntentClassifier()
classifier.add_rule("check_order", [r"查询.*订单", r"订单.*状态", r"我的订单"], priority=3)
classifier.add_rule("refund", [r"退款", r"退货", r"取消订单"], priority=2)
slot_patterns = {"order_id": r"订单号[为是]*(\S+)"}
print(classifier.extract_slots("查询订单号为ORD123456的状态", slot_patterns))

# 基于 BERT 的意图识别
import torch
import torch.nn as nn
from transformers import BertModel, BertTokenizer

class BertIntentClassifier(nn.Module):
    def __init__(self, num_intents: int, model_name: str = "bert-base-chinese"):
        super().__init__()
        self.bert = BertModel.from_pretrained(model_name)
        self.dropout = nn.Dropout(0.3)
        self.classifier = nn.Linear(self.bert.config.hidden_size, num_intents)

    def forward(self, input_ids: torch.Tensor,
                attention_mask: torch.Tensor) -> torch.Tensor:
        outputs = self.bert(input_ids=input_ids, attention_mask=attention_mask)
        pooled = outputs.pooler_output
        pooled = self.dropout(pooled)
        logits = self.classifier(pooled)
        return logits

    def predict(self, text: str, tokenizer: BertTokenizer,
                device: str = "cpu") -> Tuple[str, float]:
        inputs = tokenizer(text, return_tensors="pt", padding=True,
                          truncation=True, max_length=128)
        inputs = {k: v.to(device) for k, v in inputs.items()}
        self.eval()
        with torch.no_grad():
            logits = self.forward(**inputs)
            probs = torch.softmax(logits, dim=-1)
            conf, pred = torch.max(probs, dim=-1)
        return str(pred.item()), float(conf.item())

# 序列标注：槽位填充（BiLSTM-CRF）
class BiLSTMCRF(nn.Module):
    def __init__(self, vocab_size: int, embed_dim: int, hidden_dim: int,
                 num_tags: int):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.lstm = nn.LSTM(embed_dim, hidden_dim, bidirectional=True,
                           batch_first=True)
        self.hidden2tag = nn.Linear(hidden_dim * 2, num_tags)
        self.crf = nn.Linear(num_tags, num_tags)

    def forward(self, input_ids: torch.Tensor) -> torch.Tensor:
        embeds = self.embedding(input_ids)
        lstm_out, _ = self.lstm(embeds)
        emissions = self.hidden2tag(lstm_out)
        return emissions

# BM25 检索式对话
import math
from collections import Counter, defaultdict
from typing import List, Tuple

class BM25Retriever:
    def __init__(self, k1: float = 1.5, b: float = 0.75):
        self.k1 = k1
        self.b = b
        self.corpus = []
        self.doc_freq = defaultdict(int)
        self.doc_lengths = []
        self.avg_doc_length = 0.0

    def build(self, documents: List[str]):
        self.corpus = [doc.lower().split() for doc in documents]
        self.doc_lengths = [len(doc) for doc in self.corpus]
        self.avg_doc_length = sum(self.doc_lengths) / len(self.corpus) if self.corpus else 0

        for doc in self.corpus:
            for term in set(doc):
                self.doc_freq[term] += 1

    def score(self, query: str, doc_idx: int) -> float:
        query_terms = query.lower().split()
        doc = self.corpus[doc_idx]
        doc_len = self.doc_lengths[doc_idx]
        N = len(self.corpus)

        total_score = 0.0
        term_counts = Counter(doc)

        for term in query_terms:
            if term not in self.doc_freq or self.doc_freq[term] == 0:
                continue
            tf = term_counts.get(term, 0)
            idf = math.log((N - self.doc_freq[term] + 0.5) /
                          (self.doc_freq[term] + 0.5) + 1.0)
            numerator = tf * (self.k1 + 1)
            denominator = tf + self.k1 * (
                1 - self.b + self.b * doc_len / self.avg_doc_length
            )
            total_score += idf * numerator / denominator

        return total_score

    def retrieve(self, query: str, top_k: int = 5) -> List[Tuple[int, float]]:
        scores = [(i, self.score(query, i)) for i in range(len(self.corpus))]
        return sorted(scores, key=lambda x: -x[1])[:top_k]

# 双塔向量检索对话
import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import BertModel

class DualTowerRetriever(nn.Module):
    def __init__(self, model_name: str = "bert-base-chinese",
                 temperature: float = 0.07):
        super().__init__()
        self.encoder = BertModel.from_pretrained(model_name)
        self.temperature = temperature
        self.projection = nn.Linear(768, 256)

    def encode(self, input_ids: torch.Tensor,
               attention_mask: torch.Tensor) -> torch.Tensor:
        outputs = self.encoder(input_ids, attention_mask=attention_mask)
        cls = outputs.last_hidden_state[:, 0, :]
        return F.normalize(self.projection(cls), p=2, dim=-1)

    def forward(self, query_ids: torch.Tensor, query_mask: torch.Tensor,
                doc_ids: torch.Tensor, doc_mask: torch.Tensor) -> torch.Tensor:
        q_emb = self.encode(query_ids, query_mask)
        d_emb = self.encode(doc_ids, doc_mask)
        return torch.sum(q_emb * d_emb, dim=-1) / self.temperature

    def retrieve(self, query_embedding: torch.Tensor,
                 doc_embeddings: torch.Tensor,
                 top_k: int = 5) -> Tuple[List[int], List[float]]:
        scores = query_embedding @ doc_embeddings.T
        top_scores, top_indices = torch.topk(scores, top_k)
        return top_indices.tolist(), top_scores.tolist()

# 构建索引并检索
class FAQRetrievalSystem:
    def __init__(self, model: DualTowerRetriever, faq_pairs: List[Tuple[str, str]]):
        self.model = model
        self.questions = [q for q, _ in faq_pairs]
        self.answers = [a for _, a in faq_pairs]

    def build_index(self, tokenizer, device: str = "cpu"):
        self.model.eval()
        with torch.no_grad():
            tokens = tokenizer(self.questions, padding=True,
                              return_tensors="pt", truncation=True, max_length=64)
            self.doc_embeddings = self.model.encode(
                tokens["input_ids"].to(device),
                tokens["attention_mask"].to(device)
            )

    def search(self, query: str, tokenizer, top_k: int = 3) -> List[str]:
        self.model.eval()
        with torch.no_grad():
            tokens = tokenizer(query, return_tensors="pt",
                              truncation=True, max_length=64)
            q_emb = self.model.encode(tokens["input_ids"], tokens["attention_mask"])
            indices, scores = self.model.retrieve(q_emb, self.doc_embeddings, top_k)
        return [self.answers[i] for i in indices]

# Seq2Seq 对话模型（PyTorch）
import torch
import torch.nn as nn
import torch.nn.functional as F

class Encoder(nn.Module):
    def __init__(self, vocab_size: int, embed_dim: int, hidden_dim: int,
                 n_layers: int = 2, dropout: float = 0.3):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0)
        self.lstm = nn.LSTM(embed_dim, hidden_dim, n_layers,
                           bidirectional=True, dropout=dropout, batch_first=True)
        self.fc = nn.Linear(hidden_dim * 2, hidden_dim)

    def forward(self, src: torch.Tensor) -> tuple:
        embedded = self.embedding(src)
        outputs, (hidden, cell) = self.lstm(embedded)
        hidden = torch.tanh(self.fc(torch.cat((hidden[-2], hidden[-1]), dim=-1)))
        cell = torch.tanh(self.fc(torch.cat((cell[-2], cell[-1]), dim=-1)))
        return outputs, hidden, cell

class Decoder(nn.Module):
    def __init__(self, vocab_size: int, embed_dim: int, hidden_dim: int,
                 n_layers: int = 2, dropout: float = 0.3):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0)
        self.lstm = nn.LSTM(embed_dim, hidden_dim, n_layers,
                           dropout=dropout, batch_first=True)
        self.fc_out = nn.Linear(hidden_dim, vocab_size)
        self.dropout = nn.Dropout(dropout)

    def forward(self, input: torch.Tensor, hidden: torch.Tensor,
                cell: torch.Tensor) -> tuple:
        input = input.unsqueeze(1)
        embedded = self.dropout(self.embedding(input))
        output, (hidden, cell) = self.lstm(embedded, (hidden, cell))
        prediction = self.fc_out(output.squeeze(1))
        return prediction, hidden, cell

class Seq2SeqDialogue(nn.Module):
    def __init__(self, encoder: Encoder, decoder: Decoder, device: str):
        super().__init__()
        self.encoder = encoder
        self.decoder = decoder
        self.device = device

    def forward(self, src: torch.Tensor, trg: torch.Tensor,
                teacher_forcing_ratio: float = 0.5) -> torch.Tensor:
        batch_size = src.shape[0]
        trg_len = trg.shape[1]
        trg_vocab_size = self.decoder.fc_out.out_features
        outputs = torch.zeros(batch_size, trg_len, trg_vocab_size).to(self.device)
        enc_outputs, hidden, cell = self.encoder(src)

        decoder_input = trg[:, 0]
        for t in range(1, trg_len):
            output, hidden, cell = self.decoder(decoder_input, hidden, cell)
            outputs[:, t] = output
            teacher_force = torch.rand(1).item() < teacher_forcing_ratio
            top1 = output.argmax(1)
            decoder_input = trg[:, t] if teacher_force else top1

        return outputs

# 多样化解码策略
import torch
import torch.nn.functional as F

class DialogueGenerator:
    def __init__(self, model, tokenizer, device: str = "cpu"):
        self.model = model
        self.tokenizer = tokenizer
        self.device = device

    def greedy_decode(self, input_ids: torch.Tensor,
                      max_length: int = 50) -> str:
        """贪心解码：每步选择概率最高的 token，容易产生通用回复"""
        self.model.eval()
        generated = input_ids.clone()
        with torch.no_grad():
            for _ in range(max_length):
                outputs = self.model(generated)
                next_token = outputs[:, -1].argmax(dim=-1, keepdim=True)
                generated = torch.cat([generated, next_token], dim=-1)
                if next_token.item() == self.tokenizer.eos_token_id:
                    break
        return self.tokenizer.decode(generated[0], skip_special_tokens=True)

    def nucleus_sampling(self, input_ids: torch.Tensor,
                         max_length: int = 50, top_p: float = 0.9,
                         temperature: float = 0.8) -> str:
        """Nucleus Sampling：从累积概率达到 top_p 的最小 token 集合中采样"""
        self.model.eval()
        generated = input_ids.clone()
        with torch.no_grad():
            for _ in range(max_length):
                outputs = self.model(generated)
                logits = outputs[:, -1] / temperature
                probs = F.softmax(logits, dim=-1)
                sorted_probs, sorted_indices = torch.sort(probs, descending=True)
                cumulative_probs = torch.cumsum(sorted_probs, dim=-1)
                nucleus_mask = cumulative_probs <= top_p
                nucleus_mask[..., 1:] = nucleus_mask[..., :-1].clone()
                nucleus_mask[..., 0] = True
                filtered_probs = sorted_probs * nucleus_mask
                filtered_probs /= filtered_probs.sum()
                next_token = torch.multinomial(filtered_probs, 1)
                next_token = torch.gather(sorted_indices, -1, next_token)
                generated = torch.cat([generated, next_token], dim=-1)
                if next_token.item() == self.tokenizer.eos_token_id:
                    break
        return self.tokenizer.decode(generated[0], skip_special_tokens=True)

# RAG 文档处理与检索
import hashlib
from typing import List, Tuple
from dataclasses import dataclass

@dataclass
class Document:
    content: str
    metadata: dict
    embedding: List[float] = None

class RAGDocumentProcessor:
    def __init__(self, embed_fn, chunk_size: int = 500,
                 chunk_overlap: int = 50):
        self.embed_fn = embed_fn
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

    def chunk_document(self, text: str) -> List[str]:
        """按固定长度重叠切分文档"""
        words = text.split()
        chunks = []
        start = 0
        while start < len(words):
            end = start + self.chunk_size
            chunk = " ".join(words[start:end])
            chunks.append(chunk)
            start = end - self.chunk_overlap
        return chunks

    def process(self, documents: List[Tuple[str, dict]]) -> List[Document]:
        """处理文档：切分 + 向量化"""
        processed = []
        for text, meta in documents:
            chunks = self.chunk_document(text)
            for i, chunk in enumerate(chunks):
                doc_id = hashlib.md5(f"{meta.get('source', '')}-{i}".encode()).hexdigest()
                embedding = self.embed_fn(chunk)
                processed.append(Document(
                    content=chunk,
                    metadata={**meta, "chunk_id": doc_id, "chunk_index": i},
                    embedding=embedding
                ))
        return processed

class RAGRetriever:
    def __init__(self, index, documents: List[Document], top_k: int = 3):
        self.index = index
        self.documents = documents
        self.top_k = top_k

    def search(self, query_embedding: List[float]) -> List[Document]:
        import numpy as np
        query_vec = np.array([query_embedding], dtype=np.float32)
        distances, indices = self.index.search(query_vec, self.top_k)
        results = []
        for dist, idx in zip(distances[0], indices[0]):
            if idx != -1:
                doc = self.documents[idx]
                results.append(doc)
        return results

# RAG 生成管道
from typing import List, Optional

class RAGPipeline:
    def __init__(self, retriever: RAGRetriever, llm,
                 embed_fn, prompt_template: str = None):
        self.retriever = retriever
        self.llm = llm
        self.embed_fn = embed_fn
        self.prompt_template = prompt_template or self._default_prompt()

    def _default_prompt(self) -> str:
        return (
            "你是一个专业的客服助手。请根据以下参考资料回答用户问题。\n"
            "如果参考资料中没有相关信息，请如实告知用户，不要编造答案。\n\n"
            "参考资料：\n{context}\n\n"
            "用户问题：{question}\n\n"
            "回答："
        )

    def query(self, question: str,
              filter_fn=None) -> dict:
        """完整的 RAG 查询流程"""
        # 1. 向量化查询
        query_embedding = self.embed_fn(question)

        # 2. 检索相关文档
        docs = self.retriever.search(query_embedding)

        # 3. 可选的后过滤
        if filter_fn:
            docs = [d for d in docs if filter_fn(d)]

        # 4. 构建上下文
        context = "\n---\n".join(
            [f"[{d.metadata.get('source', 'Unknown')}] {d.content}" for d in docs]
        )

        # 5. 生成回复
        prompt = self.prompt_template.format(context=context, question=question)
        response = self.llm.generate(prompt)

        return {
            "answer": response,
            "sources": [d.metadata for d in docs],
            "context": context
        }

    def batch_query(self, questions: List[str]) -> List[dict]:
        """批量查询"""
        return [self.query(q) for q in questions]

# LLM Function Calling 客服系统
import json
from typing import List, Dict, Any

# 定义工具（Function Schema）
CUSTOMER_SERVICE_TOOLS = [
    {
        "name": "query_order",
        "description": "查询订单状态和物流信息",
        "parameters": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string", "description": "订单号"},
                "phone": {"type": "string", "description": "收货手机号"}
            },
            "required": ["order_id"]
        }
    },
    {
        "name": "process_refund",
        "description": "处理退款申请",
        "parameters": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string", "description": "订单号"},
                "reason": {"type": "string", "description": "退款原因"},
                "amount": {"type": "number", "description": "退款金额"}
            },
            "required": ["order_id", "reason"]
        }
    },
    {
        "name": "check_warranty",
        "description": "查询商品保修状态",
        "parameters": {
            "type": "object",
            "properties": {
                "product_id": {"type": "string", "description": "商品编号"},
                "purchase_date": {"type": "string", "description": "购买日期"}
            },
            "required": ["product_id"]
        }
    }
]

# 工具执行函数
def execute_tool(tool_name: str, args: Dict[str, Any]) -> str:
    tools_map = {
        "query_order": lambda a: {"status": "已发货", "tracking": "SF1234567890", "eta": "2026-04-15"},
        "process_refund": lambda a: {"refund_id": "RF001", "status": "处理中", "eta": "3-5工作日"},
        "check_warranty": lambda a: {"warranty_status": "在保", "expires": "2027-04-12"}
    }
    if tool_name not in tools_map:
        return json.dumps({"error": "未知工具"})
    return json.dumps(tools_map[tool_name](args), ensure_ascii=False)

class LLMCustomerService:
    def __init__(self, llm_client):
        self.llm = llm_client
        self.tools = CUSTOMER_SERVICE_TOOLS
        self.system_prompt = "你是专业电商客服，使用工具查询信息后回答用户。"

    def handle(self, user_message: str, history: List[dict] = None) -> str:
        messages = [{"role": "system", "content": self.system_prompt}]
        if history:
            messages.extend(history)
        messages.append({"role": "user", "content": user_message})

        response = self.llm.chat(messages, tools=self.tools)

        if response.get("tool_calls"):
            for tool_call in response["tool_calls"]:
                tool_name = tool_call["function"]["name"]
                args = json.loads(tool_call["function"]["arguments"])
                result = execute_tool(tool_name, args)
                messages.append(response)
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call["id"],
                    "content": result
                })
            final_response = self.llm.chat(messages, tools=self.tools)
            return final_response.get("content", "抱歉，暂时无法处理")
        return response.get("content", "")

# LLM 路由与分层成本优化
import hashlib
import time
from typing import Dict, Optional

class CacheEntry:
    def __init__(self, response: str, ttl: int = 3600):
        self.response = response
        self.expires_at = time.time() + ttl

    def is_valid(self) -> bool:
        return time.time() < self.expires_at

class LLMRouter:
    """根据问题复杂度路由到不同模型，优化成本"""
    def __init__(self, fast_model, smart_model, cache_ttl: int = 3600):
        self.fast_model = fast_model    # 便宜小模型（如 GPT-4o-mini）
        self.smart_model = smart_model  # 强大模型（如 GPT-4o）
        self.cache: Dict[str, CacheEntry] = {}
        self.cache_ttl = cache_ttl

    def _cache_key(self, message: str) -> str:
        return hashlib.md5(message.encode()).hexdigest()

    def route(self, message: str, history: list = None) -> str:
        # 1. 检查缓存
        key = self._cache_key(message)
        if key in self.cache and self.cache[key].is_valid():
            return self.cache[key].response

        # 2. 用轻量模型判断复杂度
        classification = self._classify_complexity(message)

        # 3. 根据复杂度选择模型
        if classification == "simple":
            response = self.fast_model.generate(message)
        else:
            response = self.smart_model.generate(message)

        # 4. 缓存结果
        self.cache[key] = CacheEntry(response, self.cache_ttl)
        return response

    def _classify_complexity(self, message: str) -> str:
        """简单分类：关键词判断 + 轻量模型辅助"""
        simple_patterns = ["你好", "谢谢", "再见", "在吗", "营业时间", "地址在哪"]
        for pattern in simple_patterns:
            if pattern in message:
                return "simple"
        if len(message) < 30:
            return "simple"
        return "complex"

    def get_stats(self) -> dict:
        total = len(self.cache)
        valid = sum(1 for e in self.cache.values() if e.is_valid())
        return {"cache_total": total, "cache_valid": valid, "hit_rate": valid / max(total, 1)}

# LangChain 客服系统 - 核心架构
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from langchain_core.runnables import RunnablePassthrough
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.memory import ConversationBufferMemory
from langchain.schema import Document
from typing import List
import json

# 定义业务工具
@tool
def query_order(order_id: str) -> str:
    """查询订单状态，输入为订单号。"""
    orders = {
        "ORD001": {"status": "已发货", "tracking": "SF12345", "items": ["无线耳机"]},
        "ORD002": {"status": "配送中", "tracking": "YT67890", "items": ["智能手表"]},
    }
    order = orders.get(order_id, {"status": "未找到", "tracking": "", "items": []})
    return json.dumps(order, ensure_ascii=False)

@tool
def check_return_policy(product_type: str) -> str:
    """查询退货政策，输入为商品类型。"""
    policies = {
        "电子产品": "7天无理由退货，需保持包装完整",
        "服装": "30天无理由退货，吊牌未拆",
        "食品": "不支持无理由退货，质量问题可协商",
        "default": "请咨询具体品类的退货政策"
    }
    return policies.get(product_type, policies["default"])

# 构建 RAG 检索器
class FAQRetriever:
    def __init__(self, faq_data: List[dict], embed_fn):
        self.embed_fn = embed_fn
        self.docs = []
        for faq in faq_data:
            doc = Document(
                page_content=f"问题: {faq['question']}\n答案: {faq['answer']}",
                metadata={"source": "FAQ", "category": faq.get("category", "general")}
            )
            self.docs.append(doc)

    def invoke(self, query: str) -> List[Document]:
        query_emb = self.embed_fn(query)
        scored = []
        for doc in self.docs:
            doc_emb = self.embed_fn(doc.page_content)
            score = sum(a * b for a, b in zip(query_emb, doc_emb))
            scored.append((score, doc))
        scored.sort(key=lambda x: -x[0])
        return [doc for _, doc in scored[:3]]

# 构建 Agent
def build_agent(llm, faq_retriever, embed_fn):
    tools = [query_order, check_return_policy]

    system_template = """你是专业电商客服助手。你可以：
1. 使用 query_order 工具查询订单
2. 使用 check_return_policy 工具查询退货政策
3. 基于参考资料回答产品相关问题

如果以上都无法回答，请礼貌地告知用户并建议转人工客服。

参考资料：
{context}"""

    prompt = ChatPromptTemplate.from_messages([
        ("system", system_template),
        MessagesPlaceholder(variable_name="chat_history"),
        ("human", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ])

    def format_docs(docs: List[Document]) -> str:
        return "\n\n".join(doc.page_content for doc in docs)

    rag_chain = (
        {"context": faq_retriever | format_docs, "input": RunnablePassthrough()}
    )

    agent = create_tool_calling_agent(llm, tools, prompt)
    return AgentExecutor(
        agent=agent, tools=tools, verbose=True,
        handle_parsing_errors=True,
        max_iterations=5
    )

# LangChain 客服系统 - 多轮对话与部署
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
import logging
import time

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("customer_service")

class CustomerServiceSystem:
    def __init__(self, agent_executor, max_history: int = 10):
        self.agent = agent_executor
        self.sessions = {}
        self.max_history = max_history

    def get_session_history(self, session_id: str):
        if session_id not in self.sessions:
            self.sessions[session_id] = InMemoryChatMessageHistory()
        return self.sessions[session_id]

    def chat(self, session_id: str, user_input: str) -> str:
        start_time = time.time()

        try:
            history = self.get_session_history(session_id)

            # 限制历史长度
            messages = history.messages
            if len(messages) > self.max_history * 2:
                history.clear()
                for msg in messages[-self.max_history:]:
                    history.add_message(msg)

            response = self.agent.invoke({
                "input": user_input,
                "chat_history": messages
            })

            elapsed = time.time() - start_time
            logger.info(
                f"Session {session_id}: {elapsed:.2f}s - "
                f"Input: {user_input[:50]}..."
            )

            return response.get("output", "抱歉，我暂时无法回答您的问题。")

        except Exception as e:
            logger.error(f"Session {session_id} error: {str(e)}")
            return "系统暂时遇到问题，请稍后再试或转人工客服。"

    def get_session_stats(self, session_id: str) -> dict:
        history = self.get_session_history(session_id)
        return {
            "session_id": session_id,
            "turn_count": len(history.messages) // 2,
            "is_active": len(history.messages) > 0
        }

# 使用示例
def main():
    # 初始化系统（假设已配置好 LLM 和检索器）
    faq_data = [
        {"question": "如何退货？", "answer": "可在订单页面申请退货", "category": "售后"},
        {"question": "运费怎么算？", "answer": "满99元包邮，不足收取6元", "category": "物流"},
        {"question": "支持哪些支付方式？", "answer": "支持支付宝、微信、银行卡", "category": "支付"},
    ]
    retriever = FAQRetriever(faq_data, embed_fn=lambda x: [0.0] * 768)
    agent = build_agent(llm=None, faq_retriever=retriever, embed_fn=lambda x: [0.0] * 768)
    system = CustomerServiceSystem(agent)

    # 模拟多轮对话
    session = "user_001"
    while True:
        user_input = input("用户: ")
        if user_input.lower() in ["退出", "quit", "bye"]:
            break
        response = system.chat(session, user_input)
        print(f"客服: {response}")
        stats = system.get_session_stats(session)
        print(f"  [对话轮数: {stats['turn_count']}]")