# 大模型API调用与集成

一、概述

为什么要学习这个主题

随着GPT-4、Claude、文心一言、通义千问等大语言模型的爆发式发展，AI能力已经从实验室走向了生产环境。企业不再满足于在网页端手动使用AI，而是希望将大模型的能力嵌入到自己的应用、工作流和产品中。无论是智能客服、代码辅助、内容生成还是数据分析，大模型API都是连接AI能力与业务场景的核心桥梁。

然而，在实际集成过程中，开发者面临诸多挑战：不同厂商API格式不统一、流式响应处理、函数调用（Function Calling）的复杂逻辑、Token消耗的成本控制等。本课程将帮助你系统掌握大模型API调用的核心技术，让你能够高效、稳定、经济地将大模型能力集成到自己的应用中。

学完本课程能做什么

1. 掌握主流大模型API调用方法：熟练使用OpenAI、Claude、国内大模型（如通义千问、文心一言）的API，并能快速切换不同服务商。 2. 实现流式输出：让AI的回复像ChatGPT一样逐字显示，提升用户体验。 3. 使用函数调用（Function Calling）：让大模型能够调用你定义的函数，实现结构化数据提取、工具调用等高级功能。 4. 管理Token与成本：精确计算每次调用的Token消耗，设计合理的成本控制策略。 5. 构建兼容OpenAI接口的服务：理解OpenAI兼容格式，能够对接第三方代理或自建服务。

适合人群和前置知识要求

**适合人群**：有1-3年开发经验的软件工程师、AI应用开发者、后端工程师、技术负责人。
**前置知识**：
熟悉Python语言基础（会使用`requests`库）
了解HTTP协议和RESTful API基本概念
对JSON数据格式有基本认识
了解大模型的基本概念（如Prompt、Token）

---

二、核心知识点

模块一：主流大模型API对比与选型

原理讲解

不同大模型厂商的API在以下方面存在差异：

**接口地址**：每个服务商有独立的Endpoint
**认证方式**：API Key、Access Token、Bearer Token等
**请求格式**：参数名称、数据结构不同
**响应格式**：返回的JSON结构不同
**能力差异**：支持的上下文长度、函数调用能力、多模态支持等

主流模型对比

| 模型 | 厂商 | 接口地址示例 | 上下文长度 | 函数调用 | 流式支持 | 价格（参考） | |------|------|--------------|------------|----------|----------|--------------| | GPT-4o | OpenAI | https://api.openai.com/v1/chat/completions | 128K | ✅ | ✅ | $5/1M input tokens | | Claude 3.5 Sonnet | Anthropic | https://api.anthropic.com/v1/messages | 200K | ✅ | ✅ | $3/1M input tokens | | Qwen-Max | 阿里云 | https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation | 32K | ✅ | ✅ | ¥0.04/1K tokens | | ERNIE-4.0 | 百度 | https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/completions | 8K | ✅ | ✅ | ¥0.12/1K tokens |

实际代码示例：统一调用接口

由于各厂商API差异较大，建议封装一个统一的调用层。以下是一个简单的抽象示例：


import requests
import json

class LLMClient:
def __init__(self, provider, api_key, base_url=None):
self.provider = provider
self.api_key = api_key
self.base_url = base_url

def chat(self, messages, model=None, **kwargs):
if self.provider == "openai":
return self._openai_chat(messages, model, **kwargs)
elif self.provider == "anthropic":
return self._anthropic_chat(messages, model, **kwargs)
elif self.provider == "qwen":
return self._qwen_chat(messages, model, **kwargs)
else:
raise ValueError(f"Unsupported provider: {self.provider}")

def _openai_chat(self, messages, model="gpt-4o", **kwargs):
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
data = {
"model": model,
"messages": messages,
**kwargs
}
url = self.base_url or "https://api.openai.com/v1/chat/completions"
response = requests.post(url, headers=headers, json=data)
return response.json()


模块二：OpenAI兼容接口

原理讲解

OpenAI的API格式已成为事实标准，许多其他厂商（如通义千问、DeepSeek、智谱GLM）都提供了兼容OpenAI格式的接口。这意味着你只需修改base_url和api_key，就能用同一套代码调用不同模型。

关键参数说明

| 参数 | 类型 | 说明 | 示例值 |
|------|------|------|--------|
| model | string | 模型名称 | gpt-4o |
| messages | array | 对话消息列表 | [{"role": "user", "content": "Hello"}] |
| temperature | float | 随机性（0-2） | 0.7 |
| max_tokens | integer | 最大输出Token数 | 2048 |
| stream | boolean | 是否流式输出 | false |
| tools | array | 函数定义列表 | [{...}] |

实际代码示例：切换不同厂商


# OpenAI兼容格式调用示例
import requests

def call_compatible_api(base_url, api_key, messages, model="default"):
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
data = {
"model": model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 2048
}
response = requests.post(f"{base_url}/v1/chat/completions",
headers=headers, json=data)
return response.json()

# 使用OpenAI
openai_response = call_compatible_api(
base_url="https://api.openai.com",
api_key="sk-xxx",
messages=[{"role": "user", "content": "你好"}],
model="gpt-4o"
)

# 使用通义千问（兼容模式）
qwen_response = call_compatible_api(
base_url="https://dashscope.aliyuncs.com/compatible-mode",
api_key="sk-xxx",
messages=[{"role": "user", "content": "你好"}],
model="qwen-plus"
)


模块三：流式调用

原理讲解

流式调用（Streaming）通过Server-Sent Events（SSE）实现。大模型生成内容时，服务端会逐块（chunk）返回数据，客户端可以实时处理每个数据块，从而实现逐字显示的效果。

关键参数

`stream: true`：开启流式模式
响应类型：`text/event-stream`
数据格式：`data: {...}\n\n`


实际代码示例：流式输出


import requests
import json

def stream_chat(api_key, messages, model="gpt-4o"):
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
data = {
"model": model,
"messages": messages,
"stream": True
}

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

full_content = ""
for line in response.iter_lines():
if line:
line = line.decode('utf-8')
if line.startswith("data: "):
data_str = line[6:]
if data_str == "[DONE]":
break
try:
chunk = json.loads(data_str)
if chunk['choices'][0]['delta'].get('content'):
content = chunk['choices'][0]['delta']['content']
full_content += content
print(content, end='', flush=True)
except json.JSONDecodeError:
continue
return full_content

# 使用示例
stream_chat("sk-xxx", [{"role": "user", "content": "讲一个笑话"}])


模块四：函数调用（Function Calling）

原理讲解

函数调用允许大模型根据用户输入，自动决定调用哪个函数、传入什么参数。模型不实际执行函数，而是返回函数名和参数，由开发者决定如何执行。这使大模型能够：

从自然语言中提取结构化数据
调用外部工具（如搜索引擎、数据库）
执行计算或业务逻辑


函数定义格式


tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如北京、上海"
},
"date": {
"type": "string",
"description": "日期，格式YYYY-MM-DD"
}
},
"required": ["city"]
}
}
}
]


实际代码示例：完整函数调用流程


import requests
import json

def call_with_tools(api_key, messages, tools):
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
data = {
"model": "gpt-4o",
"messages": messages,
"tools": tools,
"tool_choice": "auto"  # 让模型自动选择是否调用函数
}

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

# 模拟天气查询函数
def get_weather(city, date="today"):
# 实际项目中这里会调用真实API
weather_data = {
"北京": {"temperature": "25°C", "condition": "晴"},
"上海": {"temperature": "28°C", "condition": "多云"},
}
return weather_data.get(city, {"temperature": "未知", "condition": "未知"})

# 使用示例
messages = [{"role": "user", "content": "北京今天天气怎么样？"}]
response = call_with_tools("sk-xxx", messages, tools)

# 处理函数调用
if response['choices'][0]['message'].get('tool_calls'):
tool_call = response['choices'][0]['message']['tool_calls'][0]
function_name = tool_call['function']['name']
arguments = json.loads(tool_call['function']['arguments'])

if function_name == "get_weather":
result = get_weather(arguments['city'], arguments.get('date', 'today'))
print(f"天气查询结果：{result}")


---

三、实操步骤

步骤一：准备开发环境

1. 安装Python依赖

pip install requests python-dotenv


2. 创建配置文件（.env）

OPENAI_API_KEY=sk-your-key-here
QWEN_API_KEY=sk-your-qwen-key
ANTHROPIC_API_KEY=sk-ant-your-key


3. 创建主程序文件（llm_integration.py）

步骤二：实现基础API调用

1. 编写统一调用函数

import os
from dotenv import load_dotenv
import requests

load_dotenv()

def call_llm(messages, model="gpt-4o", stream=False):
api_key = os.getenv("OPENAI_API_KEY")
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
data = {
"model": model,
"messages": messages,
"stream": stream
}
response = requests.post(
"https://api.openai.com/v1/chat/completions",
headers=headers,
json=data,
stream=stream
)
return response


2. 测试调用

# 测试代码
messages = [{"role": "user", "content": "用一句话解释什么是API"}]
response = call_llm(messages)
print(response.json()['choices'][0]['message']['content'])


预期效果：控制台输出类似"API是应用程序之间通信的接口，允许不同软件系统相互交互和交换数据。"

步骤三：实现流式对话功能

1. 添加流式处理函数

def stream_response(messages, model="gpt-4o"):
response = call_llm(messages, model, stream=True)
full_text = ""
for line in response.iter_lines():
if line:
line = line.decode('utf-8')
if line.startswith("data: ") and line[6:] != "[DONE]":
chunk = json.loads(line[6:])
if chunk['choices'][0]['delta'].get('content'):
text = chunk['choices'][0]['delta']['content']
full_text += text
print(text, end='', flush=True)
print()  # 换行
return full_text


2. 实现交互式对话

def interactive_chat():
print("AI助手已启动（输入'quit'退出）")
messages = []
while True:
user_input = input("\n你: ")
if user_input.lower() == 'quit':
break
messages.append({"role": "user", "content": user_input})
print("AI: ", end='')
response = stream_response(messages)
messages.append({"role": "assistant", "content": response})

if __name__ == "__main__":
interactive_chat()


预期效果：程序启动后，可以像ChatGPT一样进行流式对话，AI回复会逐字显示。

步骤四：集成函数调用

1. 定义工具函数

def calculate(expression):
"""简单的计算器函数"""
try:
result = eval(expression)
return {"result": result}
except Exception as e:
return {"error": str(e)}

tools = [
{
"type": "function",
"function": {
"name": "calculate",
"description": "执行数学计算",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式，如 2+3*4"
}
},
"required": ["expression"]
}
}
}
]


2. 实现函数调用逻辑

def chat_with_tools(messages, tools):
response = call_llm(messages, tools=tools)
response_data = response.json()

# 检查是否需要调用函数
if response_data['choices'][0]['message'].get('tool_calls'):
tool_call = response_data['choices'][0]['message']['tool_calls'][0]
function_name = tool_call['function']['name']
arguments = json.loads(tool_call['function']['arguments'])

# 执行函数
if function_name == "calculate":
function_result = calculate(arguments['expression'])

# 将函数结果返回给模型
messages.append(response_data['choices'][0]['message'])
messages.append({
"role": "tool",
"tool_call_id": tool_call['id'],
"content": json.dumps(function_result)
})

# 获取最终回复
final_response = call_llm(messages)
return final_response.json()['choices'][0]['message']['content']
else:
return response_data['choices'][0]['message']['content']

# 测试
messages = [{"role": "user", "content": "计算 23 * 45 的结果"}]
result = chat_with_tools(messages, tools)
print(result)  # 输出：23 * 45 = 1035


预期效果：模型会调用calculate函数计算结果，并返回自然语言描述。

---

四、常见问题与故障排查

问题1：API调用返回401 Unauthorized

原因：API Key无效、过期或格式错误
排查流程：
1. 检查API Key是否在有效期内
2. 确认API Key拼写正确，没有多余空格
3. 检查认证头格式是否正确
4. 尝试在官方平台测试API Key

解决方案：

# 错误的认证方式
headers = {"Authorization": f"API Key {api_key}"}  # 错误！
# 正确的认证方式
headers = {"Authorization": f"Bearer {api_key}"}   # 正确！


问题2：流式输出乱码或中断

原因：网络不稳定、编码问题、连接超时
排查流程：
1. 检查网络连接是否稳定
2. 确认设置了正确的编码（UTF-8）
3. 增加超时时间设置
4. 检查是否有防火墙拦截

解决方案：

# 增加超时设置和错误处理
try:
response = requests.post(url, headers=headers, json=data,
stream=True, timeout=30)
response