API集成

学习如何将LLM API集成到实际项目中。

LLM API服务商

主流服务商对比

OpenAI API集成

基础设置

import os
from dotenv import load_dotenv
from openai import OpenAI
# 加载环境变量
load_dotenv()
# 初始化客户端
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

文本生成

def generate_text(prompt, model="gpt-4o-mini", temperature=0.7, max_tokens=500):
"""
生成文本
Args:
prompt: 提示词
model: 模型名称
temperature: 温度（0-2，越高越随机）
max_tokens: 最大token数
Returns:
生成的文本
"""
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": "你是一个有帮助的助手。"},
{"role": "user", "content": prompt}
],
temperature=temperature,
max_tokens=max_tokens
)
return response.choices[0].message.content
# 使用示例
prompt = "用Python写一个快速排序算法"
result = generate_text(prompt)
print(result)

流式输出

def generate_text_stream(prompt, model="gpt-4o-mini"):
"""
流式生成文本（实时显示）
Args:
prompt: 提示词
model: 模型名称
"""
stream = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
stream=True
)
full_response = ""
print("AI回答: ", end="", flush=True)
for chunk in stream:
if chunk.choices[0].delta.content is not None:
content = chunk.choices[0].delta.content
print(content, end="", flush=True)
full_response += content
print()  # 换行
return full_response
# 使用
result = generate_text_stream("讲一个短故事")

对话历史管理

class ChatBot:
"""带历史记录的聊天机器人"""
def __init__(self, system_prompt="你是一个有帮助的助手。"):
self.client = OpenAI()
self.messages = [
{"role": "system", "content": system_prompt}
]
self.max_history = 10  # 保留最近10轮对话
def chat(self, user_message):
"""
发送消息并获取回复
Args:
user_message: 用户消息
Returns:
AI回复
"""
# 添加用户消息
self.messages.append({
"role": "user",
"content": user_message
})
# 调用API
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=self.messages
)
ai_message = response.choices[0].message.content
# 添加AI回复
self.messages.append({
"role": "assistant",
"content": ai_message
})
# 限制历史长度
if len(self.messages) > self.max_history * 2 + 1:
# 保留system消息和最近的对话
self.messages = (
[self.messages[0]] +
self.messages[-self.max_history * 2:]
)
return ai_message
def get_history(self):
"""获取对话历史"""
return self.messages[1:]  # 排除system消息
# 使用示例
bot = ChatBot("你是一位Python编程专家")
while True:
user_input = input("你: ")
if user_input.lower() == 'quit':
break
response = bot.chat(user_input)
print(f"AI: {response}")

智谱AI API集成

智谱AI提供中文优秀的模型，价格便宜。

import requests
import json
class ZhipuAIClient:
"""智谱AI客户端"""
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://open.bigmodel.cn/api/paas/v4"
def generate(self, prompt, model="glm-4-flash", temperature=0.7):
"""
生成文本
Args:
prompt: 提示词
model: 模型名称
temperature: 温度
Returns:
生成的文本
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
data = {
"model": model,
"messages": [
{"role": "user", "content": prompt}
],
"temperature": temperature
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=data
)
result = response.json()
return result["choices"][0]["message"]["content"]
# 使用示例
zhipu_client = ZhipuAIClient(os.getenv("ZHIPUAI_API_KEY"))
result = zhipu_client.generate("用中文解释什么是机器学习")
print(result)

DeepSeek API集成

DeepSeek在代码生成方面表现出色。

class DeepSeekClient:
"""DeepSeek客户端"""
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://api.deepseek.com"
def generate_code(self, task, language="Python"):
"""
生成代码
Args:
task: 任务描述
language: 编程语言
Returns:
生成的代码
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
prompt = f"""
作为{language}专家，请编写代码完成以下任务：
{task}
要求：
1. 代码完整可运行
2. 添加注释
3. 提供使用示例
"""
data = {
"model": "deepseek-chat",
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.3  # 代码生成用较低温度
}
response = requests.post(
f"{self.base_url}/v1/chat/completions",
headers=headers,
json=data
)
result = response.json()
return result["choices"][0]["message"]["content"]
# 使用示例
deepseek_client = DeepSeekClient(os.getenv("DEEPSEEK_API_KEY"))
code = deepseek_client.generate_code("实现一个二叉树遍历")
print(code)

LangChain集成

LangChain提供统一的接口，轻松切换不同模型。

基础使用

from langchain_openai import ChatOpenAI, OpenAI
from langchain_community.llms import Ollama
# OpenAI
openai_llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0.7
)
# 使用
response = openai_llm.invoke("你好")
print(response.content)

切换模型

def get_llm(provider="openai"):
"""
根据提供商获取LLM
Args:
provider: openai, zhipuai, deepseek, ollama
Returns:
LLM实例
"""
if provider == "openai":
return ChatOpenAI(model="gpt-4o-mini")
elif provider == "zhipuai":
from langchain_community.chat_models import ChatZhipuAI
return ChatZhipuAI(model="glm-4-flash", api_key=os.getenv("ZHIPUAI_API_KEY"))
elif provider == "deepseek":
from langchain_community.chat_models import ChatDeepSeek
return ChatDeepSeek(api_key=os.getenv("DEEPSEEK_API_KEY"))
elif provider == "ollama":
from langchain_community.llms import Ollama
return Ollama(model="mistral")
else:
raise ValueError(f"未知的provider: {provider}")
# 使用
llm = get_llm("zhipuai")  # 切换到智谱AI
response = llm.invoke("你好")
print(response.content)

错误处理

常见错误处理

import time
from openai import OpenAI
from openai import RateLimitError, APITimeoutError
def safe_generate(prompt, max_retries=3, retry_delay=5):
"""
带重试的生成函数
Args:
prompt: 提示词
max_retries: 最大重试次数
retry_delay: 重试延迟（秒）
Returns:
生成的文本或None
"""
client = OpenAI()
for attempt in range(max_retries):
try:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
except RateLimitError:
print(f"⚠️ 速率限制，{retry_delay}秒后重试...")
time.sleep(retry_delay)
except APITimeoutError:
print(f"⚠️ 请求超时，{retry_delay}秒后重试...")
time.sleep(retry_delay)
except Exception as e:
print(f"⚠️ 错误: {e}")
return None
print("❌ 达到最大重试次数，生成失败")
return None
# 使用
result = safe_generate("写一个故事")
if result:
print(result)

成本控制

Token计数

import tiktoken
def count_tokens(text, model="gpt-4o"):
"""
计算token数量
Args:
text: 文本
model: 模型名称
Returns:
token数量
"""
encoding = tiktoken.encoding_for_model(model)
return len(encoding.encode(text))
# 使用示例
text = "Hello, world!"
token_count = count_tokens(text)
print(f"Token数量: {token_count}")
# 计算成本（GPT-4o-mini定价: $0.15/1M input tokens, $0.60/1M output tokens）
input_cost = token_count * 0.15 / 1_000_000
output_cost = token_count * 0.60 / 1_000_000
print(f"预估成本: ${input_cost:.6f} (input), ${output_cost:.6f} (output)")

成本追踪器

class CostTracker:
"""API成本追踪器"""
def __init__(self):
self.total_input_tokens = 0
self.total_output_tokens = 0
# 定价（美元/百万tokens）
self.pricing = {
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
"gpt-4o": {"input": 2.50, "output": 10.00},
"gpt-4-turbo": {"input": 10.00, "output": 30.00}
}
def track(self, model, input_tokens, output_tokens):
"""追踪一次调用"""
self.total_input_tokens += input_tokens
self.total_output_tokens += output_tokens
def get_cost(self, model):
"""计算总成本"""
pricing = self.pricing.get(model, self.pricing["gpt-4o-mini"])
input_cost = self.total_input_tokens * pricing["input"] / 1_000_000
output_cost = self.total_output_tokens * pricing["output"] / 1_000_000
return input_cost + output_cost
def print_summary(self, model="gpt-4o-mini"):
"""打印摘要"""
cost = self.get_cost(model)
print(f"\n📊 成本摘要:")
print(f"  输入tokens: {self.total_input_tokens:,}")
print(f"  输出tokens: {self.total_output_tokens:,}")
print(f"  总成本: ${cost:.4f}")
# 使用
tracker = CostTracker()
# 模拟一些API调用
tracker.track("gpt-4o-mini", 1000, 500)
tracker.track("gpt-4o-mini", 2000, 1000)
tracker.print_summary()

实践项目

项目1: 邮件回复助手

class EmailAssistant:
"""邮件回复助手"""
def __init__(self):
self.client = OpenAI()
def generate_reply(self, email_content, tone="professional"):
"""
生成邮件回复
Args:
email_content: 原始邮件内容
tone: 语气风格
Returns:
生成的回复
"""
prompt = f"""
作为邮件回复助手，请为以下邮件生成一个{tone}的回复：
原始邮件：
{email_content}
要求：
1. 礼貌得体
2. 回应要点
3. 清晰简洁
4. 150字以内
回复：
"""
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
# 使用
assistant = EmailAssistant()
email = """
您好，感谢您的来信。我们注意到您的订单尚未付款，
请在3天内完成付款，否则订单将自动取消。
"""
reply = assistant.generate_reply(email)
print(f"回复: {reply}")

项目2: 代码审查助手

class CodeReviewer:
"""代码审查助手"""
def __init__(self):
self.client = OpenAI()
def review(self, code, language="Python"):
"""
审查代码
Args:
code: 代码
language: 编程语言
Returns:
审查报告
"""
prompt = f"""
作为{language}代码审查专家，请审查以下代码：
```{language.lower()}
{code}

审查要点： 1. 代码风格 2. 潜在bug 3. 性能优化 4. 安全问题 5. 可读性

请用JSON格式输出： {{ "summary": "总体评价", "issues": ["问题1", "问题2"], "suggestions": ["建议1", "建议2"], "rating": 1-5 }} """

    response = self.client.chat.completions.create(
model="gpt-4o",  # 使用更强的模型
messages=[{"role": "user", "content": prompt}],
temperature=0.3
)
return response.choices[0].message.content

使用

reviewer = CodeReviewer()

code = """ def add(a, b): return a + b

result = add(5, "10") """

review = reviewer.review(code) print(review) ```

学习要点

✅ OpenAI、智谱AI、DeepSeek等提供不同优势的API ✅ 流式输出提供更好的用户体验 ✅ LangChain统一接口便于切换模型 ✅ 错误处理和重试机制保证稳定性 ✅ Token计数和成本控制避免意外开销

下一步: 构建 第一个LLM应用 🚀