+
69
-

回答

要实现“自动创建技能并使用技能”的核心思想是:

元编程 (Meta-programming):让 LLM 编写 Python 代码(函数)。

动态执行 (Dynamic Execution):在 Python 运行时中执行这段代码,将其注册为可用函数。

工具调用 (Tool Calling):利用 OpenAI 兼容 API 的 Function Calling 能力,既用来调用“创建技能”的工具,也用来调用“新创建好”的技能。

下面是一个完整的 Python 示例代码。这个代码构建了一个 Agent,它初始时只有一个核心技能:create_new_tool。当它遇到无法直接解决的问题时,它会编写一个新的 Python 函数,注册它,然后立即使用这个新函数来解决问题。

前置准备

你需要安装 OpenAI 的 Python SDK:

pip install openai

完整代码实现

import os
import json
import inspect
from typing import Dict, Any, List, Callable
from openai import OpenAI

# ================= 配置部分 =================
# 这里以阿里云 DashScope (通义千问) 为例,也可以换成本地 vLLM/Ollama
# 如果是本地 Qwen,base_url 通常是 "http://localhost:8000/v1"
API_KEY = "sk-xxxxxxxxxxxxxxxxxxx"  # 替换你的 API Key
BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1" 
MODEL_NAME = "qwen-plus" # 或者 qwen-max, qwen-turbo, qwen2.5-72b-instruct

client = OpenAI(
    api_key=API_KEY,
    base_url=BASE_URL,
)

# ================= 技能注册表 =================

class SkillRegistry:
    def __init__(self):
        self.functions: Dict[str, Callable] = {}
        self.tools_schema: List[Dict] = []

    def register(self, func_name: str, func_callable: Callable, schema: Dict):
        """注册一个新函数及其 Schema"""
        self.functions[func_name] = func_callable
        self.tools_schema.append(schema)
        print(f" [系统] 新技能已注册: {func_name}")

    def get_function(self, name: str) -> Callable:
        return self.functions.get(name)

    def get_tools_schema(self) -> List[Dict]:
        return self.tools_schema

registry = SkillRegistry()

# ================= 核心:定义"创建技能"的元技能 =================

def create_new_tool(function_name: str, python_code: str, description: str, parameters_json: str):
    """
    这是一个元技能,允许 AI 编写新的 Python 代码并将其注册为工具。

    Args:
        function_name: 新函数的名称 (例如 'calculate_fibonacci')
        python_code: 可执行的 Python 函数代码字符串。
        description: 对该函数功能的描述。
        parameters_json: 符合 OpenAI JSON Schema 格式的参数定义字符串。
    """
    print(f" [Agent] 正在编写新技能: {function_name}...")

    # 1. 动态执行代码
    local_scope = {}
    try:
        # 警告:exec 在生产环境中极其危险,这里仅用于演示
        exec(python_code, globals(), local_scope)

        if function_name not in local_scope:
            return f"Error: The code executed successfully but function '{function_name}' was not found."

        new_func = local_scope[function_name]

    except Exception as e:
        return f"Error executing python code: {str(e)}"

    # 2. 构建 Tool Schema
    try:
        params_dict = json.loads(parameters_json)
    except json.JSONDecodeError:
        return "Error: parameters_json must be valid JSON string."

    new_tool_schema = {
        "type": "function",
        "function": {
            "name": function_name,
            "description": description,
            "parameters": params_dict
        }
    }

    # 3. 注册到全局注册表
    registry.register(function_name, new_func, new_tool_schema)

    return f"Success: Skill '{function_name}' created and registered. You can now call it."

# 初始化注册表,加入核心技能
create_tool_schema = {
    "type": "function",
    "function": {
        "name": "create_new_tool",
        "description": "当现有的工具无法满足需求时,编写一个新的 Python 函数并注册为工具。之后你可以立即调用这个新工具。",
        "parameters": {
            "type": "object",
            "properties": {
                "function_name": {"type": "string", "description": "函数名,例如 get_weather"},
                "python_code": {"type": "string", "description": "完整的 Python 函数定义代码"},
                "description": {"type": "string", "description": "工具的功能描述"},
                "parameters_json": {"type": "string", "description": "JSON Schema 格式的 parameters 字典 (JSON string)"}
            },
            "required": ["function_name", "python_code", "description", "parameters_json"]
        }
    }
}

registry.register("create_new_tool", create_new_tool, create_tool_schema)

# ================= 主循环逻辑 =================

def run_conversation(user_prompt):
    messages = [
        {"role": "system", "content": "你是一个能够自我进化的智能助手。如果你遇到无法直接回答的问题,请使用 'create_new_tool' 编写 Python 代码来解决,然后调用那个新工具。不要拒绝编写代码。"},
        {"role": "user", "content": user_prompt}
    ]

    print(f" [用户]: {user_prompt}")

    while True:
        # 1. 调用 Qwen API
        response = client.chat.completions.create(
            model=MODEL_NAME,
            messages=messages,
            tools=registry.get_tools_schema(), # 动态传入当前所有工具
            tool_choice="auto" 
        )

        response_message = response.choices[0].message

        # 2. 检查是否有工具调用
        tool_calls = response_message.tool_calls

        if tool_calls:
            # 将模型的回复(包含 tool_calls)加入历史,否则 API 会报错
            messages.append(response_message)

            for tool_call in tool_calls:
                func_name = tool_call.function.name
                func_args = json.loads(tool_call.function.arguments)

                print(f" [模型] 决定调用工具: {func_name}")

                # 获取实际函数
                function_to_call = registry.get_function(func_name)

                if not function_to_call:
                    tool_output = f"Error: Function {func_name} not found."
                else:
                    try:
                        # 执行函数
                        tool_output = str(function_to_call(**func_args))
                    except Exception as e:
                        tool_output = f"Error execution function: {str(e)}"

                print(f"   └── 结果: {tool_output[:100]}..." if len(tool_output) > 100 else f"   └── 结果: {tool_output}")

                # 将工具执行结果返回给模型
                messages.append({
                    "tool_call_id": tool_call.id,
                    "role": "tool",
                    "name": func_name,
                    "content": tool_output,
                })
        else:
            # 没有工具调用,说明模型已经生成了最终文本回复
            print(f" [模型最终回复]: {response_message.content}")
            break

# ================= 测试运行 =================

if __name__ == "__main__":
    # 场景 1: 询问一个 LLM 通常不擅长的数学计算,迫使它写代码
    # Qwen 需要计算第 30 个斐波那契数,它可能会选择写一个递归或循环函数
    query = "请帮我计算斐波那契数列的第 20 个数字是多少?如果你不知道,请写一个工具来算。"

    run_conversation(query)

    print("-" * 50)

    # 场景 2: 再次询问(此时它应该直接使用刚才创建的工具,或者创建新工具)
    query_2 = "那第 30 个斐波那契数字呢?"
    run_conversation(query_2)

代码原理解析

System Prompt (系统提示词)

告诉 LLM 它具备“自我进化”的能力。如果遇到问题,不要仅仅回复“我不知道”,而是通过 create_new_tool 编写代码。

create_new_tool 函数

这是整个系统的核心。它接收三个关键参数:

python_code: LLM 生成的纯文本 Python 代码。

function_name: 函数名。

parameters_json: 告诉 LLM 下次如何调用这个函数的 JSON Schema。

exec():这是 Python 的内置函数,用于执行字符串形式的代码。执行后,我们将生成的函数对象抓取出来,放入 registry (字典) 中。

动态 Tool Schema

在 client.chat.completions.create 中,参数 tools 并不是写死的。

每次循环时,我们都调用 registry.get_tools_schema()。这意味着,一旦 LLM 在第一轮对话中创建了新工具,第二轮对话时,这个新工具就会出现在 API 的可选工具列表中。

运行流程示例

当你运行上面的代码时,控制台输出大概如下:

用户: "计算斐波那契第 20 个数字。"

模型: 思考后发现没有斐波那契工具,决定调用 create_new_tool。

参数包含:def calculate_fibonacci(n): ... 的代码。

系统: 执行 exec,注册 calculate_fibonacci 到内存。

模型: 收到“技能创建成功”的反馈。

模型: 再次发起请求,这次它在 tools 列表里看到了 calculate_fibonacci。

模型: 决定调用 calculate_fibonacci(n=20)。

系统: 运行刚才生成的代码,返回结果 6765。

模型: 输出最终回答:“斐波那契数列第 20 个数字是 6765。”

安全性警告

这段代码使用了 exec() 函数,这在生产环境中是极其危险的。

如果 LLM 生成了 os.system('rm -rf /'),你的文件就会被删除。

改进方案

在 Docker 容器或沙箱(Sandbox)环境中运行 exec。

使用专门的代码解释器 API(如 E2B Code Interpreter)。

严格限制 LLM 生成代码中可导入的库(禁止 os, sys, subprocess 等)。

针对 Qwen (通义千问) 的提示

Qwen 模型(特别是 Qwen-2.5 系列)在 Tool Calling 和代码生成方面非常强大,非常适合这个任务。确保你的 parameters_json 描述清晰,Qwen 通常能一次性生成正确的 JSON Schema。

网友回复

我知道答案,我要回答