用 DeepSeek 构建从自然语言到 PNR 机器指令的转换系统

# 术语表处理 - 处理机场信息，合并中英文名称
def process_airport_info(input_file, output_file):
    df = pd.read_excel(input_file, usecols=[0, 1, 2])
    df.iloc[:, 1].fillna(df.iloc[:, 2], inplace=True)
    df.drop(df.columns[2], axis=1, inplace=True)
    df.to_excel(output_file, index=False, header=True, engine='openpyxl')
    return df

# 术语表处理 - 合并机场和城市信息
def merge_airport_city(airport_file, city_file, output_file):
    df_airport = pd.read_excel(airport_file)
    df_city = pd.read_excel(city_file, header=None).iloc[:, :2]
    df_city.columns = df_airport.columns[:2]
    df_final = pd.concat([df_airport, df_city], ignore_index=True)
    df_final.to_excel(output_file, index=False, engine='openpyxl')
    return df_final

# 术语表处理 - 合并机场、城市和航司信息
def merge_all_info(info_file, airline_file, output_file):
    df_info = pd.read_excel(info_file)
    df_airline = pd.read_excel(airline_file, header=None).iloc[:, :2]
    df_airline.columns = df_info.columns[:2]
    df_final = pd.concat([df_info, df_airline], ignore_index=True)
    df_final.to_excel(output_file, index=False, engine='openpyxl')
    return df_final

# 数据格式转换 - Excel 转 txt
def excel_to_txt(excel_file, txt_file):
    df = pd.read_excel(excel_file, usecols=[0, 1])
    data_to_write = df.to_csv(sep='\t', na_rep='nan')
    with open(txt_file, 'w', encoding='utf-8') as f:
        f.write(data_to_write)

# 主程序，依次调用各处理函数
if __name__ == "__main__":
    # 处理机场信息
    airport_input = '*/*.xls'
    airport_output = '*/*.xls'
    df_airport = process_airport_info(airport_input, airport_output)

    # 合并机场和城市信息
    city_input = '*/*.xlsx'
    merge12_output = '*/*.xlsx'
    df_merge12 = merge_airport_city(airport_output, city_input, merge12_output)

    # 合并机场、城市和航司信息
    info_input = '*/*.xlsx'
    airline_input = '*/*.xlsx'
    merge_all_output = '*/*.xlsx'
    df_merge_all = merge_all_info(info_input, airline_input, merge_all_output)

    # Excel 转 txt
    excel_to_txt(merge_all_output, '*.txt')

system_prompt = f'''
你是一名指令识别助手，你需要理解用输入并完成以下工作：

1.航程类型识别：
（1）单程-显示承运方航班信息。如果有提到承运方航班，则航程类型识别结果输出 "xxx"。
（2）单程- 显示xx联盟的航空公司航班信息。如果有提到xx联盟的航空公司航班，则航程类型识别结果输出 "xxx"。
（3）双程-往返。如果在用户输入中提到了在两个城市间往返。
（4）多旅程。如果在用户输入中提到多段城市间的旅程，则航程类别结果输出 "xxx"。
2、出发/达到地点识别：理解用户输入，并在其中识别关键信息：出发地点和到达地点。
3、航司识别：理解用户输入，并在其中识别航空公司关键信息，"即可。
4、直飞识别：理解用户是否提到'直飞', 如果有提到'直飞'，在识别结果中,则输出"ture", 如果没有提到则输出"false"
5、无经停识别：理解用户是否提到'无经停', 如果有提到'无经停'，在识别结果中,则输出"ture", 如果没有提到则输出"false"
6、完整显示：如果航司类别结果输出为"xxx",则该项输出"ture", 如果不是"xxx"则输出"false"
7、按到达时间排列：理解用户是否提到'按到达时间排列', 如果有提到'按到达时间排列'，在识别结果中,则输出"ture", 如果没有提到则输出"false"
注意，今天的时间为：{today}时
'''

def stream_llm_response(client, messages, model="deepseek-ai/DeepSeek-V3", max_tokens=4096):
    """Stream response from LLM and return accumulated text"""
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        max_tokens=max_tokens,
        temperature=0,  # Lowest temperature for most deterministic output
        presence_penalty=2.0,  # Highest presence penalty to encourage diverse content
        stream=True
    )
    response_text = ""
    for chunk in response:
        if chunk.choices[0].delta.content is not None:
            chunk_text = chunk.choices[0].delta.content
            print(chunk_text, end='')  # Print each text chunk as it arrives
            response_text += chunk_text
    
    return response_text
    
def save_response(response_text, output_path):
    """Save response text to file"""
    with open(output_path, 'w', encoding='utf-8') as f:
        f.write(response_text)
    print(f"\nSaved to: {output_path}")  

def get_whole_word_pattern(search_term):
    pattern = r'\b' + re.escape(search_term) + r'\b'
    print(f"构造的正则表达式模式: {pattern}")
    return pattern
    
def find_code(df, col_index, search_term, faiss_index):
    print(f"\n搜索开始: {search_term}")
    # 检查列号是否在有效范围内
    if col_index < 0 or col_index >= len(df.columns):
        print(f"错误：列号 '{col_index}' 超出了DataFrame的列范围。")
        return None
    
    if not search_term:
        print("搜索词为空。")
        return None
    pattern = get_whole_word_pattern(search_term)
    # 使用列号代替列名来访问列
    mask = df.iloc[:, col_index].str.contains(pattern, na=False, regex=True, case=False)
    print(f"DataFrame匹配掩码: {mask}")
    
    if df.loc[mask].shape[0] > 0:
        print(f"在DataFrame中找到匹配项: {df.loc[mask].iloc[:, col_index].values}")
        # 直接使用列号获取'机场码'列的值
        return df.loc[mask].iloc[0, 0]
    else:
        print("在DataFrame中未找到匹配项，执行FAISS搜索...")

model_name = "BAAI/bge-m3"
model_kwargs = {'device': 'cuda'}  
encode_kwargs = {'normalize_embeddings': True}  # 归一化嵌入向量，提高相似度计算的准确性
embeddings = HuggingFaceBgeEmbeddings(
    model_name=model_name, 
    model_kwargs=model_kwargs, 
    encode_kwargs=encode_kwargs
)

from langchain.document_loaders.telegram import text_to_docs
from langchain.vectorstores import FAISS
from langchain.embeddings.huggingface import HuggingFaceBgeEmbeddings

# 读取术语表文件
with open(file_path, 'r', encoding='utf-8') as file:
    sampled_texts = file.readlines()

# 数据预处理：移除每行末尾的换行符
sampled_texts = [text.strip() for text in sampled_texts]

# 将文本转换为文档格式，便于后续处理
docs = text_to_docs(sampled_texts)

# 设置嵌入模型
model_name = "BAAI/bge-m3"
model_kwargs = {'device': 'cuda'}
encode_kwargs = {'normalize_embeddings': True}
embeddings = HuggingFaceBgeEmbeddings(
    model_name=model_name, 
    model_kwargs=model_kwargs, 
    encode_kwargs=encode_kwargs
)

# 构建FAISS向量索引
faiss_index = FAISS.from_documents(docs, embeddings)

# 打印模型参数数量，了解模型规模
total_params = sum(p.numel() for p in embeddings.client.parameters())
print(f"Total number of parameters: {total_params}")

def find_code(df, col_index, search_term, faiss_index):
    """
    查找实体对应的代码，使用混合检索策略
    
    参数:
    df - 包含实体和代码的DataFrame
    col_index - 实体名称所在的列索引
    search_term - 要查找的实体名称
    faiss_index - FAISS向量索引
    
    返回:
    找到的代码，如果未找到则返回None
    """
    # 处理空值
    if not search_term or search_term == "false":
        return None
    
    print(f"\n搜索开始: {search_term}")
    
    # 步骤1: 精确匹配 - 使用正则表达式
    pattern = r'\b' + re.escape(search_term) + r'\b'
    mask = df.iloc[:, col_index].str.contains(pattern, na=False, regex=True, case=False)
    
    if df.loc[mask].shape[0] > 0:
        print(f"在DataFrame中找到精确匹配项: {df.loc[mask].iloc[:, col_index].values}")
        return df.loc[mask].iloc[0, 0]  # 返回找到的代码
    
    # 步骤2: 向量相似度检索
    print("在DataFrame中未找到精确匹配项，执行FAISS向量检索...")
    results = faiss_index.similarity_search(search_term, k=1)  # 找到最相似的1个结果
    
    if results and len(results) > 0:
        doc = results[0]  # 获取最相似的Document对象
        page_content = doc.page_content  # 获取page_content
        print(f"FAISS返回的页面内容: {page_content}")
        
        # 解析页面内容，提取代码
        parts = page_content.split('\t')
        if len(parts) > 1:
            code = parts[0]  # 代码在第一列
            print(f"提取的代码: {code}")
            return code
    
    print(f"未找到 '{search_term}' 的匹配项。")
    return None

def format_time(time_str):
    """
    将自然语言时间表达转换为PNR时间格式（HHMM）
    参数:
    time_str - 自然语言时间表达，如"上午"、"下午"、"13点45分"
    返回:
    PNR时间格式，如"0600"、"1200"、"1345"
    """
    # 定义时间映射
    time_map = {
        '上午': '0600',  # 上午默认为6点
        '下午': '1200',  # 下午默认为12点
        '晚上': '1800',  # 晚上默认为18点
        '凌晨': '0000'   # 凌晨默认为0点
    }
    # 处理空值或布尔值
    if not time_str or time_str == "false":
        return None
    
    # 检查是否为映射中的时间段
    if time_str in time_map:
        return time_map[time_str]
    
    # 处理具体时间，如"13点45分"
    parts = time_str.split('点')
    if len(parts) == 2:
        hour_str, minute_str_with_suffix = parts
        minute_str = minute_str_with_suffix.rstrip('分')
        
        try:
            hour = int(hour_str)
            # 处理12小时制到24小时制的转换
            if '下午' in time_str and hour < 12:
                hour += 12
            
            minute = int(minute_str) if minute_str else 0
            if 0 <= hour < 24 and 0 <= minute < 60:
                return f"{hour:02d}{minute:02d}"
        except ValueError:
            pass
    return None

def format_date(date_str):
    """
    将自然语言日期表达转换为PNR日期格式（DDMMM[YY]）
    参数:
    date_str - 自然语言日期表达，如"2025年07月31日"
    返回:
    PNR日期格式，如"31jul25"
    """
    # 月份映射
    month_map = {
        '01': 'jan', '02': 'feb', '03': 'mar', '04': 'apr', '05': 'may', '06': 'jun',
        '07': 'jul', '08': 'aug', '09': 'sep', '10': 'oct', '11': 'nov', '12': 'dec'
    }
    # 处理空值或布尔值
    if not date_str or date_str == "false":
        return None
    try:
        # 处理2025年格式
        if "2025" in date_str:
            parts = date_str.split('年')[1].split('月')
            day = parts[1].split('日')[0].zfill(2)  # 确保日期是两位数
            month_key = parts[0].zfill(2)  # 确保月份是两位数
            return f"{day}{month_map[month_key]}25"  # 添加年份后缀
        # 处理2024年格式
        else:
            parts = date_str.split('年')[1].split('月')
            day = parts[1].split('日')[0].zfill(2)
            month_key = parts[0].zfill(2)
            return f"{day}{month_map[month_key]}"  # 不添加年份后缀，默认为当年
    except (IndexError, KeyError):
        print(f"日期格式错误: {date_str}")
        return None

def build_command(trip_type, departure_code1, arrival_code1, airline_code1, departure_code2, arrival_code2, direct, no_stopovers, date_str=None, time_str=None, yizhongzhuan_code1=None, yizhongzhuan_code2=None, yizhongzhuan_code3=None, date_str2=None, time_str2=None, airline_code2=None, erzhongzhuan_code1=None, erzhongzhuan_code2=None, erzhongzhuan_code3=None):
    """
    根据航程类型和实体信息构建PNR指令
    参数:
    trip_type - 航程类型，如"001"、"010"等
    departure_code1 - 第一段行程的出发地代码
    arrival_code1 - 第一段行程的目的地代码
    airline_code1 - 第一段行程的航空公司代码
    ...
    返回:
    构建的PNR指令
    """
    # 格式化日期和时间
    date_code1 = format_date(date_str)
    time_code1 = format_time(time_str)
    date_code2 = format_date(date_str2)
    time_code2 = format_time(time_str2)

    # 初始化指令
    command = ""
    # 根据航程类型构建指令
    if trip_type == '001':  # 单程-完整显示
        command = f"av:h/{departure_code1}{arrival_code1}/{date_code1}"
        if time_code1:
            command += f"/{time_code1}"
        if airline_code1:
            command += f"/{airline_code1}"
        if direct:
            command += "/d" 
        if no_stopovers:
            command += "/n" 
    
    elif trip_type == '002':  # 单程-按到达时间排列
        command = f"av:a/{departure_code1}{arrival_code1}/{date_code1}"
        # ... 类似的处理 ...
    
    elif trip_type == '003':  # 单程-显示承运方航班信息
        command = f"av:o/{departure_code1}{arrival_code1}/{date_code1}"
        # ... 类似的处理 ...
  
    elif trip_type == '010':  # 双程-往返
        # 构建往程部分
        command = f"av:{departure_code1}{arrival_code1}/{date_code1}"
        
        if time_code1:
            command += f"/{time_code1}"
        if airline_code1:
            command += f"/{airline_code1}"
        if direct:
            command += "/d" 
        if no_stopovers:
            command += "/n" 
        for zhongzhuan_code in [yizhongzhuan_code1, yizhongzhuan_code2, yizhongzhuan_code3]:
            if zhongzhuan_code:
                command += f"/{zhongzhuan_code}"
        # 添加返程部分
        if date_code2:
            command += f"&{date_code2}"
            if time_code2:
                command += f"/{time_code2}"
            if airline_code2:
                command += f"/{airline_code2}"
            if direct:
                command += "/d" 
            if no_stopovers:
                command += "/n" 
            for zhongzhuan_code in [erzhongzhuan_code1, erzhongzhuan_code2, erzhongzhuan_code3]:
                if zhongzhuan_code:
                    command += f"/{zhongzhuan_code}"
    elif trip_type == '011':  # 多旅程
        # 构建第一段行程
        command = f"av:{departure_code1}{arrival_code1}/{date_code1}"
        
        if time_code1:
            command += f"/{time_code1}"
        if airline_code1:
            command += f"/{airline_code1}"
        if direct:
            command += "/d" 
        if no_stopovers:
            command += "/n" 
        for zhongzhuan_code in [yizhongzhuan_code1, yizhongzhuan_code2, yizhongzhuan_code3]:
            if zhongzhuan_code:
                command += f"/{zhongzhuan_code}"
        
        # 添加第二段行程
        if departure_code2 and arrival_code2:
            command += f"&{departure_code2}{arrival_code2}"
            if date_code2:
                command += f"/{date_code2}"
            if time_code2:
                command += f"/{time_code2}"
            if airline_code2:
                command += f"/{airline_code2}"
            if direct:
                command += "/d" 
            if no_stopovers:
                command += "/n" 
            for zhongzhuan_code in [erzhongzhuan_code1, erzhongzhuan_code2, erzhongzhuan_code3]:
                if zhongzhuan_code:
                    command += f"/{zhongzhuan_code}"

    # 清理指令，移除无效部分
    command = command.replace(" ", "")  # 移除空格
    command = command.replace("/None", "")  # 移除无效参数
    command = command.replace("None", "")  # 移除无效值
    command = command.replace("&None", "")  # 移除无效连接符
    
    return command

def validate_command(command):
    """
    验证生成的PNR指令是否符合规范
    参数:
    command - 生成的PNR指令
    返回:
    验证结果，如果符合规范则返回True，否则返回错误信息
    """
    # 基本格式验证
    if not command.startswith("av:"):
        return False, "指令必须以'av:'开头"
    
    # 分割指令
    parts = command.split("/")
    if len(parts) < 2:
        return False, "指令格式不正确，缺少必要的分隔符"
    
    # 验证航程类型
    av_part = parts[0]
    if ":" in av_part:
        av_type = av_part.split(":")[1]
        if av_type and av_type not in "haeosmtw":
            return False, f"不支持的航程类型: {av_type}"
    
    # 验证城市代码
    if len(parts) > 1:
        city_part = parts[1]
        if len(city_part) < 6:  # 至少需要出发地和目的地各3个字符
            return False, "城市代码格式不正确"
    
    # 验证日期
    if len(parts) > 2:
        date_part = parts[2]
        if not re.match(r'^\d{2}[a-z]{3}(\d{2})?$', date_part):
            return False, f"日期格式不正确: {date_part}"
    
    # 验证时间（如果有）
    if len(parts) > 3:
        time_part = parts[3]
        if time_part and not re.match(r'^\d{4}$', time_part) and not time_part.upper() in ["CA", "MU", "CZ", "3U"]:
            return False, f"时间格式不正确: {time_part}"
    
    return True, "指令格式正确"

# 创建Gradio界面
iface = gr.Interface(
    fn=main,
    inputs=gr.Textbox(label="请输入您的飞行计划"),
    outputs=gr.Textbox(label="生成的指令"),
    title="飞行指令转换器",
    description="输入您的飞行计划，系统将生成相应的指令。"
)

# 启动Gradio应用
iface.launch(share=True)