大模型应用：马尔可夫链\HMM与混元大模型的融合：经典序列算法+语义理解.109

import numpy as np
import matplotlib.pyplot as plt
import graphviz
from openai import OpenAI

# ---------------------- 第一步：定义马尔可夫链核心参数 ----------------------
# 1. 状态空间与映射（便于可视化和理解）
state_names = {
    0: "初始态",
    1: "咨询商品",
    2: "询问价格",
    3: "投诉售后",
    4: "下单",
    5: "结束态"
}
num_states = len(state_names)

# 2. 初始状态分布（初始态概率为1，其他为0）
pi = np.array([1.0, 0.0, 0.0, 0.0, 0.0, 0.0])

# 3. 状态转移矩阵（基于业务经验定义）
# 行：当前状态，列：下一状态，值：转移概率
transition_matrix = np.array([
    [0.0, 0.8, 0.0, 0.1, 0.0, 0.1],  # 初始态→咨询(0.8)、投诉(0.1)、结束(0.1)
    [0.0, 0.2, 0.7, 0.0, 0.0, 0.1],  # 咨询商品→继续咨询(0.2)、问价格(0.7)、结束(0.1)
    [0.0, 0.1, 0.1, 0.0, 0.6, 0.2],  # 询问价格→咨询(0.1)、继续问价(0.1)、下单(0.6)、结束(0.2)
    [0.0, 0.0, 0.0, 0.5, 0.0, 0.5],  # 投诉售后→继续投诉(0.5)、结束(0.5)
    [0.0, 0.0, 0.0, 0.0, 0.1, 0.9],  # 下单→继续下单(0.1)、结束(0.9)
    [0.0, 0.0, 0.0, 0.0, 0.0, 1.0]   # 结束态→吸收态，不跳转
])

# ---------------------- 第二步：可视化马尔可夫链状态转移 ----------------------
def plot_markov_chain():
    """生成马尔可夫链状态转移图"""
    import os

    # 尝试自动找到 Graphviz 安装路径（常见路径）
    possible_paths = [
        r"D:\Program Files\Graphviz\bin"
    ]

    # 设置 Graphviz 可执行文件路径
    for path in possible_paths:
        if os.path.exists(path):
            os.environ["PATH"] += os.pathsep + path
            break

    # 创建有向图，指定支持中文的字体
    dot = graphviz.Digraph(comment='对话状态马尔可夫链',
                          graph_attr={'fontname': 'SimHei'},
                          node_attr={'fontname': 'SimHei'},
                          edge_attr={'fontname': 'SimHei'})
    # 添加节点
    for state_id, state_name in state_names.items():
        dot.node(str(state_id), state_name)
    # 添加边（只显示概率>0的转移）
    for i in range(num_states):
        for j in range(num_states):
            prob = transition_matrix[i][j]
            if prob > 0:
                dot.edge(str(i), str(j), label=f"{prob:.1f}")
    # 保存并渲染图片（需要安装graphviz软件，或注释此行，仅生成源码）
    dot.render('markov_chain_dialogue', format='png', cleanup=True)
    print("马尔可夫链状态转移图已保存为 markov_chain_dialogue.png")
    return dot

# 生成状态转移图（需要安装graphviz软件，否则注释掉下面这行）
plot_markov_chain()

# ---------------------- 第三步：大模型语义解析（腾讯混元为例） ----------------------
import json
# 初始化腾讯混元客户端（替换为你的API Key）
client = OpenAI(
    api_key="sk-bWlJ**************************************8Ze",
    base_url="https://api.hunyuan.cloud.tencent.com/v1"
)

def semantic_analysis(user_input):
    """
    大模型语义解析：输入用户文本，输出核心意图和情感
    参数：user_input - 用户输入的文本
    返回：dict - 包含intent（意图）、emotion（情感）、entity（实体）
    """
    prompt = f"""你是语义解析助手。分析用户输入，只返回纯JSON格式，不要任何其他文字。

用户输入：{user_input}

要求：
1. 核心意图只能选：咨询商品、询问价格、投诉售后、下单、无意图
2. 情感倾向：积极/中性/消极
3. 关键实体：提取商品名称或问题描述

返回格式：
{{"intent": "", "emotion": "", "entity": ""}}"""
    try:
        response = client.chat.completions.create(
            model="hunyuan-lite",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.0
        )
        # 提取纯JSON内容
        content = response.choices[0].message.content.strip()

        # 尝试直接解析
        try:
            result = json.loads(content)
        except:
            # 如果失败，尝试提取JSON部分
            import re
            json_match = re.search(r'\{.*\}', content, re.DOTALL)
            if json_match:
                result = json.loads(json_match.group())
            else:
                raise Exception("无法解析JSON")

        # 验证必要字段
        if "intent" not in result:
            result["intent"] = "无意图"
        if "emotion" not in result:
            result["emotion"] = "中性"
        if "entity" not in result:
            result["entity"] = ""

        return result
    except Exception as e:
        # 异常处理：返回默认值
        print(f"语义解析出错：{e}")
        print(f"模型返回内容：{content}")
        return {"intent": "无意图", "emotion": "中性", "entity": ""}

# ---------------------- 第四步：状态-语义匹配与转移 ----------------------
# 意图-状态映射表
intent_to_state = {
    "咨询商品": 1,
    "询问价格": 2,
    "投诉售后": 3,
    "下单": 4,
    "无意图": 0
}

def predict_next_state(current_state, semantic_result):
    """
    基于当前状态和语义解析结果，预测下一步状态
    参数：
        current_state - 当前状态ID
        semantic_result - 大模型语义解析结果
    返回：next_state - 下一步状态ID
    """
    # 1. 获取当前状态允许的转移概率分布
    allowed_probs = transition_matrix[current_state]
    # 2. 从语义解析结果中获取意图对应的目标状态
    intent = semantic_result["intent"]
    target_state = intent_to_state.get(intent, 5)  # 默认跳转到结束态
    
    # 3. 检查目标状态是否在允许的转移范围内（概率>0）
    if allowed_probs[target_state] > 0:
        next_state = target_state
    else:
        # 若不允许，则选择允许范围内概率最大的状态
        next_state = np.argmax(allowed_probs)
    
    # 4. 动态调整转移概率（比如消极情感提升投诉状态概率）
    if semantic_result["emotion"] == "消极":
        # 若情感消极，将投诉状态（3）的概率提升0.2
        if allowed_probs[3] > 0:
            allowed_probs[3] = min(allowed_probs[3] + 0.2, 1.0)
            # 归一化概率（保证和为1）
            allowed_probs = allowed_probs / allowed_probs.sum()
            next_state = np.argmax(allowed_probs)
    
    return next_state

# ---------------------- 第五步：大模型生成符合状态的回复 ----------------------
def generate_response(next_state, semantic_result):
    """
    基于目标状态和语义解析结果，生成回复
    参数：
        next_state - 下一步状态ID
        semantic_result - 大模型语义解析结果
    返回：response - 生成的回复文本
    """
    state_name = state_names[next_state]
    intent = semantic_result["intent"]
    entity = semantic_result["entity"]
    
    prompt = f"""
    你是专业的电商客服，需要根据以下约束生成回复：
    当前目标状态：{state_name}
    用户核心意图：{intent}
    关键实体：{entity}
    约束条件：
    1. 回复必须符合{state_name}的核心目标；
    2. 语气友好、专业，长度不超过60字；
    3. 禁止偏离当前状态（比如状态是询问价格，就只聊价格相关）。
    请生成回复：
    """
    try:
        response = client.chat.completions.create(
            model="hunyuan-lite",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7  # 适度随机性，保持自然
        )
        return response.choices[0].message.content.strip()
    except Exception as e:
        print(f"生成回复出错：{e}")
        return f"非常抱歉，我暂时无法回复你的问题。"

# ---------------------- 第六步：完整对话流程演示 ----------------------
def run_dialogue_demo():
    """运行多轮对话演示"""
    current_state = 0  # 初始状态为0（初始态）
    print("===== 电商客服对话演示 =====")
    print(f"初始状态：{state_names[current_state]}")
    
    while current_state != 5:  # 未到结束态则继续
        # 1. 获取用户输入（模拟）
        user_input = input("\n请输入你的问题/需求：")
        if not user_input:
            print("输入不能为空，请重新输入！")
            continue
        
        # 2. 大模型语义解析
        semantic_result = semantic_analysis(user_input)
        print(f"语义解析结果：{semantic_result}")
        
        # 3. 预测下一步状态
        next_state = predict_next_state(current_state, semantic_result)
        print(f"当前状态：{state_names[current_state]} → 下一步状态：{state_names[next_state]}")
        
        # 4. 生成回复
        response = generate_response(next_state, semantic_result)
        print(f"客服回复：{response}")
        
        # 5. 更新当前状态
        current_state = next_state
    
    print("\n对话结束！")

# 运行演示（注释掉plot_markov_chain()可跳过图片生成）
if __name__ == "__main__":
    # plot_markov_chain()  # 生成状态转移图（需安装graphviz软件）
    run_dialogue_demo()

import numpy as np
from hmmlearn import hmm
import matplotlib.pyplot as plt

plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

# ---------------------- 第一步：构建HMM模型（语音断点识别） ----------------------
# 语音断点识别的核心：区分"有效语音段"（隐状态0）和"静音段"（隐状态1）
# 观测值：语音声学特征的能量值（连续值）

# 1. 定义HMM参数
n_hidden_states = 2  # 隐状态：0=有效语音，1=静音

# 2. 初始化HMM模型（使用GaussianHMM处理连续能量值）
model = hmm.GaussianHMM(n_components=n_hidden_states, n_iter=100, covariance_type="diag")

# 3. 设置初始参数（基于语音处理经验）
model.startprob_ = np.array([0.5, 0.5])  # 初始状态：50%概率是语音，50%是静音
model.transmat_ = np.array([
    [0.8, 0.2],  # 有效语音→继续语音(0.8)、静音(0.2)
    [0.1, 0.9]   # 静音→语音(0.1)、继续静音(0.9)
])
# 均值和协方差：语音段能量高，静音段能量低
model.means_ = np.array([[3.0], [0.5]])  # 语音平均能量3.0，静音0.5
model.covars_ = np.array([[0.5], [0.2]])  # 对应的方差

# ---------------------- 第二步：生成模拟语音观测序列 ----------------------
def generate_voice_observations(n_samples=100):
    """生成模拟的语音能量观测序列（模拟语音+静音交替）"""
    observations = []
    current_state = np.random.choice(n_hidden_states, p=model.startprob_)
    for _ in range(n_samples):
        # 根据当前状态生成观测值（从高斯分布采样）
        mean = model.means_[current_state, 0]
        std = np.sqrt(0.5 if current_state == 0 else 0.2)  # 直接使用设定的方差值
        obs = np.random.normal(mean, std)
        # 限制在合理范围内[0, 4]
        obs = np.clip(obs, 0, 4)
        observations.append(obs)
        # 转移到下一个状态
        current_state = np.random.choice(n_hidden_states, p=model.transmat_[current_state])
    return np.array(observations).reshape(-1, 1)

# 生成模拟数据
obs_sequence = generate_voice_observations(100)

# ---------------------- 第三步：HMM解码（识别语音断点） ----------------------
# Viterbi算法：找最可能的隐状态序列（语音/静音）
hidden_states = model.predict(obs_sequence)

# ---------------------- 第四步：可视化结果 ----------------------
def plot_voice_breakpoint():
    """可视化语音断点识别结果"""
    plt.figure(figsize=(12, 6))
    # 绘制观测值（能量）
    plt.subplot(2, 1, 1)
    plt.plot(obs_sequence, label='观测值（语音能量）', color='blue')
    plt.xlabel('时间步')
    plt.ylabel('能量值')
    plt.title('语音能量观测序列')
    plt.legend()
    
    # 绘制隐状态（语音/静音）
    plt.subplot(2, 1, 2)
    plt.step(range(len(hidden_states)), hidden_states, where='post', label='隐状态（0=语音，1=静音）', color='red')
    plt.xlabel('时间步')
    plt.ylabel('隐状态')
    plt.title('HMM识别的语音/静音断点')
    plt.legend()
    
    plt.tight_layout()
    plt.savefig('hmm_voice_breakpoint.png')
    print("语音断点识别结果已保存为 hmm_voice_breakpoint.png")

plot_voice_breakpoint()

# ---------------------- 第五步：大模型语义验证 ----------------------
def llm_voice_verification(audio_text, hidden_states):
    """
    大模型验证语音断点的语义合理性
    参数：
        audio_text - 语音转文字结果
        hidden_states - HMM识别的隐状态序列
    返回：验证结果
    """
    # 统计语音段和静音段的时长
    voice_duration = np.sum(hidden_states == 0)
    silence_duration = np.sum(hidden_states == 1)
    
    prompt = f"""
    你是语音断点验证助手，需要分析以下信息：
    1. 语音转文字结果：{audio_text}
    2. HMM识别的语音段时长：{voice_duration}时间步
    3. HMM识别的静音段时长：{silence_duration}时间步
    请验证：
    - 语音段的文字是否有完整语义；
    - 静音段的位置是否符合自然对话的停顿（比如句子之间）；
    - 给出断点识别的合理性评分（0-10分）和改进建议。
    """
    # 调用大模型（使用混元大模型）
    from openai import OpenAI
    import json

    client = OpenAI(
        api_key="sk-bW**********************************5NP8Ze",
        base_url="https://api.hunyuan.cloud.tencent.com/v1"
    )
    try:
        response = client.chat.completions.create(
            model="hunyuan-lite",
            messages=[
                {"role": "system", "content": "你是语音断点验证助手，需要分析语音识别结果和断点的语义合理性。"},
                {"role": "user", "content": prompt}
            ]
        )
        return response.choices[0].message.content
    except Exception as e:
        return f"验证出错：{e}"

# 模拟语音转文字结果，验证断点合理性
audio_text = "你好，我想咨询一下XX手机的价格，请问现在有优惠吗？"
verification_result = llm_voice_verification(audio_text, hidden_states)
print("\n大模型语义验证结果：")
print(verification_result)