
如何快速实现REST API集成以优化业务流程
本文将通过具体案例,详细介绍如何接入和使用主流大模型的API,包含完整代码示例和最佳实践。
安装依赖
pip install openai
初始化配置
from openai import OpenAI
client = OpenAI(
api_key='your_api_key',
base_url='https://api.openai.com/v1' # 可选,默认即为此值
)
聊天对话
def chat_with_gpt(prompt, model="gpt-4-turbo-preview"):
try:
response = client.chat.completions.create(
model=model,
messages=[
{"role": "user", "content": prompt}
],
temperature=0.7,
max_tokens=2000
)
return response.choices[0].message.content
except Exception as e:
print(f"Error: {str(e)}")
return None
# 使用示例
result = chat_with_gpt("请帮我写一个Python函数来计算斐波那契数列")
print(result)
流式输出
def stream_chat(prompt):
try:
stream = client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=[{"role": "user", "content": prompt}],
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="")
except Exception as e:
print(f"Error: {str(e)}")
# 使用示例
stream_chat("讲一个关于AI的故事")
安装依赖
pip install anthropic
初始化配置
from anthropic import Anthropic
anthropic = Anthropic(
api_key="your_api_key"
)
基础对话
def chat_with_claude(prompt, model="claude-3-opus-20240229"):
try:
message = anthropic.messages.create(
model=model,
max_tokens=2000,
temperature=0.7,
messages=[
{"role": "user", "content": prompt}
]
)
return message.content[0].text
except Exception as e:
print(f"Error: {str(e)}")
return None
# 使用示例
result = chat_with_claude("帮我分析一下Python和JavaScript的异同")
print(result)
系统提示使用
def chat_with_system_prompt(user_prompt, system_prompt):
try:
message = anthropic.messages.create(
model="claude-3-opus-20240229",
system=system_prompt,
messages=[
{"role": "user", "content": user_prompt}
]
)
return message.content[0].text
except Exception as e:
print(f"Error: {str(e)}")
return None
# 使用示例
system = "你是一个专业的技术文档翻译专家"
user_message = "请翻译以下技术文档..."
result = chat_with_system_prompt(user_message, system)
安装依赖
pip install google-generativeai
初始化配置
import google.generativeai as genai
genai.configure(api_key='your_api_key')
文本生成
def chat_with_gemini(prompt, model="gemini-pro"):
try:
model = genai.GenerativeModel(model)
response = model.generate_content(prompt)
return response.text
except Exception as e:
print(f"Error: {str(e)}")
return None
# 使用示例
result = chat_with_gemini("解释什么是量子计算")
print(result)
多轮对话
def chat_session():
try:
model = genai.GenerativeModel('gemini-pro')
chat = model.start_chat(history=[])
while True:
user_input = input("You: ")
if user_input.lower() == 'exit':
break
response = chat.send_message(user_input)
print(f"Gemini: {response.text}\n")
except Exception as e:
print(f"Error: {str(e)}")
# 使用示例
chat_session()
class AIModelAPI:
def __init__(self):
self.openai_client = OpenAI(api_key='your_openai_key')
self.anthropic = Anthropic(api_key='your_anthropic_key')
genai.configure(api_key='your_gemini_key')
self.gemini = genai.GenerativeModel('gemini-pro')
def get_completion(self, prompt, model_type="gpt"):
try:
if model_type == "gpt":
response = self.openai_client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
elif model_type == "claude":
message = self.anthropic.messages.create(
model="claude-3-opus-20240229",
messages=[{"role": "user", "content": prompt}]
)
return message.content[0].text
elif model_type == "gemini":
response = self.gemini.generate_content(prompt)
return response.text
except Exception as e:
print(f"Error with {model_type}: {str(e)}")
return None
# 使用示例
ai = AIModelAPI()
prompt = "解释什么是机器学习"
for model in ["gpt", "claude", "gemini"]:
print(f"\n{model.upper()} 回答:")
print(ai.get_completion(prompt, model))
import time
from functools import wraps
def retry_on_error(max_retries=3, delay=1):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
retries = 0
while retries < max_retries:
try:
return func(*args, **kwargs)
except Exception as e:
retries += 1
if retries == max_retries:
raise e
print(f"Error occurred: {str(e)}, retrying... ({retries}/{max_retries})")
time.sleep(delay)
return None
return wrapper
return decorator
# 使用示例
@retry_on_error(max_retries=3)
def safe_chat_with_gpt(prompt):
return ai.get_completion(prompt, "gpt")
• 使用异步调用提高并发性能
• 实现请求队列管理
• 做好错误处理和重试
• 控制请求频率
• 合理设置max_tokens
• 使用模型的较小版本
• 实现token计数
• 缓存常用结果
• API密钥管理
• 内容过滤
• 请求限流
• 日志记录
# Token计数示例
from transformers import GPT2Tokenizer
def count_tokens(text):
tokenizer = GPT2Tokenizer.from_pretrained("gpt2")
return len(tokenizer.encode(text))
# 缓存示例
import functools
@functools.lru_cache(maxsize=1000)
def cached_completion(prompt, model_type):
return ai.get_completion(prompt, model_type)
1. API错误处理
• 网络超时
• 额度限制
• 参数错误
• 服务限制
2. 响应解析
• JSON解析
• 编码处理
• 格式转换
• 内容提取
3. 并发控制
• 请求队列
• 限流策略
• 超时处理
• 资源管理
选择合适的API接入方式,需要考虑以下因素:
• 业务需求特点
• 性能要求
• 成本预算
• 维护难度
建议先用小规模测试验证,然后再逐步扩大使用范围。同时要注意保持代码的可维护性和可扩展性。
文章转自微信公众号@前端道萌