Digital_Human/llm.py

import time
import os
from basereal import BaseReal
from logger import logger
from knowledge_base import query_knowledge_b

def llm_response(message, nerfreal:BaseReal, knowledge_base_type=None, during_intro=False):
    start = time.perf_counter()
    logger.info(f"收到用户问题: {message}")
    logger.info(f"知识库类型: {knowledge_base_type}")
    logger.info(f"是否在介绍过程中: {during_intro}")
    
    # 首先尝试从知识库B获取答案
    kb_response, match_score = query_knowledge_b(message)
    
    # 检查是否从知识库B获得了有效答案（匹配度≥85%）
    if kb_response and match_score >= 0.85:
        logger.info(f"从知识库B获取答案，匹配度: {match_score:.2f}")
        logger.info(f"答案: {kb_response}")
        # 根据是否在介绍过程中决定使用哪种方法
        if during_intro:
            # 在介绍过程中被打断，使用专门的处理方法
            logger.info("在介绍过程中，使用handle_interruption_during_intro方法")
            # 传递during_intro参数，以便TTS在回答完毕后自动恢复
            nerfreal.handle_interruption_during_intro(kb_response, {'knowledge_base': knowledge_base_type, 'mode': 'qa', 'during_intro': during_intro})
        else:
            # 不是在介绍过程中，使用常规方法
            logger.info("不在介绍过程中，使用put_user_question方法")
            nerfreal.put_user_question(kb_response)  # 使用高优先级方法
        logger.info("知识库B答案已发送到播放队列")
        end = time.perf_counter()
        logger.info(f"问答处理总时间: {end-start:.2f}s")
        return
    else:
        logger.info(f"知识库B无匹配答案或匹配度不足，最高匹配度: {match_score:.2f}，使用LLM处理")
    
    # 如果知识库B没有匹配答案，使用LLM
    api_key = os.getenv("DASHSCOPE_API_KEY")
    if not api_key:
        # 如果没有配置API密钥，给出友好提示
        no_api_response = "抱歉，我无法找到相关答案。请配置DASHSCOPE_API_KEY环境变量以启用智能问答功能，或者尝试其他相关问题。"
        logger.info(f"未配置API密钥，返回提示信息: {no_api_response}")
        # 使用put_user_question而不是flush_talk，以保持一致性
        if during_intro:
            nerfreal.handle_interruption_during_intro(no_api_response, {'knowledge_base': knowledge_base_type, 'mode': 'qa'})
        else:
            nerfreal.put_user_question(no_api_response)
        end = time.perf_counter()
        logger.info(f"问答处理总时间: {end-start:.2f}s")
        
        return
    
    from openai import OpenAI
    client = OpenAI(
        # 如果您没有配置环境变量，请在此处用您的API Key进行替换
        api_key=api_key,
        # 填写DashScope SDK的base_url
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    )
    end = time.perf_counter()
    logger.info(f"llm Time init: {end-start}s")
    
    #构建系统提示词
    system_prompt = "You are a helpful assistant."
    if knowledge_base_type:
        if knowledge_base_type == "技术文档库":
            system_prompt = "您是技术专家助手，请基于技术文档库回答问题，提供准确的技术信息和解决方案。"
        elif knowledge_base_type == "生活百科库":
            system_prompt = "您是生活百科助手，请基于生活百科库回答问题，提供实用的生活建议和信息。"
        elif knowledge_base_type == "本地生活App":
            system_prompt = "您是本地生活App助手，请基于本地生活App知识库回答问题，提供关于美食外卖、休闲娱乐、生活服务、团购优惠等相关信息。"
    
    completion = client.chat.completions.create(
        model="qwen-plus",
        messages=[
            {'role': 'system', 'content': system_prompt},
            {'role': 'user', 'content': message}
        ],
        stream=True,
        # 通过以下设置，在流式输出的最后一行展示token使用信息
        stream_options={"include_usage": True}
    )
    result=""
    first = True
    for chunk in completion:
        if len(chunk.choices)>0:
            #print(chunk.choices[0].delta.content)
            if first:
                end = time.perf_counter()
                logger.info(f"llm Time to first chunk: {end-start}s")
                first = False
            msg = chunk.choices[0].delta.content
            if msg:
                result += msg
    end = time.perf_counter()
    logger.info(f"llm Time to last chunk: {end-start}s")
    logger.info(f"LLM完整回答: {result}")
    if during_intro:
        nerfreal.handle_interruption_during_intro(result, {'knowledge_base': knowledge_base_type, 'mode': 'qa'})
    else:
        nerfreal.put_user_question(result)  # 使用高优先级方法
    logger.info("LLM答案已发送到播放队列")
    logger.info(f"问答处理总时间: {end-start:.2f}s")