注意： temperature和top_p不要同时调太高。我见过有人两个都设到0.95，结果模型开始胡言乱语。一般固定一个，调另一个就行。
4.3 错误处理与重试机制：别让用户看到500
API调用总会出错。网络闪断、服务重启、模型加载超时……我曾经在一个生产环境里，因为没做重试，半夜被报警电话吵醒三次。
常见的错误码：
- 400：请求参数有问题，检查JSON格式
- 404：模型不存在，先
ollama pull一下 - 500：服务内部错误，通常是模型推理异常
- 503：服务过载，等会儿再试
我写了一个带重试的客户端，分享给你：
import time
import requests
import json
from typing import Optional
class OllamaClient:
def __init__(self, base_url="http://localhost:11434", max_retries=3):
self.base_url = base_url
self.max_retries = max_retries
def generate(self, prompt: str, model: str = "qwen2:7b",
stream: bool = False, **kwargs) -> Optional[str]:
url = f"{self.base_url}/api/generate"
payload = {
"model": model,
"prompt": prompt,
"stream": stream,
**kwargs
}
for attempt in range(self.max_retries):
try:
response = requests.post(url, json=payload, timeout=30)
response.raise_for_status()
if stream:
return self._handle_stream(response)
else:
return response.json().get('response', '')
except requests.exceptions.Timeout:
print(f"超时了，第{attempt+1}次重试...")
time.sleep(2 ** attempt) # 指数退避
except requests.exceptions.HTTPError as e:
if response.status_code == 503:
print("服务过载，等待后重试...")
time.sleep(5)
else:
print(f"HTTP错误: {e}")
break # 非临时性错误，不重试
except Exception as e:
print(f"未知错误: {e}")
time.sleep(1)
return None
def _handle_stream(self, response):
full_response = ""
for line in response.iter_lines():
if line:
data = json.loads(line.decode('utf-8'))
if 'response' in data:
full_response += data['response']
if data.get('done', False):
break
return full_response
核心原则： 重试要带指数退避（exponential backoff）。第一次等2秒，第二次等4秒，第三次等8秒。别一秒钟重试10次，那叫DDoS攻击自己。
4.4 API性能调优：榨干每一分算力
最后聊聊性能。RAG系统里，API调用是瓶颈。我总结了几条实战经验：
4.4.1 连接池复用
每次请求都新建TCP连接，太浪费了。用requests.Session复用连接：
session = requests.Session()
adapter = requests.adapters.HTTPAdapter(
pool_connections=10,
pool_maxsize=20
)
session.mount('http://', adapter)
session.mount('https://', adapter)4.4.2 批量请求与并发
如果你有多个独立的问题要问模型，别串行处理。用concurrent.futures并发请求：
from concurrent.futures import ThreadPoolExecutor, as_completed
def batch_generate(prompts, model="qwen2:7b", max_workers=4):
client = OllamaClient()
results = {}
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(client.generate, p, model): p
for p in prompts
}
for future in as_completed(futures):
prompt = futures[future]
try:
results[prompt] = future.result()
except Exception as e:
print(f"处理失败: {prompt[:20]}... 错误: {e}")
results[prompt] = None
return results
注意并发数： 别开太多线程。我试过开20个并发，Ollama直接OOM了。一般4-8个线程比较安全，具体看你的GPU显存。
4.4.3 模型预热
模型第一次加载很慢，因为要把参数加载到显存。我习惯在服务启动时先发一个空请求：
# 预热模型
def warmup_model(model="qwen2:7b"):
client = OllamaClient()
client.generate("你好", model=model)
print(f"{model} 预热完成")4.4.4 缓存策略
对于重复的问题，别每次都调模型。用LRU缓存：
from functools import lru_cache
@lru_cache(maxsize=100)
def cached_generate(prompt: str, model: str = "qwen2:7b"):
client = OllamaClient()
return client.generate(prompt, model)
我的经验： 缓存命中率能做到30%-50%。特别是RAG系统里，用户经常问相似的问题。但注意，缓存要设置过期时间，不然模型更新了用户还在用旧答案。
4.5 本章知识体系
我把这一章的核心逻辑画了张图，方便你理解各个模块之间的关系：
嗯，这一章内容不少。流式输出解决用户体验，自定义参数控制模型行为，错误处理保证系统稳定，性能调优提升吞吐量。四者缺一不可。
我在实际项目中，把这套东西封装成了一个微服务，每天处理几十万次API调用，基本没出过问题。你照着这个思路做，也能搭出一个健壮的RAG系统。