Qwen学习笔记1：Qwen模型使用ReAct原理实现function calling，增强模型能力_qwen1.5 function call

前言

在前面在本地化部署千问模型后，了解到千问模型可以自动调用函数工具去查找外部知识来增强大模型回复能力，这对于优秀的开源模型能力来说，是必不可少的，因此进一步学习。

Qwen模型，采用基于ReAct的思想实现了外部工具的API调用。ReAct的原理可以参看其官网如下：

简单来说ReAct原理就是通过让大模型一步一步的进行思考，并且行动，拿到相应的结果后再进行下一步的推理过程。

本文其实就是参考了哔哩哔哩教程：

【太牛了】Qwen结合ReAct，几分钟就能构建一个AI Agent，保姆级实操讲解，理论与实践相结合，讲述ReAct是如何作用于Qwen模型的_哔哩哔哩_bilibili

大家可以到原视频网址进行学习，以下就是自己手敲了全部的代码，给大家借鉴参考。

主要代码

导入相关包

from transformers import AutoModelForCausalLM, AutoTokenizer
from transformers.generation import GenerationConfig

启动模型

# model_path = './model/qwen/Qwen-1_8B-Chat'
model_path = './model/qwen/Qwen-7B-Chat'
tokenizer=AutoTokenizer.from_pretrained(model_path,trust_remote_code=True)
model=AutoModelForCausalLM.from_pretrained(model_path,device_map="auto",trust_remote_code=True)

先测试一下，模型无法直接回答的问题：

query = '请帮我查一下：我们的大模型技术实战课程目前一共上线了多少节？'
response, history = model.chat(tokenizer, query, history=None,)
print(response)

先构建一个提示词模板，让模型进入思考模式：

PROMPT_REACT = """
Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Begin!
Question:{query}"""

给它格式化的加入Query

import json
 
reason_prompt = PROMPT_REACT.format(query=query)
 
print(reason_prompt)

结果如下：

带入模型中，查看模型是否进入思考模式

# 带入模型中，查看模型是否进入思考模式
response, history = model.chat(tokenizer, reason_prompt, history=None,)
print(response)

大模型回答：

Thought: 任务是查询大模型技术实战课程的节数。我可以直接搜索相关的信息，或者询问有权限的人。我会选择前者。
Answer: 目前，我们大模型技术实战课程已经上线了20节课。

用JSON格式数据模拟数据库

class CourseDatabase:
    def __init__(self):
        self.database = {
            "大模型技术实战":{
                "课时": 200,
                "每周更新次数": 3,
                "每次更新小时": 2
            },
             "机器学习实战":{
                "课时": 230,
                "每周更新次数": 2,
                "每次更新小时": 1.5
            },
            "深度学习实战":{
                "课时": 150,
                "每周更新次数": 1,
                "每次更新小时": 3
            },
            "AI数据分析":{
                "课时": 10,
                "每周更新次数": 1,
                "每次更新小时": 1
            },
        }
    def course_query(self, course_name):
        return self.database.get(course_name, "目前没有该课程信息")

# 创建课程数据库实例
course_db = CourseDatabase()

# 查询已有课程的详细信息
course_name = "大模型技术实战"
print(course_db.course_query(course_name))

{'课时': 200, '每周更新次数': 3, '每次更新小时': 2}

# 查询不存在课程的详细信息
course_name = "人工智能"
print(course_db.course_query(course_name))

目前没有该课程信息

定义一个工具库，等会用于大模型进行调用：

TOOLS = [
    {
        'name_for_human': '课程信息数据库',
        'name_for_model': 'CourseDatabase',
        'description_for_model': '课程信息数据库存储有各课程的详细信息,包括目前的上线课时，每周更新次数以及每次更新的小时数。通过输入课程名称，可以返回该课程的详细信息。',
        'parameters': [{
            'name': 'course_query',
            'description': '课程名称,所需查询信息的课程名称',
            'required': True,
            'schema': {
                'type': 'string'
            },
        }],
    },
    # 其他工具的定义可以在这里继续添加
] 

再将该工具的信息通过格式代码生成“工具调用版提示词模板”

# 将一个插件的关键信息拼接成一段文本的模板
TOOL_DESC = """{name_for_model}: Call this tool to interact with the {name_for_human} API. What is the {name_for_human} API useful for? {description_for_model} Parameters:{parameters}
"""
 
PROMPT_REACT = """Answer the following questions as best you con. You have access to the following
{tool_descs}
Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can be repeated zero or more times)
Begin!
Question: {query}"""

定义函数进行处理：

import json
def generate_action_prompt(query):
    """
    根据用户查询生成最终的动作提示字符串。
    函数内部直接引用全局变量 TOOLS, TOOL_DESC, 和 PROMPT_REACT.
    参数：
    - query: 用户的查询字符串。
    返回：
    - action_prompt: 格式化后的动作提示字符串。
    """
 
    tool_descs = []
    tool_names = []
 
    for info in TOOLS:
        tool_descs.append(
            TOOL_DESC.format(
                name_for_model = info['name_for_model'],
                name_for_human = info['name_for_human'],
                description_for_model = info['description_for_model'],
                parameters = json.dumps(info['parameters'], ensure_ascii=False),
            )
        )
        tool_names.append(info['name_for_model'])
 
    tool_descs_str = '\n\n'.join(tool_descs)
    tool_names_str = ','.join(tool_names)
 
    action_prompt = PROMPT_REACT.format(tool_descs=tool_descs_str, tool_names=tool_names_str, query=query)
    return action_prompt

查看一下效果

action_prompt = generate_action_prompt(query)
print(action_prompt)

Answer the following questions as best you con. You have access to the following

CourseDatabase: Call this tool to interact with the 课程信息数据库 API. What is the 课程信息数据库 API useful for? 课程信息数据库存储有各课程的详细信息,包括目前的上线课时，每周更新次数以及每次更新的小时数。通过输入课程名称，可以返回该课程的详细信息。 Parameters:[{"name": "course_query", "description": "课程名称,所需查询信息的课程名称", "required": true, "schema": {"type": "string"}}]


Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [CourseDatabase]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can be repeated zero or more times)

Begin!

Question: 请帮我查一下：我们的大模型技术实战课程目前一共上线了多少节？

在这一步需要添加，中止符的处理流程，如果没有终止符，模型在后续可能会输出一些胡乱编辑的内容。

react_stop_words = [
    tokenizer.encode('Observation:'),
    tokenizer.encode('Observation:\n'),
]

# 使用action_prompt生成回复
response, history = model.chat(tokenizer, action_prompt, history=None, \
                              stop_words_ids=react_stop_words)
print(response)

Thought: 我需要使用课程信息数据库API来获取这个信息。
Action: CourseDatabase
Action Input: {"course_query": "大模型技术实战"}
Observation:

可以看到在Observation:这就停止输出了。

定义函数解析从模型拿到的结果：

def parse_plugin_action(text: str):
    """
    解析模型的ReAct输出文本提取名称及其参数。
    参数：
    - text： 模型ReAct提示的输出文本
    返回值：
    - action_name: 要调用的动作（方法）名称。
    - action_arguments: 动作（方法）的参数。
    """
    # 查找“Action:”和“Action Input：”的最后出现位置
    action_index = text.rfind('\nAction:')
    action_input_index = text.rfind('\nAction Input:')
    observation_index = text.rfind('\nObservation:')
 
    # 如果文本中有“Action:”和“Action Input：”
    if 0 <= action_index < action_input_index:
        if observation_index < action_input_index:
            text = text.rstrip() + '\nObservation:'
            observation_index = text.rfind('\nObservation:')
 
    # 确保文本中同时存在“Action:”和“Action Input：”
    if 0 <= action_index < action_input_index < observation_index:
        # 提取“Action:”和“Action Input：”之间的文本为动作名称
        action_name = text[action_index + len('\nAction:'):action_input_index].strip()
        # 提取“Action Input：”之后的文本为动作参数
        action_arguments = text[action_input_index + len('\nAction Input:'):observation_index].strip()
        return action_name, action_arguments
 
    # 如果没有找到符合条件的文本，返回空字符串
    return '', ''

再定义一个函数，通过解析的数据，进一步执行需要的操作，生成需要的请求信息：

import json
def execute_plugin_from_react_output(response):
    """
    根据模型的ReAct输出执行相应的插件调用，并返回调用结果。
    参数：
    - response: 模型的ReAct输出字符串。
    返回：
    - result_dict: 包括状态码和插件调用结果的字典。
    """
    # 从模型的ReAct输出中提取函数名称及函数入参
    plugin_configuration = parse_plugin_action(response)
    first_config_line = plugin_configuration[1:][0].split('\n')[0]
    config_parameters = json.loads(first_config_line)
    result_dict = {"status_code": 200}
 
    for k, v in config_parameters.items():
        if k in TOOLS[0]["parameters"][0]['name']:
            # 通过eval函数执行存储在字符串中的python表达式，并返回表达式计算结果。其执行过程实质上是实例化类
            tool_instance = eval(TOOLS[0]["name_for_model"])()
            # 然后通过getattr函数传递对象和字符串形式的属性或方法名来动态的访问该属性和方法h
            tool_func = getattr(tool_instance, k)
            # 这一步实际上执行的过程就是：course_db,course_query('大模型技术实战')
            tool_result = tool_func(v)
            result_dict["result"] = tool_result
            return result_dict
 
    result_dict["status_code"] = 404
    result_dict["result"] = "未找到匹配的插件配置"
    return result_dict
 

tool_result = execute_plugin_from_react_output(response)
print(tool_result)

{'status_code': 200, 'result': {'课时': 200, '每周更新次数': 3, '每次更新小时': 2}}

拼接成第二轮对话的输入：

response += " " + str(tool_result)
print(response)

Thought: 需要调用课程信息数据库API来获取课程上线节数。
Action: CourseDatabase
Action Input: {"course_query": "大模型技术实战"}
Observation: {'status_code': 200, 'result': {'课时': 200, '每周更新次数': 3, '每次更新小时': 2}}

输入第二次对话的请求：

response, history = model.chat(tokenizer, response, history=history, \
                              stop_words_ids=react_stop_words)
print(response)

Thought: 从返回结果中可以看出，我们的大模型技术实战课程目前已经上线了200节课，每周更新3次，每次更新2小时。
Answer: 目前，我们的大模型技术实战课程已经上线了200节课。

从返回结果上看，以上通过手写代码，实现了将千问模型通过调用外部函数，增强了它本身的功能。

补充：完整版的提示词模板代码：

# 将一个插件的关键信息拼接成一段文本的模板
TOOL_DESC = """{name_for_model}: Call this tool to interact with the {name_for_human} API. What is the {name_for_human} API useful for? {description_for_model} Parameters:{parameters}
"""
 
PROMPT_REACT = """Answer the following questions as best you con. You have access to the following
{tool_descs}
Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can be repeated zero or more times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Begin!
Question: {query}"""