结构化工具调用的两个实现

AI 模型在与人类交互时通常使用自然语言进行交流。但如果它们需要与需要结构化格式的外部系统交互怎么办？工具调用，也称为功能调用，在这种情况下很有帮助，使大型语言模型（LLM）能够生成结构化的响应。

我们将代码分为三个不同的 Python 文件，分别用于 LangChain 和 Groq 实现：

app.py -> 运行模型的主要文件。
models.py -> 定义结构化响应模式以确保数据一致性。
tools.py -> 包含 LLM 将调用以执行操作的功能。

1、models.py

这段代码对 Groq 和 LangChain 实现都是通用的。

from enum import Enum  
  
class ActionEnum(str, Enum):  
    create = 'create'  
    write = 'write'  
    read = 'read'

定义允许的操作：这防止了 LLM 选择自己的操作，如 mkdir、run、delete，并强制它只能选择 create、write 或 read。

from pydantic import BaseModel, Field, ConfigDict  
  
class ToolCall(BaseModel):  
    model_config = ConfigDict(use_enum_values=True)  
  
    path: str = Field(description='文件名')  
    action: ActionEnum = Field(description='保存在文件中的内容。')  
    content: str = Field(description='传递给命令的参数。')

path — LLM 想要创建的文件/文件夹名称。
action — 必须是 "create"、"write" 或 "read" 中的一个。
content — 要写入文件的数据（如果适用）。

class ResponseModel(BaseModel):  
    tool_calls: list[ToolCall]

确保可以处理多个工具调用。

示例输出：

{
“tool_calls”: [
{“action”: “create”, “path”: “notes.txt”, “content”: “”},
{“action”: “write”, “path”: “notes.txt”, “content”: “这是示例。”}
]
}

2、tools.py

import os  
from typing import Literal  
from langchain.tools import tool  
  
@tool  
def run_command(path: str, action: Literal["create", "write", "read"], content: str = None):  
    """   
    处理创建文件或文件夹、将内容写入文件或将内容从文件中读取的操作。  
    """  
    dirname = os.path.dirname(path)  
      
    if dirname and not os.path.exists(dirname):  
        os.makedirs(dirname, exist_ok=True)  
      
    if action == 'create':  
        if '.' in os.path.basename(path):  
            with open(path, 'w', encoding='utf-8') as file:  
                file.write(content)  
                print(f"{path} 文件已创建。")  
        else:  
            os.makedirs(path, exist_ok=True)  
            print(f"{path} 文件夹已创建。")  
  
  
    elif action == 'write':  
        if not os.path.basename(path) or '.' not in os.path.basename(path):  
            raise ValueError("无法向文件夹写入内容。")  
          
        with open(path, 'w', encoding='utf-8') as file:  
            file.write(content)  
            print(f"内容已成功写入 {path}。")  
  
      
    elif action == 'read':  
        if not os.path.exists(path):  
            raise FileNotFoundError(f"未找到 {path} 的文件。")  
          
        if os.path.isdir(path):  
            raise IsADirectoryError('无法读取文件夹路径。')  
          
        with open(path, 'r', encoding='utf-8') as file:  
            print(f"正在读取 {path} 的内容。")  
            return file.read()  
          
      
    else:  
        raise ValueError(f"不支持的操作：{action}")

@tool 装饰器将此函数注册为 LangChain 中的可调用工具，我们不用于 Groq 实现。

该函数接受 三个参数：

path — 文件或文件夹名称。
action — 必须是 "create"、"write" 或 "read" 中的一个。
content — 要写入的数据（仅在 "write" 时需要）。

run_command ：这赋予 LLM 创建文件、写入内容和从文件中读取的能力。

3、app.py — LangChain 实现

import os  
import json  
import argparse  
  
from langchain.tools import StructuredTool  
from langchain_groq import ChatGroq  
from tools import run_command  
from models import ResponseModel  
  
load_dotenv()  
apiKey = os.getenv('GROQ_API_KEY')  
  
client = ChatGroq(temperature=0.6, model='deepseek-r1-distill-llama-70b', api_key=apiKey)

我们使用 ChatGroq 初始化一个 Groq 客户端，参数如下：

temperature=0.6 — 平衡创造力和精确度。
model='deepseek-r1-distill-llama-70b' — 指定一个 LLM 模型。
api_key=apiKey — 认证以访问 Groq 的 API。

run_command_tool = StructuredTool.from_function(  
        func=run_command,  
        name='run_command',  
        description="创建、写入或读取文件或文件夹。",  
        args_schema=ResponseModel  
    )

为了使 run_command 成为 LLM 可执行的工具，我们使用 LangChain 的 StructuredTool。

from_function(func=run_command) — 告诉 LangChain 这是 LLM 需要调用的函数。
name='run_command' — LLM 将通过名称 "run_command" 识别此工具。
description="创建、写入或读取文件或文件夹。" — 给 LLM 一个关于此函数功能的简短说明。
args_schema=ResponseModel — 确保函数遵循结构化的输入格式。

model = client.bind_tools([run_command_tool])

然后我们将 run_command 工具绑定到模型，以便当任务涉及文件处理时，它知道可以使用 run_command，而不是仅仅回复文本。

def run_model(query):  
  
    system_prompt = "你是一个有用的助手，帮助用户而不产生错误。"  
  
    messages = [  
        {"role": "system", "content": system_prompt},  
        {"role": "user", "content": query},  
    ]

messages 列表包含**，**系统提示，告诉 LLM 如何行为，以及用户查询，这是由用户发送给模型的内容。

例如：

“创建一个 Python 文件来预测给定年份是否为闰年。”

    llm_call = model.invoke(messages)  
    tool_calls = llm_call.additional_kwargs.get("tool_calls", [])

如果 AI 决定使用工具（因为它可能并不总是需要工具），我们提取工具调用从响应中。

    if tool_calls:  
        for tool_call in tool_calls:  
            selected_tool = {"run_command": run_command}[tool_call.get("function")["name"].lower()]  
            arguments = tool_call["function"]["arguments"]  
            for args in json.loads(arguments)["tool_calls"]:  
                selected_tool.invoke(args)  
      
    return "过程完成。"

由于我们只有一个工具 (run_command)，我们将 "run_command" 映射到实际的 Python 函数。
参数仍然是 JSON 字符串格式，所以我们使用 json.loads(arguments) 将其转换为字典。
LLM 可能会一次请求多个操作（例如，先创建一个文件夹，然后在其中写入文件），所以我们循环遍历 "tool_calls" 并逐个执行它们。
我们不需要返回任何花哨的东西，因为 LLM 已经执行了操作。

if __name__ == '__main__':  
    parser = argparse.ArgumentParser()  
    parser.add_argument("query", help="你想传递给模型的查询。例如：创建一个 Python 文件来预测给定年份是否为闰年。")  
    args = parser.parse_args()  
    run_model(args.query)

这段代码设置了一个命令行界面 (CLI)，允许用户在运行脚本时作为参数传递查询。它接收用户输入，将其发送给模型 (run_model(args.query))，并根据结构化工具调用框架执行任何必要的操作。

4、app.py — Groq 实现

import os  
import argparse  
from dotenv import load_dotenv  
  
import instructor  
from groq import Groq  
  
from models import ResponseModel  
from tools import run_command  
  
load_dotenv()  
apiKey = os.getenv('GROQ_API_KEY')  
  
client = instructor.from_groq(Groq(), model=instructor.Mode.JSON)

像我们之前做的那样，我们使用 instructor.from_groq() 初始化 Groq 客户端以获取结构化输出。

tool_schema = {  
    "name": "run_command",  
    "description": "创建、写入或读取文件或文件夹。",  
    "parameters": {  
        "type": "object",  
        "properties": {  
            "path": {  
                "type": "string",  
                "description": "以 / 结尾的文件夹名称或文件名。例如：app.py, env/"   
            },  
            "action": {  
                "type": "string",  
                "description": "需要执行的操作，只能是 create、read、write 中的一个。",  
                "enum" : ["create", "write", "read"]  
            },  
            "content": {  
                "type": "string",  
                "description": "需要写入文件的内容。"  
            }  
        },  
        "required": ["path", "action"]  
    }  
}

使用 enum 将 "action" 字段限制为 "create"、"write" 和 "read"，防止无效操作。
此模式确保 LLM 生成所需的字段。

def run_model(query):  
  
    system_prompt = "你是一个有用的助手，帮助用户而不产生错误。"  
  
    messages = [  
        {"role": "system", "content": system_prompt},  
        {"role": "user", "content": query},  
    ]

定义 messages 并设置系统提示和查询变量，如我们之前所做的。

llm_call = client.chat.completions.create(  
        model='deepseek-r1-distill-llama-70b',  
        response_model=ResponseModel,  
        messages=messages,  
        tools=tool_schema,  
        tool_choice="auto"  
    )

调用 Groq 的 LLM，传递结构化的消息。
response_model=ResponseModel 确保响应遵循预期格式。
tools=tool_schema 告诉模型它可以使用 run_command 工具。
tool_choice="auto" 允许模型决定是否调用工具或仅用文本回复。

    for tool_call in llm_call.tool_calls:  
        path = tool_call.path  
        action = tool_call.action  
        content = tool_call.content  
  
        run_command(path, action, content)  
  
    return None

检查 LLM 是否请求了任何工具执行。
如果是这样，它提取 path、action 和 content 字段并执行 run_command()。

5、执行结果

原文链接：Structured Tool Calling with LangChain and Groq

汇智网翻译整理，转载请标明出处