AI 工具调用新手入门：从零搭建你的第一个智能服务

1次阅读

共计 2655 个字符，预计需要花费 7 分钟才能阅读完成。

AI 工具调用已经成为现代开发中的常见需求，无论是构建智能客服、内容生成还是数据分析应用，都离不开对 AI 服务的调用。最常用的方式是通过 REST API 或 SDK 来集成 AI 能力。REST API 是一种基于 HTTP 协议的接口调用方式，而 SDK 则是封装好的开发工具包，简化了调用过程。

主流 AI 服务提供商包括 OpenAI、Azure AI、Google Cloud AI 等。它们在调用方式上各有特点：

OpenAI：提供简单的 API Key 认证，适合快速上手。
Azure AI：集成在 Azure 云平台中，适合企业级应用。
Google Cloud AI：功能强大，但配置较为复杂。

API Key 是最常见的认证方式。以下是一个 Python 示例，展示如何管理 API Key：

import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 获取 API Key
api_key = os.getenv('OPENAI_API_KEY')
if not api_key:
    raise ValueError("API Key not found in environment variables")

以下是一个完整的 Python 示例，展示如何调用 OpenAI 的文本生成 API：

import requests

headers = {"Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}

payload = {
    "model": "gpt-3.5-turbo",
    "messages": [{"role": "user", "content": "Hello, how are you?"}]
}

response = requests.post("https://api.openai.com/v1/chat/completions", headers=headers, json=payload)

if response.status_code == 200:
    print(response.json())
else:
    print(f"Error: {response.status_code}")

在网络请求中，错误处理和重试机制至关重要：

import time

max_retries = 3
retry_delay = 2

for attempt in range(max_retries):
    try:
        response = requests.post("https://api.openai.com/v1/chat/completions", headers=headers, json=payload)
        response.raise_for_status()
        print(response.json())
        break
    except requests.exceptions.RequestException as e:
        print(f"Attempt {attempt + 1} failed: {e}")
        if attempt < max_retries - 1:
            time.sleep(retry_delay)
        else:
            print("Max retries reached")

通过批量发送请求，可以减少网络延迟：

batch_payload = {
    "model": "gpt-3.5-turbo",
    "messages": [{"role": "user", "content": "Hello, how are you?"},
        {"role": "user", "content": "What is the weather today?"}
    ]
}

缓存常见响应可以显著提高性能：

from functools import lru_cache

@lru_cache(maxsize=100)
def get_cached_response(prompt):
    response = requests.post("https://api.openai.com/v1/chat/completions", headers=headers, json={"model": "gpt-3.5-turbo", "messages": [{"role": "user", "content": prompt}]})
    return response.json()

使用多线程或异步 IO 可以提高吞吐量：

import concurrent.futures

with concurrent.futures.ThreadPoolExecutor() as executor:
    futures = [executor.submit(get_cached_response, prompt) for prompt in prompts]
    results = [future.result() for future in concurrent.futures.as_completed(futures)]

API Key 泄露 ：永远不要将 API Key 硬编码在代码中，使用环境变量或密钥管理服务。
超时设置 ：未设置超时可能导致请求 hang 住，建议设置合理的超时时间。
响应解析错误 ：确保正确处理 API 返回的 JSON 数据，避免因字段缺失导致程序崩溃。
频率限制 ：了解服务的频率限制，避免因频繁调用被封禁。
模型选择不当 ：根据需求选择合适的模型，避免资源浪费。

创建一个 Flask 应用，接收用户输入。
调用 OpenAI API 生成回复。
将回复返回给用户。

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/chat', methods=['POST'])
def chat():
    user_input = request.json.get('message')
    response = get_cached_response(user_input)
    return jsonify({"reply": response["choices"][0]["message"]["content"]})

if __name__ == '__main__':
    app.run()