微调碾压RAG？大模型意图识别工程化实践

介绍一下李白帮我播放两个男人在天台决斗的电影打开右下角那个电视剧我想听一个小朋友不好好吃饭就会肚子疼的故事深圳最近有哪些新闻？夏洛特烦恼这部电影中夏洛这个角色是不是沈腾演的？电影战狼3的上映时间是什么时候？

# 思考步骤为了完成上述目标，请按照下面的过程，一步一步思考后完成: 1.首先需要先理解候选的意图信息共有如下{}个类别，每个意图的含义或常见描述如下：{}

2. 然后，观察以下意图识别分类案例，学习意图识别分类任务的能力：{}

3. 意图识别前，请重点学习并记住以下识别准则：{}

（4）输出示例

4. 请以JSON格式进行输出，输出示例：{}

prompt = """# 角色你是一位意图样本生成专家，擅长根据给定的模板生成意图及对应的槽位信息。你能够准确地解析用户输入，并将其转化为结构化的意图和槽位数据。
## 技能### 技能1：解析用户指令- **任务**：根据用户提供的自然语言指令，识别出用户的意图。
### 技能2：生成结构化意图和槽位信息- 意图分类：video_search,music_search,information_search- 槽位分类：-- information_search: classification,video_category,video_name,video_season-- music_search: music_search,music_singer,music_tag,music_release_time-- video_search: video_actor,video_name,video_episode- **任务**：将解析出的用户意图转换为结构化的JSON格式。  - 确保每个意图都有相应的槽位信息，不要自行编造槽位。  - 槽位信息应包括所有必要的细节，如演员、剧名、集数、歌手、音乐标签、发布时间等。
### 技能3：在线搜索- 如果遇到关于电影情节的描述，可以调用搜索引擎获取到电影名、演员等信息称补充到actor,name等槽位中
### 输出示例  - 以JSON格式输出，例如：    -"这周杭州的天气如何明天北京有雨吗"：{'infor_search':{'extra_info':['这周杭州的天气如何明天北京有雨吗']}}-"我一直在追赵丽颖的楚乔传我看到第二十集了它已经更新了吗我可以看下一集吗"：{'video_search':{'video_actor':['赵丽颖'],'video_name':['楚乔传'],'video_episode':['第21集'],'extra_info':['我一直在追赵丽颖的楚乔传我看到第二十集了它已经更新了吗我可以看下一集吗']}}
## 限制- 只处理与意图生成相关的任务。- 生成的意图和槽位信息必须准确且完整。- 在解析过程中，确保理解用户的意图并正确映射到相应的服务类型- 如果遇到未知的服务类型或槽位信息，可以通过调用搜索工具进行补充和确认。- 直接输出Json，不要输出其他思考过程

[{"instruction":"播放一首七仔的歌曲我想听他的经典老歌最好是70年代的音乐风格","output":"{'music_search':{'singer':['张学友'],'music_tag':['经典老歌'],'release_time':['70年代']}}"}]

这里有非常多成熟的方案可以选择，可以考虑阿里云百炼平台的RAG方案，此处不再赘述。

{"data":{"instruction": "我想听音乐"}}{"data":{"instruction": "太吵了，把声音开小一点"}}{"data":{"instruction": "我不想听了，把歌关了吧"}}{"data":{"instruction": "我想去杭州玩，帮我查下天气预报"}}

标注完成后可以将标注结果导出至OSS，在本示例中，输出文件内容如下：

{"data":{"instruction":"我想听音乐","_itag_index":""},"label-1787402095227383808":{"results":[{"questionId":"2","data":"play_music()","markTitle":"output","type":"survey/value"}]},"abandonFlag":0,"abandonRemark":null}{"data":{"instruction":"太吵了，把声音开小一点","_itag_index":""},"label-1787402095227383808":{"results":[{"questionId":"2","data":"volume_down()","markTitle":"output","type":"survey/value"}]},"abandonFlag":0,"abandonRemark":null}{"data":{"instruction":"我不想听了，把歌关了吧","_itag_index":""},"label-1787402095227383808":{"results":[{"questionId":"2","data":"music_exit()","markTitle":"output","type":"survey/value"}]},"abandonFlag":0,"abandonRemark":null}{"data":{"instruction":"我想去杭州玩，帮我查下天气预报","_itag_index":""},"label-1787402095227383808":{"results":[{"questionId":"2","data":"weather_search(杭州)","markTitle":"output","type":"survey/value"}]},"abandonFlag":0,"abandonRemark":null}

# 输入文件路径和输出文件路径input_file_path = 'test_json.manifest'output_file_path = 'train.json'
converted_data = []with open(input_file_path, 'r', encoding='utf-8') as file:for line in file:data = json.loads(line)instruction = data['data']['instruction']for key in data.keys():if key.startswith('label-'):output = data[key]['results'][0]['data']converted_data.append({'instruction': instruction, 'output': output})break
with open(output_file_path, 'w', encoding='utf-8') as outfile:json.dump(converted_data, outfile, ensure_ascii=False, indent=4)

实际用于训练的数据格式如下：

<|im_start|>system\n你是一个意图识别专家，可以根据用户的问题识别出意图，并返回对应的意图和参数<|im_end|>\n<|im_start|>user\n我想听音乐<|im_end|>\n<|im_start|>assistant\nplay_music()<|im_end|>\n

#encoding=utf-8from transformers import AutoModelForCausalLM, AutoTokenizerimport jsonfrom tqdm import tqdm
device = "cuda" # the device to load the model onto
# 修改模型路径model_name = '/mnt/workspace/model/qwen14b-lora-3e4-256-train/'print(model_name)model = AutoModelForCausalLM.from_pretrained(    model_name,    torch_dtype="auto",    device_map="auto")tokenizer = AutoTokenizer.from_pretrained(model_name)
count = 0ecount = 0

# 修改训练数据路径test_data = json.load(open('/mnt/workspace/data/testdata.json'))system_prompt = '你是一个意图识别专家，可以根据用户的问题识别出意图，并返回对应的函数调用和参数。'
for i in tqdm(test_data[:]):    prompt = '<|im_start|>system\n' + system_prompt + '<|im_end|>\n<|im_start|>user\n' + i['instruction'] + '<|im_end|>\n<|im_start|>assistant\n'    gold = i['output']    gold = gold.split(';')[0] if ';' in gold else gold
    model_inputs = tokenizer([prompt], return_tensors="pt").to(device)    generated_ids = model.generate(        model_inputs.input_ids,        max_new_tokens=64,        pad_token_id=tokenizer.eos_token_id,        eos_token_id=tokenizer.eos_token_id,        do_sample=False    )    generated_ids = [        output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)    ]    pred = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]    if gold.split('(')[0] == pred.split('(')[0]:        count += 1        gold_list = set(gold.strip()[:-1].split('(')[1].split(','))        pred_list = set(pred.strip()[:-1].split('(')[1].split(','))        if gold_list == pred_list:            ecount += 1    else:        pass
print("意图识别准确率：", count/len(test_data))print("参数识别准确率：", ecount/len(test_data))

import argparseimport jsonfrom typing import Iterable, List
import requests
def post_http_request(prompt: str,                      system_prompt: str,                      history: list,                      host: str,                      authorization: str,                      max_new_tokens: int = 2048,                      temperature: float = 0.95,                      top_k: int = 1,                      top_p: float = 0.8,                      langchain: bool = False,                      use_stream_chat: bool = False) -> requests.Response:    headers = {        "User-Agent": "Test Client",        "Authorization": f"{authorization}"    }    pload = {        "prompt": prompt,        "system_prompt": system_prompt,        "top_k": top_k,        "top_p": top_p,        "temperature": temperature,        "max_new_tokens": max_new_tokens,        "use_stream_chat": use_stream_chat,        "history": history    }    response = requests.post(host, headers=headers,                             json=pload, stream=use_stream_chat)    return response
def get_response(response: requests.Response) -> List[str]:    data = json.loads(response.content)    output = data["response"]    history = data["history"]    return output, history
if __name__ == "__main__":    parser = argparse.ArgumentParser()    parser.add_argument("--top-k", type=int, default=4)    parser.add_argument("--top-p", type=float, default=0.8)    parser.add_argument("--max-new-tokens", type=int, default=2048)    parser.add_argument("--temperature", type=float, default=0.95)    parser.add_argument("--prompt", type=str, default="How can I get there?")    parser.add_argument("--langchain", action="store_true")
    args = parser.parse_args()
    prompt = args.prompt    top_k = args.top_k    top_p = args.top_p    use_stream_chat = False    temperature = args.temperature    langchain = args.langchain    max_new_tokens = args.max_new_tokens
    host = "EAS服务公网地址"    authorization = "EAS服务公网Token"
    print(f"Prompt: {prompt!r}\n", flush=True)    # 在客户端请求中可设置语言模型的system prompt。    system_prompt = "你是一个意图识别专家，可以根据用户的问题识别出意图，并返回对应的意图和参数"
    # 客户端请求中可设置对话的历史信息，客户端维护当前用户的对话记录，用于实现多轮对话。通常情况下可以使用上一轮对话返回的histroy信息，history格式为List[Tuple(str, str)]。    history = []    response = post_http_request(        prompt, system_prompt, history,        host, authorization,        max_new_tokens, temperature, top_k, top_p,        langchain=langchain, use_stream_chat=use_stream_chat)    output, history = get_response(response)    print(f" --- output: {output} \n --- history: {history}", flush=True)
# 服务端返回JSON格式的响应结果，包含推理结果与对话历史。def get_response(response: requests.Response) -> List[str]:    data = json.loads(response.content)    output = data["response"]    history = data["history"]    return output, history

{"role": "system", "content": "你是个意图识别专家，你要通过用户的输入，识别用户的意图以及槽位信息，注意：不要添加没有的意图和槽位。"}