开发者大模型 API 接入指南

本文是面向开发者的大模型 API 实战手册，覆盖从基础接入到生产环境的完整知识：鉴权配置、流式输出、工具调用、RAG 构建、异步并发和成本控制。

环境配置

pip install openai anthropic

from openai import OpenAI

client = OpenAI(
    api_key="你的 APIMaster Key",
    base_url="https://apimaster.ai/v1",
)

推荐用环境变量管理密钥：

# .env 文件（不要提交到 git）
OPENAI_API_KEY=你的APIMaster Key
OPENAI_BASE_URL=https://apimaster.ai/v1

核心用法

基础对话

def 问答(问题: str, 模型: str = "claude-sonnet-4-6") -> str:
    resp = client.chat.completions.create(
        model=模型,
        messages=[{"role": "user", "content": 问题}],
    )
    return resp.choices[0].message.content

系统提示 + 多轮对话

class AI对话:
    def __init__(self, 系统提示: str, 模型: str = "claude-sonnet-4-6"):
        self.模型 = 模型
        self.消息历史 = [{"role": "system", "content": 系统提示}]
    
    def 发送(self, 用户消息: str) -> str:
        self.消息历史.append({"role": "user", "content": 用户消息})
        resp = client.chat.completions.create(
            model=self.模型,
            messages=self.消息历史,
        )
        回复 = resp.choices[0].message.content
        self.消息历史.append({"role": "assistant", "content": 回复})
        return 回复

助手 = AI对话("你是一个专业的 Python 代码审查员，提供简洁但准确的反馈。")
print(助手.发送("这个函数有什么问题？def add(a, b): return a+b"))
print(助手.发送("能否针对 NumPy 数组优化一下？"))

流式输出

def 流式生成(提示词: str, 模型: str = "gpt-4o"):
    with client.chat.completions.stream(
        model=模型,
        messages=[{"role": "user", "content": 提示词}],
    ) as stream:
        for text in stream.text_stream:
            yield text

# 使用示例
for 片段 in 流式生成("写一篇关于 Python 协程的技术博客"):
    print(片段, end="", flush=True)

JSON 结构化输出

import json

def 结构化提取(文本: str, 要提取的字段: list) -> dict:
    提示 = f"从文本中提取以下字段：{要提取的字段}，以 JSON 格式返回。\n\n文本：{文本}"
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": 提示}],
        response_format={"type": "json_object"},
    )
    return json.loads(resp.choices[0].message.content)

结果 = 结构化提取(
    "张明，北京科技公司 CTO，邮箱 zhangming@tech.com",
    ["姓名", "公司", "职位", "邮箱"]
)
print(结果)

工具调用（Function Calling）

import json

工具定义 = [
    {
        "type": "function",
        "function": {
            "name": "执行SQL查询",
            "description": "在数据库中执行只读查询",
            "parameters": {
                "type": "object",
                "properties": {
                    "sql": {"type": "string"},
                    "数据库": {"type": "string", "enum": ["用户库", "订单库"]},
                },
                "required": ["sql", "数据库"],
            },
        },
    }
]

def 执行工具(工具名, 参数):
    # 你的实际实现
    return {"结果": "模拟数据"}

def 智能代理(用户问题: str):
    消息 = [{"role": "user", "content": 用户问题}]
    
    while True:
        resp = client.chat.completions.create(
            model="gpt-4o",
            messages=消息,
            tools=工具定义,
        )
        
        if resp.choices[0].finish_reason != "tool_calls":
            return resp.choices[0].message.content
        
        消息.append(resp.choices[0].message)
        for tc in resp.choices[0].message.tool_calls:
            结果 = 执行工具(tc.function.name, json.loads(tc.function.arguments))
            消息.append({"role": "tool", "tool_call_id": tc.id, "content": json.dumps(结果, ensure_ascii=False)})

嵌入向量 + 语义搜索

import numpy as np

def 获取嵌入(文本列表: list) -> list:
    resp = client.embeddings.create(
        model="text-embedding-3-small",
        input=文本列表,
    )
    return [item.embedding for item in resp.data]

def 语义搜索(查询: str, 文档库: list, top_k: int = 3) -> list:
    查询向量 = 获取嵌入([查询])[0]
    文档向量 = 获取嵌入(文档库)
    
    def 余弦相似度(a, b):
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
    
    相似度 = [(i, 余弦相似度(查询向量, v)) for i, v in enumerate(文档向量)]
    相似度.sort(key=lambda x: x[1], reverse=True)
    
    return [文档库[i] for i, _ in 相似度[:top_k]]

RAG（检索增强生成）

def RAG问答(用户问题: str, 知识库: list) -> str:
    # 检索相关文档
    相关文档 = 语义搜索(用户问题, 知识库, top_k=3)
    上下文 = "\n\n".join(相关文档)
    
    # 基于检索内容生成回答
    resp = client.chat.completions.create(
        model="claude-sonnet-4-6",
        messages=[
            {"role": "system", "content": f"仅根据以下资料回答问题：\n\n{上下文}"},
            {"role": "user", "content": 用户问题},
        ],
    )
    return resp.choices[0].message.content

异步并发（批量处理）

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI(
    api_key="你的Key",
    base_url="https://apimaster.ai/v1",
)

async def 批量处理(任务列表: list) -> list:
    协程列表 = [
        async_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": 任务}],
            max_tokens=200,
        )
        for 任务 in 任务列表
    ]
    结果 = await asyncio.gather(*协程列表)
    return [r.choices[0].message.content for r in 结果]

# 并发处理 50 篇文章摘要
摘要列表 = asyncio.run(批量处理([f"总结：{a}" for a in 文章列表]))

生产环境清单

• API Key 存储在环境变量，不写入代码
• 429/500 错误的指数退避重试逻辑
• 设置 max_tokens 防止输出超长
• 长时任务使用流式输出
• 记录 Token 用量便于成本分析
• 根据任务复杂度路由到不同模型（节省成本）

模型选型速查

场景	推荐模型	原因
快速原型开发	deepseek-v4	便宜，代码能力强
生产聊天机器人	claude-haiku-4-5	快速、稳定
代码生成与审查	deepseek-v4 / claude-sonnet-4-6	代码最佳性价比
复杂文档分析	claude-sonnet-4-6	20 万 Token 上下文
批量文本处理	gpt-4o-mini	最便宜的通用模型

获取大模型 API 接入 → · 查看模型价格 →