Day9：函数封装与参数校验：打造稳健的大模型工具集

摘要：在使用大语言模型进行开发时，将业务逻辑封装为可调用函数，并对参数进行有效校验和处理函数调用异常至关重要。本文以通义千问大模型应用场景为背景，深入讲解如何将业务逻辑封装成函数，借助 pydantic 库实现参数校验，以及妥善处理函数调用过程中出现的各种异常。同时，通过封装 “查询天气”“搜索内容”“读取本地文件” 这三个实用工具函数，进一步巩固相关知识和技能。

一、引言

在与大语言模型交互的过程中，为了提高代码的复用性、可读性以及稳定性，我们需要将具体的业务逻辑封装为函数，并确保传入函数的参数符合要求，同时能够妥善处理函数调用时可能出现的异常情况。今天，我们将围绕函数封装、参数校验以及异常处理展开学习，并以通义千问大模型应用为实例进行实践。

二、核心任务

（一）将业务逻辑封装为可调用函数

把业务逻辑封装为函数，就是将完成特定任务的代码片段进行整合，通过定义函数名、参数列表和函数体，使其成为一个可重复调用的模块。例如，如果我们有一段代码用于处理用户输入并调用通义千问模型获取回答，就可以将其封装为一个函数：

import requests
import json


def call_qwen(prompt, api_key, model="qwen-plus"):
    url = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api_key}"
    }
    data = {
        "model": model,
        "messages": [
            {"role": "user", "content": prompt},
        ],
        "stream": False
    }
    response = requests.post(url, headers=headers, json=data)
    if response.status_code == 200:
        return response.json()
    else:
        return None

（二）添加参数校验（使用 pydantic）

pydantic 是一个用于数据校验和设置数据类型提示的库。它可以帮助我们确保传入函数的参数符合特定的类型和规则。首先，需要安装 pydantic：

pip install pydantic

然后，我们可以使用 pydantic 来定义参数模型，并在函数中进行参数校验。以之前的 call_qwen 函数为例：

from pydantic import BaseModel, field_validator


class QwenCallParams(BaseModel):
    prompt: str
    api_key: str
    model: str = "qwen-plus"

    @field_validator('prompt')
    def prompt_not_empty(cls, v):
        if not v.strip():
            raise ValueError('prompt不能为空')
        return v

    @field_validator('api_key')
    def api_key_not_empty(cls, v):
        if not v.strip():
            raise ValueError('api_key不能为空')
        return v


def call_qwen(params: QwenCallParams):
    url = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {params.api_key}"
    }
    data = {
        "model": params.model,
        "messages": [
            {"role": "user", "content": params.prompt},
        ],
        "stream": False
    }
    response = requests.post(url, headers=headers, json=data)
    if response.status_code == 200:
        return response.json()
    else:
        return None

（三）处理函数调用异常

在函数调用过程中，可能会出现各种异常情况，如参数错误、调用失败等。我们需要对这些异常进行捕获和处理，以提供更友好的用户体验和稳定的程序运行。

1. 参数错误处理：pydantic 已经帮我们处理了大部分参数类型和规则校验，如果参数不符合要求，会抛出相应的异常。我们可以在调用函数时捕获这些异常并进行处理。

   try:
       params = QwenCallParams(prompt="", api_key="")
       result = call_qwen(params)
   except ValueError as e:
       print(f"参数错误: {e}")
   
   

2. 调用失败处理：对于函数内部的调用失败情况，比如网络问题导致无法连接到通义千问的 API，我们可以捕获 requests 库可能抛出的异常。

import requests


def call_qwen(params: QwenCallParams):
    url = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {params.api_key}"
    }
    data = {
        "model": params.model,
        "messages": [
            {"role": "user", "content": params.prompt},
        ],
        "stream": False
    }
    try:
        response = requests.post(url, headers=headers, json=data)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        print(f"调用失败: {e}")
        return None

三、补充任务：封装 3 个实用工具函数

（一）“查询天气” 函数

import requests
from pydantic import BaseModel, field_validator


class WeatherQueryParams(BaseModel):
    city: str
    api_key: str

    @field_validator('city')
    def city_not_empty(cls, v):
        if not v.strip():
            raise ValueError('城市名称不能为空')
        return v

    @field_validator('api_key')
    def api_key_not_empty(cls, v):
        if not v.strip():
            raise ValueError('api_key不能为空')
        return v


def query_weather(params: WeatherQueryParams):
    base_url = "http://api.weatherapi.com/v1/current.json"
    params_dict = {
        "key": params.api_key,
        "q": params.city
    }
    try:
        response = requests.get(base_url, params=params_dict)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        print(f"查询天气失败: {e}")
        return None

（二）“搜索内容” 函数

import requests
from pydantic import BaseModel, field_validator


class SearchParams(BaseModel):
    query: str
    api_key: str

    @field_validator('query')
    def query_not_empty(cls, v):
        if not v.strip():
            raise ValueError('查询内容不能为空')
        return v

    @field_validator('api_key')
    def api_key_not_empty(cls, v):
        if not v.strip():
            raise ValueError('api_key不能为空')
        return v


def search_content(params: SearchParams):
    base_url = "https://api.example.com/search"  # 假设的搜索API地址
    params_dict = {
        "key": params.api_key,
        "q": params.query
    }
    try:
        response = requests.get(base_url, params=params_dict)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        print(f"搜索内容失败: {e}")
        return None

（三）“读取本地文件” 函数

from pydantic import BaseModel, field_validator
import os


class ReadLocalFileParams(BaseModel):
    file_path: str

    @field_validator('file_path')
    def file_path_exists(cls, v):
        if not os.path.exists(v):
            raise ValueError('文件路径不存在')
        return v


def read_local_file(params: ReadLocalFileParams):
    try:
        with open(params.file_path, 'r', encoding='utf - 8') as f:
            return f.read()
    except Exception as e:
        print(f"读取本地文件失败: {e}")
        return None

四、总结

通过今天的学习，我们掌握了如何将业务逻辑封装为可调用函数，利用 pydantic 库进行参数校验，以及妥善处理函数调用过程中出现的异常情况。并且通过封装 “查询天气”“搜索内容”“读取本地文件” 这三个实用工具函数，进一步加深了对这些知识的理解和运用。

在实际开发中，良好的函数封装、参数校验和异常处理能够提高代码的质量和稳定性，降低维护成本。希望大家能够熟练运用这些技能，在与大语言模型相关的开发工作中更加得心应手。

记得将封装好的 3 个工具函数整理保存，方便后续使用和参考。如果在学习过程中遇到问题，可以回顾相关知识点，或者查阅 pydantic 和 requests 库的官方文档进行解决。