开发者大模型 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 ask(question: str, model: str = "claude-sonnet-4-6") -> str:
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": question}],
    )
    return resp.choices[0].message.content

系统提示 + 多轮对话

class AIChat:
    def __init__(self, system_prompt: str, model: str = "claude-sonnet-4-6"):
        self.model = model
        self.history = [{"role": "system", "content": system_prompt}]
    
    def send(self, user_msg: str) -> str:
        self.history.append({"role": "user", "content": user_msg})
        resp = client.chat.completions.create(
            model=self.model,
            messages=self.history,
        )
        reply = resp.choices[0].message.content
        self.history.append({"role": "assistant", "content": reply})
        return reply

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

流式输出

def stream_generate(prompt: str, model: str = "gpt-5.4"):
    with client.chat.completions.stream(
        model=model,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for text in stream.text_stream:
            yield text

# 使用示例
for chunk in stream_generate("写一篇关于 Python 协程的技术博客"):
    print(chunk, end="", flush=True)

JSON 结构化输出

import json

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

result = extract_json(
    "张明，北京科技公司 CTO，邮箱 zhangming@tech.com",
    ["姓名", "公司", "职位", "邮箱"]
)
print(result)

工具调用（Function Calling）

import json

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

def run_tool(tool_name, args):
    # 你的实际实现
    return {"result": "mock data"}

def smart_agent(user_question: str):
    messages = [{"role": "user", "content": user_question}]
    
    while True:
        resp = client.chat.completions.create(
            model="gpt-5.4",
            messages=messages,
            tools=tool_defs,
        )
        
        if resp.choices[0].finish_reason != "tool_calls":
            return resp.choices[0].message.content
        
        messages.append(resp.choices[0].message)
        for tc in resp.choices[0].message.tool_calls:
            result = run_tool(tc.function.name, json.loads(tc.function.arguments))
            messages.append({"role": "tool", "tool_call_id": tc.id, "content": json.dumps(result, ensure_ascii=False)})

嵌入向量 + 语义搜索

import numpy as np

def get_embeddings(texts: list) -> list:
    resp = client.embeddings.create(
        model="text-embedding-3-small",
        input=texts,
    )
    return [item.embedding for item in resp.data]

def semantic_search(query: str, corpus: list, top_k: int = 3) -> list:
    query_vec = get_embeddings([query])[0]
    doc_vecs = get_embeddings(corpus)
    
    def cosine_sim(a, b):
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
    
    scores = [(i, cosine_sim(query_vec, v)) for i, v in enumerate(doc_vecs)]
    scores.sort(key=lambda x: x[1], reverse=True)
    
    return [corpus[i] for i, _ in scores[:top_k]]

RAG（检索增强生成）

def rag_qa(user_question: str, knowledge_base: list) -> str:
    # 检索相关文档
    relevant_docs = semantic_search(user_question, knowledge_base, top_k=3)
    context = "\n\n".join(relevant_docs)
    
    # 基于检索内容生成回答
    resp = client.chat.completions.create(
        model="claude-sonnet-4-6",
        messages=[
            {"role": "system", "content": f"仅根据以下资料回答问题：\n\n{context}"},
            {"role": "user", "content": user_question},
        ],
    )
    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 batch_process(tasks: list) -> list:
    coros = [
        async_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": task}],
            max_tokens=200,
        )
        for task in tasks
    ]
    results = await asyncio.gather(*coros)
    return [r.choices[0].message.content for r in results]

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

生产环境清单

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

模型选型速查

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

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