Python 是你所需要的一切吗?介绍 Dria-Agent-α

社区文章 发布于 2025 年 1 月 10 日

引言

目前,大型语言模型(LLM)与**工具**交互的主要方式(有多种叫法,如工具使用、函数调用等)是向 LLM 提供其可使用的工具(包括工具的参数)的规范,然后 LLM 输出一个 JSON schema,其中包含要使用的工具及其参数。虽然这种方法直接且可靠,但它限制了 LLM 的表达能力,LLM 能够通过 Python 等编程语言引出更复杂的推理和解决方案。为此,我们定义了一个 LLM 通过 Python 使用工具的框架,称为**Python 函数调用**,该框架提示 LLM 输出 Python 代码中的动作。

使用 Python 与工具交互的动机如下:

  1. LLM 中的推理主要由预训练中的程序知识驱动,特别是代码文档。[1]
  2. 与使用基于 JSON 的函数调用的 LLM 相比,具备使用 Python 能力的 LLM 在代理场景中表现更好。[2]
  3. Python 是一种非常流行的编程语言 [3],在许多 LLM 的预训练数据中可能非常丰富,并且其伪代码般的语法使其非常接近人类自然语言。

示例

我们从一个简单的例子开始。用户查询:“您能帮我检查明天 10:00-12:00 是否有空,如果有空,就帮我预约与论文导师的会议吗?如果您预约成功,请将其添加到我的提醒事项中。”以及可用的函数:

def check_availability(day: str, start_time: str, end_time: str) -> bool:
    """ 
    Check if a time slot is available on a given day.

    Args:
    - day: The day to check in YYYY-MM-DD format
    - start_time: Start time in HH:MM format
    - end_time: End time in HH:MM format

    Returns:
    - True if slot is available, False otherwise
    """
    pass

def make_appointment(day: str, start_time: str, end_time: str) -> dict:
    """ 
    Make an appointment for a given time slot.

    Args:
    - day: The day to make appointment in YYYY-MM-DD format
    - start_time: Start time in HH:MM format
    - end_time: End time in HH:MM format
    - title: The title of the appointment

    Returns:
    - A dictionary with the appointment details and if it's made or not.
        dict keys:
            - day (str): The day the appointment is on, in YYYY-MM-DD format
            - start_time (str): Start time in HH:MM format
            - end_time (str): End time in HH:MM format
            - appointment_made (bool): Whether the appointment is successfully made or not. 
    """
    pass

def add_to_reminders(reminder_text: str) -> bool:
    """ 
    Add a text to reminders.

    Args: 
    - reminder_text: The text to add to reminders

    Returns:
    - Whether the reminder was successfully created or not.
    """
    pass

使用基于 JSON 的函数调用,这将需要多次聊天回合才能处理,因为它涉及两个条件(检查用户在给定时间段内是否可用,以及检查用户是否已成功预约与论文导师的会议),这将要求 LLM 接收函数调用结果并根据结果继续下一步。在 Python 函数调用中,这成为 LLM 可以在单个聊天回合中执行的简单任务,如下所示:

# Get today's date and calculate tomorrow's date
from datetime import datetime, timedelta
today = datetime.now()
tomorrow = (today + timedelta(days=1)).strftime("%Y-%m-%d")

# Define the time slots
start_time = "10:00"
end_time = "12:00"

# Check availability first
is_available = check_availability(tomorrow, start_time, end_time)

# Only proceed with making the appointment if it's available
appointment_result = (
    make_appointment(
        day=tomorrow,
        start_time=start_time,
        end_time=end_time,
        title="Meeting with Thesis Supervisor"
    )
    if is_available
    else {"appointment_made": False}
)

# Add to reminders only if the appointment was made
if appointment_result["appointment_made"]:
    add_to_reminders("Meeting with Thesis Supervisor scheduled for 10:00 AM tomorrow")

代码执行

由于模型生成用于函数调用的 Python 代码,我们利用 exec-python 来解析模型输出并执行代码以及函数。这使得 Dria-Agent-α 的 Python 函数调用方法能够直接集成,以安全且受控的方式处理生成代码的执行。

执行环境提供结构化输出,跟踪函数调用、变量状态以及执行过程中发生的任何错误。例如:

x = [1, 2]
y = [2, 3]
z = pair_sum(x, y)
k = pair_sum(z, z)

生成:

{
  "function_results": {
    "pair_sum": ["z","k"]
  },
  "variables": {
    "x": [1,2],
    "y": [2,3],
    "z": [3,5],
    "k": [6,10]
  },
  "errors": []
}

这种结构化输出对于多轮代理对话特别有价值,因为它允许模型保持对先前计算及其结果的感知,从而在后续交互中实现更复杂的推理链和依赖于状态的决策。

方法论

Dria-Agent-α 是利用通过 Dria 生成的合成数据开发的。Dria 是一个在分布式系统中运行的 LLM 网络,为跨不同模型的数据生成提供高吞吐量和强大的管道工具。

我们设计了一个框架,该框架创建需要复杂问题解决技能的真实场景,挑战模型将问题分解为可管理步骤并有效地利用所提供的函数,模仿实际的工具使用案例。我们计划在代码清理后于 2025 年 2 月之前发布合成数据生成管道,该管道包括以下步骤:

  1. 手动定义代表不同工具使用领域的类别和子类别(如存储库结构中的域/子域对所示)
  2. 使用多阶段管道生成合成场景
  3. 模拟函数生成
  4. 用户查询生成
  5. 模拟函数验证
  6. 通过代码执行验证
  7. 最终数据集编译

数据解剖

我们的训练数据由几个组件组成:

  • 用户查询
  • 带有文档字符串和参数的 Python 函数(以及它们的 JSON 等价物)
  • 模拟函数实现,为正确参数产生预期输出,为不正确参数产生不同输出
  • 验证每个查询所需函数调用及其预期输出的清单

示例条目

{
    "difficulty": "hard",
    "function_schema_python": "def check_user_permissions(username: str, folder_path: str) -> dict:\n    \"\"\"Checks the permissions of a specific user for a given network folder.\n\n    :param username: The username to check permissions for.\n    :param folder_path: The network folder path to check.\n    :return: Dictionary containing permission details:\n        - read (bool): Whether user has read permissions\n        - write (bool): Whether user has write permissions\n        - execute (bool): Whether user has execute permissions\n        - owner (str): Owner of the folder\n    :raises ValueError: If username or folder path is invalid.\"\"\"\n    pass\ndef modify_folder_permissions(username: str, folder_path: str, permissions: dict) -> bool:\n    \"\"\"Modifies the permissions for a specific user on a network folder.\n\n    :param username: The username to modify permissions for.\n    :param folder_path: The network folder path to modify.\n    :param permissions: Dictionary containing permission settings:\n        - read (bool): Whether to grant read permissions\n        - write (bool): Whether to grant write permissions\n        - execute (bool): Whether to grant execute permissions\n    :return: True if permissions were successfully modified, False otherwise.\n    :raises ValueError: If invalid parameters are provided.\"\"\"\n    pass\ndef verify_folder_access(username: str, folder_path: str) -> bool:\n    \"\"\"Verifies if a user can actually access a specific folder after permission changes.\n\n    :param username: The username to verify access for.\n    :param folder_path: The network folder path to verify.\n    :return: True if user can access the folder, False otherwise.\n    :raises ValueError: If username or folder path is invalid.\"\"\"\n    pass\n",
    "function_schema_json": [
        {
            "name": "check_user_permissions",
            "description": "Checks the permissions of a specific user for a given network folder.",
            "parameters": {
                "type": "object",
                "properties": {
                    "username": {
                        "type": "string",
                        "description": "The username to check permissions for."
                    },
                    "folder_path": {
                        "type": "string",
                        "description": "The network folder path to check."
                    }
                },
                "required": [
                    "username",
                    "folder_path"
                ],
                "additionalProperties": false
            }
        },
        {
            "name": "modify_folder_permissions",
            "description": "Modifies the permissions for a specific user on a network folder.",
            "parameters": {
                "type": "object",
                "properties": {
                    "username": {
                        "type": "string",
                        "description": "The username to modify permissions for."
                    },
                    "folder_path": {
                        "type": "string",
                        "description": "The network folder path to modify."
                    },
                    "permissions": {
                        "type": "object",
                        "description": "Dictionary containing permission settings:"
                    }
                },
                "required": [
                    "username",
                    "folder_path",
                    "permissions"
                ],
                "additionalProperties": false
            }
        },
        {
            "name": "verify_folder_access",
            "description": "Verifies if a user can actually access a specific folder after permission changes.",
            "parameters": {
                "type": "object",
                "properties": {
                    "username": {
                        "type": "string",
                        "description": "The username to verify access for."
                    },
                    "folder_path": {
                        "type": "string",
                        "description": "The network folder path to verify."
                    }
                },
                "required": [
                    "username",
                    "folder_path"
                ],
                "additionalProperties": false
            }
        }
    ],
    "mock_functions": "def check_user_permissions(username: str, folder_path: str) -> dict:\n    \"\"\"\n    Checks the permissions of a specific user for a given network folder.\n    \n    :param username: The username to check permissions for.\n    :param folder_path: The network folder path to check.\n    :return: Dictionary containing permission details:\n        - read (bool): Whether user has read permissions\n        - write (bool): Whether user has write permissions\n        - execute (bool): Whether user has execute permissions\n        - owner (str): Owner of the folder\n    :raises ValueError: If username or folder path is invalid.\n    \"\"\"\n    if not username or not folder_path:\n        raise ValueError(\"Username and folder path must be provided\")\n    \n    if username.lower() == \"alex\" and folder_path == \"\\\\\\\\server\\\\shared\\\\documents\":\n        return {\n            \"read\": False,\n            \"write\": False,\n            \"execute\": False,\n            \"owner\": \"Administrator\"\n        }\n    return {\n        \"read\": True,\n        \"write\": True,\n        \"execute\": True,\n        \"owner\": \"Administrator\"\n    }\ndef modify_folder_permissions(username: str, folder_path: str, permissions: dict) -> bool:\n    \"\"\"\n    Modifies the permissions for a specific user on a network folder.\n    \n    :param username: The username to modify permissions for.\n    :param folder_path: The network folder path to modify.\n    :param permissions: Dictionary containing permission settings:\n        - read (bool): Whether to grant read permissions\n        - write (bool): Whether to grant write permissions\n        - execute (bool): Whether to grant execute permissions\n    :return: True if permissions were successfully modified, False otherwise.\n    :raises ValueError: If invalid parameters are provided.\n    \"\"\"\n    if not username or not folder_path:\n        raise ValueError(\"Username and folder path must be provided\")\n    \n    required_keys = [\"read\", \"write\", \"execute\"]\n    if not all(key in permissions for key in required_keys):\n        raise ValueError(\"Permissions dictionary must contain read, write, and execute keys\")\n\n    if username.lower() == \"alex\" and folder_path == \"\\\\\\\\server\\\\shared\\\\documents\":\n        return True\n    return False\ndef verify_folder_access(username: str, folder_path: str) -> bool:\n    \"\"\"\n    Verifies if a user can actually access a specific folder after permission changes.\n    \n    :param username: The username to verify access for.\n    :param folder_path: The network folder path to verify.\n    :return: True if user can access the folder, False otherwise.\n    :raises ValueError: If username or folder path is invalid.\n    \"\"\"\n    if not username or not folder_path:\n        raise ValueError(\"Username and folder path must be provided\")\n    \n    if username.lower() == \"alex\" and folder_path == \"\\\\\\\\server\\\\shared\\\\documents\":\n        return True\n    return False",
    "user_query": "Hi, it's Linda. Could you modify the permissions for Alex on \\\\server\\shared\\documents to allow read and write access?",
    "checklist": {
        "functions": [
            "check_user_permissions",
            "modify_folder_permissions",
            "verify_folder_access"
        ],
        "values": [
            {
                "read": false,
                "write": false,
                "execute": false,
                "owner": "Administrator"
            },
            true,
            true
        ]
    }
}

我们以两个主要目标合成地生成了训练数据:

  • 对分布外 (OOD) 查询的稳健性能
  • 单次解决复杂多工具问题的能力

一个主要挑战是为实现更好的泛化能力创建全面的课程。我们专注于实际用例,特别强调以开发者为中心的场景,因为它们构成了代理请求的很大一部分。

传统方法使用课程元素作为生成用户查询的种子。然而,这种方法在我们的语境中导致了几个问题:

  • 不正确的函数实现
  • 不可行的模拟逻辑
  • 不准确的清单项
  • 用户查询缺乏足够的信息来正确选择函数参数

为了解决这些挑战,我们开发了一种以场景为先的方法。我们根据课程项目生成详细的场景,其中包含用户背景信息、特定于上下文的详细信息和相关的补充信息。这种全面的方法使我们能够创建具有足够上下文的模拟函数和用户查询,从而有效地避免模拟函数中不可行的逻辑和用户查询中的信息空白。

数据验证

合成数据固有的挑战需要验证机制。为了解决这个问题,我们实施了两种关键的验证方法:带 ICL 的验证器和受 RLEF [4] 启发的基于执行反馈的验证。

虽然 OpenAI o1 在验证模拟函数和场景可行性方面表现出色,但用它验证整个训练数据集在经济上是不可行的。相反,我们:

  1. 使用 OpenAI o1 生成了一个验证数据集
  2. 试验了不同的模型和验证方法
  3. 将它们的性能与 OpenAI o1 的验证结果进行比较

使用过程奖励模型的集束搜索

我们的第一种方法涉及使用较小的模型扩展 测试时间计算 (TTC)

  • Llama3.1 8B 与 O1 的验证集达到了 38% 的一致性
  • 使用 TTC,束大小为 16,并使用 Qwen2.5-Coder-32B-Instruct 作为过程奖励模型,性能提高到约 65%
  • 束大小高达 64 显示出改进,但伴随着显著的计算开销
  • 我们的实验表明,需要一个模型任务特定的 PRM 才能显著改进。

上下文学习

第二种方法是使用上下文学习来引导更具成本效益的模型的推理能力。我们使用 OpenAI o1 输出创建了一个 dSPY 优化的少样本提示,并建立了一个由 Qwen2.5-Coder-32B-Instruct 和 Claude Sonnet 组成的模型池。该系统生成了带有详细推理的验证输出,将较简单的示例路由到较小的模型,同时将复杂案例引导到较大的模型。这种分层方法在保持计算效率的同时,与 OpenAI o1 的验证结果达到了大约 80% 的一致性。

代码执行

最终的验证步骤涉及实现执行反馈循环。我们使用 Qwen2.5-Coder-32B-Instruct 执行生成的解决方案,收集堆栈跟踪和清单分数作为反馈。每个问题此过程迭代多达三次。我们只保留清单输出分数高于 0.75 的条目,以确保最终数据集中的高质量解决方案。这种基于执行的验证有助于消除语法正确但未能满足任务功能要求的解决方案。

模型

到目前为止,我们已经在 Qwen2.5-Coder-3B-Instruct 和 Qwen2.5-Coder-7B-Instruct 上训练了两个模型。我们的模型被称为 **Dria-Agent-α**,它们是 Dria 发布的第一代代理模型。这些模型,Dria-Agent-α-3BDria-Agent-α-7B,可在 Hugging Face 上获取。

未来工作

这是我们框架的首次迭代,我们正在开发下一次迭代,其中将涉及 RLEF[4] 和 rStar-Math[5] 的方法。这些模型的发布旨在展示 Python 函数调用的能力,并为未来的 Dria-Agent 模型铺平道路。

参考文献

社区

注册登录 发表评论