def keyword_based_intent_recognition(text: str, intent_keywords: dict) -> tuple:
    """
    基于关键词的意图识别
    :param text: 用户输入文本
    :param intent_keywords: 意图-关键词映射字典
    :return: 识别出的意图、置信度
    """
    # 统一转为小写，避免大小写干扰
    text_lower = text.lower()
    max_score = 0
    best_intent = "未知"
    
    # 遍历所有意图的关键词
    for intent, keywords in intent_keywords.items():
        score = 0
        for keyword in keywords:
            if keyword.lower() in text_lower:
                score += 1  # 每匹配一个关键词，得分+1
        
        # 更新最优意图
        if score > max_score and score > 0:
            max_score = score
            best_intent = intent
    
    # 计算置信度（归一化到0-1）
    confidence = min(max_score / 3.0, 1.0) if best_intent != "未知" else 0.0
    return best_intent, confidence
# 测试代码
if __name__ == "__main__":
    # 定义意图-关键词映射
    INTENT_KEYWORDS = {
        "问候": ["你好", "嗨", "早上好", "下午好", "晚上好"],
        "天气查询": ["天气", "气温", "温度", "下雨", "晴天"],
        "计算": ["计算", "加", "减", "乘", "除", "等于"]
    }
    
    # 测试用例
    test_cases = [
        "你好，今天天气怎么样？",
        "帮我算一下5乘6等于多少",
        "晚上好，外面冷不冷？"
    ]
    
    for case in test_cases:
        intent, conf = keyword_based_intent_recognition(case, INTENT_KEYWORDS)
        print(f"输入：{case} → 意图：{intent}（置信度：{conf:.2f}）")

from sentence_transformers import SentenceTransformer
from modelscope import snapshot_download
import torch
import torch.nn.functional as F
cache_dir = "D:\\modelscope\\hub"
embedding_model_dir = snapshot_download(
    model_id="sentence-transformers/all-MiniLM-L6-v2",
    cache_dir=cache_dir,
    revision="master"
)
# 初始化语义模型（轻量级，仅80MB）
embedding_model = SentenceTransformer(embedding_model_dir)
def semantic_based_intent_recognition(text: str, intent_examples: dict, top_k: int = 3) -> tuple:
    """
    基于语义向量的意图识别
    :param text: 用户输入文本
    :param intent_examples: 意图-示例句子映射字典
    :param top_k: 返回Top-K相似度结果
    :return: 最佳意图、置信度、Top-K相似度字典
    """
    # 空输入直接返回未知
    if not text.strip():
        return "未知", 0.0, {}
    
    # 1. 将用户输入转为向量
    text_embedding = embedding_model.encode([text], convert_to_tensor=True)
    
    # 2. 遍历所有意图，计算相似度
    intent_scores = {}
    for intent, examples in intent_examples.items():
        # 将该意图的所有示例转为向量
        example_embeddings = embedding_model.encode(examples, convert_to_tensor=True)
        # 计算用户输入与每个示例的余弦相似度
        similarities = F.cosine_similarity(text_embedding, example_embeddings, dim=1)
        # 取最大相似度作为该意图的得分
        intent_scores[intent] = similarities.max().item()
    
    # 3. 确定最佳意图
    best_intent = max(intent_scores.keys(), key=lambda x: intent_scores[x])
    best_score = intent_scores[best_intent]
    
    # 4. 整理Top-K结果
    sorted_scores = dict(sorted(intent_scores.items(), key=lambda x: x[1], reverse=True)[:top_k])
    
    return best_intent, best_score, sorted_scores
# 测试代码
if __name__ == "__main__":
    # 定义意图-示例句子映射
    INTENT_EXAMPLES = {
        "问候": ["你好啊", "大家好", "嗨", "早上好", "今天天气不错"],
        "天气查询": ["今天天气怎么样", "明天会下雨吗", "现在温度是多少", "外面要不要带伞"],
        "计算": ["帮我算一下1加1", "3乘5等于多少", "100减50是多少"]
    }
    
    # 测试用例（包含同义不同词的场景）
    test_cases = [
        "你好，今天外面冷不冷？",
        "帮我算一下8除以2等于几",
        "明天出门需要带伞吗？"
    ]
    
    for case in test_cases:
        intent, conf, top_scores = semantic_based_intent_recognition(case, INTENT_EXAMPLES)
        print(f"\n输入：{case}")
        print(f"最佳意图：{intent}（置信度：{conf:.3f}）")
        print(f"Top3相似度：{top_scores}")

from transformers import AutoTokenizer, AutoModelForCausalLM
from modelscope import snapshot_download
import torch
cache_dir = "D:\\modelscope\\hub"
# 加载大模型（Qwen1.5-1.8B-Chat，本地部署）
model_name = "qwen/Qwen1.5-1.8B-Chat"
local_model_path = snapshot_download(model_name, cache_dir=cache_dir)
tokenizer = AutoTokenizer.from_pretrained(local_model_path, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
    local_model_path,
    trust_remote_code=True,
    torch_dtype=torch.float16,
    device_map="auto"
)
model.eval()
def llm_based_intent_recognition(text: str, intent_list: list) -> tuple:
    """
    基于大模型的意图识别
    :param text: 用户输入文本
    :param intent_list: 可选意图列表
    :return: 识别出的意图、置信度
    """
    # 构造提示词（关键：明确指令，避免大模型输出多余内容）
    prompt = f"""用户输入：{text}
从以下意图中选择最匹配的一个：{', '.join(intent_list)}
答案："""
    
    # 调用大模型
    inputs = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=512).to(model.device)
    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=5,   # 减少输出长度，只需要意图名称
            temperature=0.0,    # 完全确定性生成
            do_sample=False,     # 关闭采样，保证结果可重复
            output_scores=True,   # 返回分数
            return_dict_in_generate=True,  # 返回字典格式
            pad_token_id=tokenizer.eos_token_id,
            repetition_penalty=1.1  # 减少重复
        )
    
    # 解析结果
    response = tokenizer.decode(outputs.sequences[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True).strip()
    
    # 调试：打印原始输出（包括所有token和概率）
    full_text = tokenizer.decode(outputs.sequences[0], skip_special_tokens=True)
    # print(f"  [调试] 完整文本（含输入）: '{full_text}'")
    print(f"  [调试] 模型原始输出: '{response}'")
    # print(f"  [调试] 完整生成序列: {outputs.sequences[0].tolist()}")
    print(f"  [调试] 输入长度: {inputs['input_ids'].shape[1]}")
    print(f"  [调试] 生成token数: {len(outputs.sequences[0]) - inputs['input_ids'].shape[1]}")
    
    # 提取匹配的意图（处理换行符、前缀词等问题）
    matched_intent = None
    for intent in intent_list:
        # 优先完全匹配或起始匹配
        if response == intent or response.startswith(intent):
            matched_intent = intent
            break
        # 其次包含匹配（去除多余前缀词）
        elif intent in response or response in intent:
            matched_intent = intent
            break
    
    if matched_intent:
        response = matched_intent
    
    # 验证结果是否在意图列表中
    if response in intent_list:
        # 计算实际置信度：基于输出的log概率
        with torch.no_grad():
            # 获取每个生成token的scores
            scores = outputs.scores  # tuple of tensors, each tensor is (batch_size, vocab_size)
            # 计算每个token的log概率
            log_probs = [torch.log_softmax(score, dim=-1) for score in scores]
            # 获取实际生成的token ids（不包括输入部分）
            generated_ids = outputs.sequences[0][inputs["input_ids"].shape[1]:]
            
            # 编码意图词为tokens
            intent_tokens = tokenizer.encode(response, add_special_tokens=False)
            
            # 找到意图词在生成序列中的起始位置
            intent_start_pos = -1
            for i in range(len(generated_ids) - len(intent_tokens) + 1):
                # 检查从位置i开始的子序列是否匹配意图词
                if generated_ids[i:i+len(intent_tokens)].tolist() == intent_tokens:
                    intent_start_pos = i
                    break
            
            print(f"  [调试] 生成的tokens长度 {len(outputs.sequences[0]) - inputs['input_ids'].shape[1]}: {generated_ids.tolist()}")
            print(f"  [调试] 意图词'{response}'的tokens: {intent_tokens}")
            print(f"  [调试] 意图词起始位置: {intent_start_pos}")
            
            # 提取意图词对应的token概率
            intent_token_probs = []
            
            if intent_start_pos >= 0:
                for i, token_id in enumerate(intent_tokens):
                    actual_pos = intent_start_pos + i
                    if actual_pos < len(log_probs):
                        prob = torch.exp(log_probs[actual_pos][0, token_id])
                        intent_token_probs.append(prob)
                        print(f"  [调试] token {i} (位置{actual_pos}): id={token_id}, prob={prob:.4f}")
                    else:
                        print(f"  [调试] token {i}: 超出log_probs范围")
            else:
                print(f"  [调试] 未找到意图词在生成序列中的位置")
            
            # 使用几何平均计算置信度
            if intent_token_probs:
                product = 1.0
                for p in intent_token_probs:
                    product *= p
                avg_confidence = float(product ** (1.0 / len(intent_token_probs)))
            else:
                avg_confidence = 0.0
        
        return response, avg_confidence
    else:
        return "未知", 0.0
# 测试代码
if __name__ == "__main__":
    INTENT_LIST = ["问候", "天气查询", "计算", "翻译", "告别"]
    
    test_cases = [
        "你好，帮我查一下明天的天气",
        "算一下10乘20等于多少",
        "把“你好世界”翻译成英文"
    ]
    
    for case in test_cases:
        intent, conf = llm_based_intent_recognition(case, INTENT_LIST)
        print(f"输入：{case} → 意图：{intent}（置信度：{conf:.2f}）")

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import uvicorn
app = FastAPI(title="意图识别系统")
# 跨域配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
# 数据模型
class IntentRequest(BaseModel):
    text: str
# API接口
@app.post("/api/classify-intent")
async def classify_intent(request: IntentRequest):
    try:
        if not request.text.strip():
            raise HTTPException(status_code=400, detail="输入文本不能为空")
        
        result = classify_intent_ensemble(request.text)
        return {"success": True, "data": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"识别失败：{str(e)}")
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)