ollama+Phi-4-mini-reasoning实操手册：批量推理API封装与并发压力测试

1. 引言：从单次对话到批量处理

如果你已经用Ollama部署了Phi-4-mini-reasoning，体验过它出色的推理能力，那么接下来可能会遇到一个现实问题：怎么让这个模型同时处理成百上千个任务？

想象一下这些场景：

• 你需要批量分析大量用户反馈，提取关键观点
• 你的应用后台需要同时响应多个用户的推理请求
• 你想测试模型在高并发下的稳定性和响应速度

这时候，如果还停留在手动在Web界面一个个输入问题，效率就太低了。今天，我就带你从"单兵作战"升级到"集团军作战"，手把手教你如何：

1. 封装一个标准的API接口，让任何程序都能调用Phi-4-mini-reasoning
2. 实现批量推理功能，一次性处理多个任务
3. 进行并发压力测试，看看模型到底能扛住多少请求

无论你是开发者、数据分析师，还是AI应用爱好者，这套方案都能让你把Phi-4-mini-reasoning的能力真正用到生产环境中。

2. 环境准备：快速搭建API服务基础

2.1 确认Ollama服务状态

在开始之前，确保你的Ollama服务正在运行。打开终端，执行：

# 检查Ollama服务状态
ollama serve

# 在另一个终端窗口测试模型是否可用
ollama run phi-4-mini-reasoning "你好，请回复'服务正常'"

如果模型能正常回复，说明基础环境没问题。

2.2 安装必要的Python库

我们需要几个关键的Python库来构建API服务：

pip install fastapi uvicorn httpx pydantic

简单解释一下每个库的作用：

• FastAPI：用来快速构建高性能的Web API
• Uvicorn：ASGI服务器，用来运行FastAPI应用
• Httpx：异步HTTP客户端，用来调用Ollama的接口
• Pydantic：数据验证，确保输入输出的格式正确

2.3 了解Ollama的API接口

Ollama本身提供了REST API，我们可以直接调用。默认情况下，Ollama的API运行在http://localhost:11434。

最重要的两个端点：

• POST /api/generate：用于文本生成
• POST /api/chat：用于对话（Phi-4-mini-reasoning主要用这个）

我们可以先用curl简单测试一下：

curl http://localhost:11434/api/generate -d '{
  "model": "phi-4-mini-reasoning:latest",
  "prompt": "请计算：15 + 27等于多少？",
  "stream": false
}'

如果看到返回了计算结果，说明API调用成功。

3. 基础API封装：让模型变成标准服务

3.1 创建最简单的API服务

我们先从最基础的开始，创建一个能让外部程序调用Phi-4-mini-reasoning的API。

创建一个文件phi4_api.py：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import httpx
import asyncio

app = FastAPI(title="Phi-4-mini-reasoning API服务")

# 定义请求数据模型
class ChatRequest(BaseModel):
    prompt: str  # 用户输入的问题
    max_tokens: int = 512  # 最大生成token数
    temperature: float = 0.7  # 温度参数，控制随机性

# 定义响应数据模型  
class ChatResponse(BaseModel):
    response: str  # 模型回复
    model: str  # 使用的模型名称
    processing_time: float  # 处理时间（秒）

@app.post("/chat", response_model=ChatResponse)
async def chat_with_phi4(request: ChatRequest):
    """
    与Phi-4-mini-reasoning进行单次对话
    """
    start_time = asyncio.get_event_loop().time()
    
    try:
        # 调用Ollama的API
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(
                "http://localhost:11434/api/generate",
                json={
                    "model": "phi-4-mini-reasoning:latest",
                    "prompt": request.prompt,
                    "stream": False,
                    "options": {
                        "num_predict": request.max_tokens,
                        "temperature": request.temperature
                    }
                }
            )
            
            if response.status_code != 200:
                raise HTTPException(
                    status_code=response.status_code,
                    detail=f"Ollama API错误: {response.text}"
                )
            
            result = response.json()
            
            # 计算处理时间
            end_time = asyncio.get_event_loop().time()
            processing_time = end_time - start_time
            
            return ChatResponse(
                response=result.get("response", ""),
                model="phi-4-mini-reasoning:latest",
                processing_time=round(processing_time, 3)
            )
            
    except httpx.TimeoutException:
        raise HTTPException(status_code=504, detail="请求超时")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"内部错误: {str(e)}")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

3.2 启动并测试API服务

运行这个服务：

python phi4_api.py

服务启动后，打开浏览器访问 http://localhost:8000/docs，你会看到自动生成的API文档界面。

现在用Python测试一下：

import requests
import json

# 测试API
test_data = {
    "prompt": "如果一个长方形的长是8厘米，宽是5厘米，它的面积是多少？请一步步推理。",
    "max_tokens": 200,
    "temperature": 0.7
}

response = requests.post(
    "http://localhost:8000/chat",
    json=test_data
)

if response.status_code == 200:
    result = response.json()
    print(f"问题: {test_data['prompt']}")
    print(f"回答: {result['response']}")
    print(f"处理时间: {result['processing_time']}秒")
else:
    print(f"错误: {response.status_code}, {response.text}")

你应该能看到模型返回了完整的推理过程和答案。

4. 批量推理实现：一次处理多个任务

4.1 设计批量处理接口

单次调用虽然简单，但实际应用中我们经常需要批量处理。比如一次分析100条用户评论，或者同时回答多个问题。

我们来创建一个批量处理的版本。新建文件phi4_batch_api.py：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import httpx
import asyncio
from datetime import datetime

app = FastAPI(title="Phi-4-mini-reasoning 批量API服务")

# 批量请求中的单个任务
class BatchTask(BaseModel):
    task_id: str  # 任务ID，用于标识
    prompt: str  # 问题
    max_tokens: int = 256
    temperature: float = 0.7

# 批量请求
class BatchRequest(BaseModel):
    tasks: List[BatchTask]  # 任务列表
    max_concurrent: int = 5  # 最大并发数
    timeout_per_task: int = 30  # 每个任务超时时间（秒）

# 单个任务结果
class TaskResult(BaseModel):
    task_id: str
    success: bool
    response: Optional[str] = None
    error: Optional[str] = None
    processing_time: float

# 批量响应
class BatchResponse(BaseModel):
    batch_id: str  # 批次ID
    start_time: str  # 开始时间
    end_time: str  # 结束时间
    total_tasks: int  # 总任务数
    successful_tasks: int  # 成功任务数
    failed_tasks: int  # 失败任务数
    total_processing_time: float  # 总处理时间
    avg_processing_time: float  # 平均处理时间
    results: List[TaskResult]  # 所有任务结果

@app.post("/batch-chat", response_model=BatchResponse)
async def batch_chat(request: BatchRequest):
    """
    批量处理多个对话任务
    """
    batch_start = datetime.now()
    batch_id = f"batch_{batch_start.strftime('%Y%m%d_%H%M%S')}"
    
    # 限制最大并发数，避免压垮服务
    max_concurrent = min(request.max_concurrent, 10)  # 最多10个并发
    if max_concurrent < 1:
        max_concurrent = 1
    
    # 使用信号量控制并发
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def process_single_task(task: BatchTask) -> TaskResult:
        """处理单个任务"""
        task_start = asyncio.get_event_loop().time()
        
        async with semaphore:
            try:
                async with httpx.AsyncClient(timeout=request.timeout_per_task) as client:
                    response = await client.post(
                        "http://localhost:11434/api/generate",
                        json={
                            "model": "phi-4-mini-reasoning:latest",
                            "prompt": task.prompt,
                            "stream": False,
                            "options": {
                                "num_predict": task.max_tokens,
                                "temperature": task.temperature
                            }
                        }
                    )
                    
                    if response.status_code == 200:
                        result = response.json()
                        task_end = asyncio.get_event_loop().time()
                        
                        return TaskResult(
                            task_id=task.task_id,
                            success=True,
                            response=result.get("response", ""),
                            processing_time=round(task_end - task_start, 3)
                        )
                    else:
                        return TaskResult(
                            task_id=task.task_id,
                            success=False,
                            error=f"API错误: {response.status_code}",
                            processing_time=round(asyncio.get_event_loop().time() - task_start, 3)
                        )
                        
            except httpx.TimeoutException:
                return TaskResult(
                    task_id=task.task_id,
                    success=False,
                    error="请求超时",
                    processing_time=round(asyncio.get_event_loop().time() - task_start, 3)
                )
            except Exception as e:
                return TaskResult(
                    task_id=task.task_id,
                    success=False,
                    error=f"处理错误: {str(e)}",
                    processing_time=round(asyncio.get_event_loop().time() - task_start, 3)
                )
    
    # 并发处理所有任务
    tasks = [process_single_task(task) for task in request.tasks]
    results = await asyncio.gather(*tasks)
    
    # 统计信息
    batch_end = datetime.now()
    successful_tasks = sum(1 for r in results if r.success)
    failed_tasks = len(results) - successful_tasks
    
    # 计算总处理时间（不是简单的相加，而是从开始到结束的时间）
    total_duration = (batch_end - batch_start).total_seconds()
    
    # 计算平均处理时间（只计算成功的任务）
    successful_results = [r for r in results if r.success]
    avg_time = sum(r.processing_time for r in successful_results) / len(successful_results) if successful_results else 0
    
    return BatchResponse(
        batch_id=batch_id,
        start_time=batch_start.isoformat(),
        end_time=batch_end.isoformat(),
        total_tasks=len(results),
        successful_tasks=successful_tasks,
        failed_tasks=failed_tasks,
        total_processing_time=round(total_duration, 3),
        avg_processing_time=round(avg_time, 3),
        results=results
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8001)

4.2 测试批量处理功能

启动批量API服务：

python phi4_batch_api.py

然后创建一个测试脚本test_batch.py：

import requests
import json
import time

# 准备测试数据
test_tasks = []
for i in range(10):
    test_tasks.append({
        "task_id": f"task_{i+1}",
        "prompt": f"问题{i+1}: 计算{10+i}的平方是多少？请给出计算过程。",
        "max_tokens": 150,
        "temperature": 0.7
    })

# 添加一些不同类型的推理问题
math_problems = [
    "一个水池有进水管和出水管，进水管单独注满需要6小时，出水管单独排空需要8小时。如果同时打开进水管和出水管，需要多少小时才能注满水池？",
    "甲乙两人同时从A、B两地相向而行，甲的速度是5km/h，乙的速度是7km/h，两地相距36km。他们相遇需要多少小时？",
    "一个三位数，百位数字是十位数字的2倍，十位数字是个位数字的3倍，这个数可能是多少？"
]

for i, problem in enumerate(math_problems):
    test_tasks.append({
        "task_id": f"math_{i+1}",
        "prompt": problem,
        "max_tokens": 300,
        "temperature": 0.3  # 数学问题温度设低一些，减少随机性
    })

batch_request = {
    "tasks": test_tasks,
    "max_concurrent": 3,  # 同时处理3个任务
    "timeout_per_task": 45  # 每个任务45秒超时
}

print(f"开始批量处理 {len(test_tasks)} 个任务...")
start_time = time.time()

response = requests.post(
    "http://localhost:8001/batch-chat",
    json=batch_request
)

end_time = time.time()

if response.status_code == 200:
    result = response.json()
    
    print(f"\n=== 批量处理结果 ===")
    print(f"批次ID: {result['batch_id']}")
    print(f"总任务数: {result['total_tasks']}")
    print(f"成功任务: {result['successful_tasks']}")
    print(f"失败任务: {result['failed_tasks']}")
    print(f"总耗时: {result['total_processing_time']}秒")
    print(f"平均处理时间: {result['avg_processing_time']}秒")
    
    print(f"\n=== 详细结果（前3个）===")
    for i, task_result in enumerate(result['results'][:3]):
        print(f"\n任务 {task_result['task_id']}:")
        print(f"  状态: {'成功' if task_result['success'] else '失败'}")
        if task_result['success']:
            print(f"  回答: {task_result['response'][:100]}...")  # 只显示前100字符
            print(f"  处理时间: {task_result['processing_time']}秒")
        else:
            print(f"  错误: {task_result['error']}")
    
    # 保存完整结果到文件
    with open(f"batch_result_{result['batch_id']}.json", "w", encoding="utf-8") as f:
        json.dump(result, f, ensure_ascii=False, indent=2)
    print(f"\n完整结果已保存到: batch_result_{result['batch_id']}.json")
    
else:
    print(f"请求失败: {response.status_code}")
    print(response.text)

print(f"\n客户端总耗时: {round(end_time - start_time, 3)}秒")

运行测试脚本，你会看到批量处理的结果统计和部分详细结果。

5. 并发压力测试：看看模型能扛多少

5.1 设计压力测试方案

现在API和批量处理都有了，我们需要知道这个系统能承受多大的压力。压力测试主要关注几个指标：

1. 并发能力：同时能处理多少个请求
2. 响应时间：在不同并发下的平均响应时间
3. 成功率：在高并发下的请求成功率
4. 资源使用：CPU和内存的占用情况

创建一个压力测试脚本stress_test.py：

import asyncio
import httpx
import time
import statistics
from typing import List, Dict
import matplotlib.pyplot as plt
import json
from datetime import datetime

class Phi4StressTester:
    def __init__(self, base_url: str = "http://localhost:11434"):
        self.base_url = base_url
        self.results = []
        
    async def single_request(self, client: httpx.AsyncClient, request_id: int) -> Dict:
        """执行单个请求"""
        start_time = time.time()
        
        try:
            # 使用不同的提示词，模拟真实场景
            prompts = [
                "请解释什么是勾股定理，并给出一个应用例子。",
                "计算：25 × 34 + 18 ÷ 3 等于多少？请一步步计算。",
                "用简单的语言解释什么是人工智能。",
                "如果一个正方形的面积是64平方厘米，它的周长是多少？",
                "写一个简短的关于秋天的诗歌。"
            ]
            
            prompt = prompts[request_id % len(prompts)]
            
            response = await client.post(
                f"{self.base_url}/api/generate",
                json={
                    "model": "phi-4-mini-reasoning:latest",
                    "prompt": prompt,
                    "stream": False,
                    "options": {
                        "num_predict": 100,
                        "temperature": 0.7
                    }
                },
                timeout=30.0
            )
            
            end_time = time.time()
            processing_time = end_time - start_time
            
            if response.status_code == 200:
                return {
                    "request_id": request_id,
                    "success": True,
                    "processing_time": processing_time,
                    "response_length": len(response.json().get("response", "")),
                    "error": None
                }
            else:
                return {
                    "request_id": request_id,
                    "success": False,
                    "processing_time": processing_time,
                    "response_length": 0,
                    "error": f"HTTP {response.status_code}"
                }
                
        except httpx.TimeoutException:
            end_time = time.time()
            return {
                "request_id": request_id,
                "success": False,
                "processing_time": end_time - start_time,
                "response_length": 0,
                "error": "Timeout"
            }
        except Exception as e:
            end_time = time.time()
            return {
                "request_id": request_id,
                "success": False,
                "processing_time": end_time - start_time,
                "response_length": 0,
                "error": str(e)
            }
    
    async def run_test(self, concurrent_requests: int, total_requests: int):
        """运行压力测试"""
        print(f"\n开始压力测试: {concurrent_requests}并发，总共{total_requests}个请求")
        print("=" * 50)
        
        self.results = []
        start_time = time.time()
        
        # 使用信号量控制并发数
        semaphore = asyncio.Semaphore(concurrent_requests)
        
        async def limited_request(client, request_id):
            async with semaphore:
                return await self.single_request(client, request_id)
        
        # 创建客户端并发送请求
        async with httpx.AsyncClient() as client:
            tasks = []
            for i in range(total_requests):
                task = asyncio.create_task(limited_request(client, i))
                tasks.append(task)
            
            # 等待所有任务完成
            results = await asyncio.gather(*tasks)
            self.results = results
        
        end_time = time.time()
        total_duration = end_time - start_time
        
        # 分析结果
        self.analyze_results(total_duration, concurrent_requests)
        
    def analyze_results(self, total_duration: float, concurrent_requests: int):
        """分析测试结果"""
        successful = [r for r in self.results if r["success"]]
        failed = [r for r in self.results if not r["success"]]
        
        success_rate = len(successful) / len(self.results) * 100
        
        if successful:
            processing_times = [r["processing_time"] for r in successful]
            avg_time = statistics.mean(processing_times)
            min_time = min(processing_times)
            max_time = max(processing_times)
            p95 = statistics.quantiles(processing_times, n=20)[18]  # 95百分位
        else:
            avg_time = min_time = max_time = p95 = 0
        
        # 计算QPS（每秒查询数）
        qps = len(successful) / total_duration if total_duration > 0 else 0
        
        print(f"测试完成!")
        print(f"总耗时: {total_duration:.2f}秒")
        print(f"总请求数: {len(self.results)}")
        print(f"成功请求: {len(successful)}")
        print(f"失败请求: {len(failed)}")
        print(f"成功率: {success_rate:.1f}%")
        print(f"QPS: {qps:.2f} (每秒成功请求数)")
        print(f"\n响应时间统计:")
        print(f"  平均: {avg_time:.3f}秒")
        print(f"  最小: {min_time:.3f}秒")
        print(f"  最大: {max_time:.3f}秒")
        print(f"  P95: {p95:.3f}秒 (95%的请求在这个时间内完成)")
        
        # 打印错误详情（如果有）
        if failed:
            print(f"\n错误详情:")
            error_counts = {}
            for f in failed:
                error = f["error"]
                error_counts[error] = error_counts.get(error, 0) + 1
            
            for error, count in error_counts.items():
                print(f"  {error}: {count}次")
        
        # 保存结果
        test_result = {
            "timestamp": datetime.now().isoformat(),
            "concurrent_requests": concurrent_requests,
            "total_requests": len(self.results),
            "successful_requests": len(successful),
            "failed_requests": len(failed),
            "success_rate": success_rate,
            "total_duration": total_duration,
            "qps": qps,
            "avg_response_time": avg_time,
            "min_response_time": min_time,
            "max_response_time": max_time,
            "p95_response_time": p95,
            "errors": error_counts if failed else {}
        }
        
        filename = f"stress_test_result_{concurrent_requests}concurrent.json"
        with open(filename, "w", encoding="utf-8") as f:
            json.dump(test_result, f, ensure_ascii=False, indent=2)
        
        print(f"\n详细结果已保存到: {filename}")
        
        return test_result
    
    def visualize_results(self, results_list: List[Dict]):
        """可视化多个测试结果"""
        if not results_list:
            print("没有可可视化的结果")
            return
            
        concurrencies = [r["concurrent_requests"] for r in results_list]
        qps_values = [r["qps"] for r in results_list]
        avg_times = [r["avg_response_time"] for r in results_list]
        success_rates = [r["success_rate"] for r in results_list]
        
        fig, axes = plt.subplots(2, 2, figsize=(12, 10))
        
        # QPS vs 并发数
        axes[0, 0].plot(concurrencies, qps_values, 'bo-', linewidth=2, markersize=8)
        axes[0, 0].set_xlabel('并发数')
        axes[0, 0].set_ylabel('QPS (请求/秒)')
        axes[0, 0].set_title('并发数 vs QPS')
        axes[0, 0].grid(True, alpha=0.3)
        
        # 响应时间 vs 并发数
        axes[0, 1].plot(concurrencies, avg_times, 'ro-', linewidth=2, markersize=8)
        axes[0, 1].set_xlabel('并发数')
        axes[0, 1].set_ylabel('平均响应时间 (秒)')
        axes[0, 1].set_title('并发数 vs 响应时间')
        axes[0, 1].grid(True, alpha=0.3)
        
        # 成功率 vs 并发数
        axes[1, 0].bar(concurrencies, success_rates, color='green', alpha=0.7)
        axes[1, 0].set_xlabel('并发数')
        axes[1, 0].set_ylabel('成功率 (%)')
        axes[1, 0].set_title('并发数 vs 成功率')
        axes[1, 0].set_ylim([0, 105])
        axes[1, 0].grid(True, alpha=0.3, axis='y')
        
        # 吞吐量分析
        throughput = [qps * 60 for qps in qps_values]  # 每分钟请求数
        axes[1, 1].plot(concurrencies, throughput, 'go-', linewidth=2, markersize=8)
        axes[1, 1].set_xlabel('并发数')
        axes[1, 1].set_ylabel('吞吐量 (请求/分钟)')
        axes[1, 1].set_title('系统吞吐量分析')
        axes[1, 1].grid(True, alpha=0.3)
        
        plt.tight_layout()
        plt.savefig('stress_test_analysis.png', dpi=300, bbox_inches='tight')
        plt.show()
        print("分析图表已保存为: stress_test_analysis.png")

async def main():
    """主测试函数"""
    tester = Phi4StressTester()
    
    # 测试不同的并发级别
    test_scenarios = [
        (1, 10),   # 1并发，10个请求
        (2, 20),   # 2并发，20个请求
        (3, 30),   # 3并发，30个请求
        (5, 50),   # 5并发，50个请求
        (8, 80),   # 8并发，80个请求
        (10, 100), # 10并发，100个请求
    ]
    
    all_results = []
    
    print("Phi-4-mini-reasoning 压力测试开始")
    print("=" * 50)
    
    for concurrency, total in test_scenarios:
        await tester.run_test(concurrency, total)
        all_results.append(tester.results)
        
        # 每次测试后休息一下，让系统恢复
        if concurrency < 10:  # 低并发时休息短一些
            await asyncio.sleep(2)
        else:  # 高并发后休息长一些
            await asyncio.sleep(5)
    
    # 收集汇总结果
    summary_results = []
    for i, (concurrency, total) in enumerate(test_scenarios):
        successful = [r for r in all_results[i] if r["success"]]
        failed = [r for r in all_results[i] if not r["success"]]
        
        if successful:
            processing_times = [r["processing_time"] for r in successful]
            avg_time = statistics.mean(processing_times)
        else:
            avg_time = 0
            
        summary_results.append({
            "concurrent_requests": concurrency,
            "total_requests": total,
            "successful_requests": len(successful),
            "failed_requests": len(failed),
            "success_rate": len(successful) / total * 100,
            "avg_response_time": avg_time,
            "qps": len(successful) / (sum([r["processing_time"] for r in all_results[i]]) / len(all_results[i])) if all_results[i] else 0
        })
    
    # 可视化结果
    tester.visualize_results(summary_results)
    
    # 打印汇总表格
    print("\n" + "=" * 80)
    print("压力测试汇总报告")
    print("=" * 80)
    print(f"{'并发数':<10} {'总请求':<10} {'成功数':<10} {'成功率':<10} {'平均响应时间':<15} {'QPS':<10}")
    print("-" * 80)
    
    for result in summary_results:
        print(f"{result['concurrent_requests']:<10} "
              f"{result['total_requests']:<10} "
              f"{result['successful_requests']:<10} "
              f"{result['success_rate']:<9.1f}% "
              f"{result['avg_response_time']:<14.3f}秒 "
              f"{result['qps']:<9.2f}")

if __name__ == "__main__":
    asyncio.run(main())

5.2 运行压力测试并分析结果

运行压力测试脚本：

# 安装matplotlib用于图表（如果还没安装）
pip install matplotlib

# 运行压力测试
python stress_test.py

测试会依次运行不同并发级别的请求，并生成：

1. 详细的测试报告：每个并发级别的成功率、响应时间等
2. 可视化图表：展示并发数 vs QPS、响应时间、成功率的关系
3. 数据文件：保存每次测试的详细结果

5.3 解读压力测试结果

根据测试结果，你可以了解到：

1. 最佳并发数：在哪个并发级别下QPS最高而响应时间还能接受
2. 系统瓶颈：当并发数增加时，是响应时间先增加还是错误率先增加
3. 稳定性：系统在持续压力下的表现
4. 资源建议：根据测试结果决定需要多少服务器资源

比如，你可能会发现：

• 在3-5并发时，系统性能最佳
• 超过8并发时，响应时间明显增加
• 超过10并发时，开始出现超时错误

这些信息对你部署生产环境非常有帮助。

6. 生产环境优化建议

6.1 性能优化策略

根据压力测试结果，这里有一些优化建议：

1. 连接池管理：

# 使用连接池减少连接建立开销
import httpx
from httpx import Limits

# 创建带连接池的客户端
client = httpx.AsyncClient(
    limits=Limits(max_connections=100, max_keepalive_connections=20),
    timeout=30.0
)

2. 请求批处理优化：

# 对于大量小请求，可以合并发送
async def batch_generate(prompts: List[str], batch_size: int = 10):
    """批量生成，减少API调用次数"""
    results = []
    for i in range(0, len(prompts), batch_size):
        batch = prompts[i:i+batch_size]
        # 这里可以优化为一次发送多个prompt
        # 需要Ollama支持批量生成API
        batch_results = await asyncio.gather(
            *[generate_single(prompt) for prompt in batch]
        )
        results.extend(batch_results)
    return results

3. 缓存策略：

# 对常见问题使用缓存
from functools import lru_cache
import hashlib

@lru_cache(maxsize=1000)
def get_cached_response(prompt: str, temperature: float = 0.7) -> Optional[str]:
    """缓存常见问题的回答"""
    prompt_hash = hashlib.md5(f"{prompt}_{temperature}".encode()).hexdigest()
    # 检查缓存
    cached = cache.get(prompt_hash)
    if cached:
        return cached
    return None

6.2 监控与告警

在生产环境中，监控是必不可少的：

# 简单的监控装饰器
import time
from typing import Callable
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def monitor_performance(func: Callable):
    """监控函数性能的装饰器"""
    async def wrapper(*args, **kwargs):
        start_time = time.time()
        try:
            result = await func(*args, **kwargs)
            end_time = time.time()
            processing_time = end_time - start_time
            
            # 记录性能指标
            logger.info(f"{func.__name__} 执行时间: {processing_time:.3f}秒")
            
            # 如果处理时间过长，发出警告
            if processing_time > 5.0:  # 超过5秒
                logger.warning(f"{func.__name__} 处理时间过长: {processing_time:.3f}秒")
                
            return result
        except Exception as e:
            logger.error(f"{func.__name__} 执行错误: {str(e)}")
            raise
    return wrapper

# 使用装饰器
@monitor_performance
async def chat_with_phi4(prompt: str):
    # ... 原有的聊天函数实现
    pass

6.3 错误处理与重试机制

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),  # 最多重试3次
    wait=wait_exponential(multiplier=1, min=1, max=10),  # 指数退避
    retry_error_callback=lambda retry_state: None  # 重试失败后的回调
)
async def robust_chat_request(prompt: str, max_retries: int = 3):
    """带重试机制的聊天请求"""
    for attempt in range(max_retries):
        try:
            async with httpx.AsyncClient(timeout=30.0) as client:
                response = await client.post(
                    "http://localhost:11434/api/generate",
                    json={
                        "model": "phi-4-mini-reasoning:latest",
                        "prompt": prompt,
                        "stream": False
                    }
                )
                response.raise_for_status()
                return response.json()
        except (httpx.TimeoutException, httpx.NetworkError) as e:
            if attempt == max_retries - 1:
                raise
            wait_time = 2 ** attempt  # 指数退避
            logger.warning(f"请求失败，{wait_time}秒后重试: {str(e)}")
            await asyncio.sleep(wait_time)
        except Exception as e:
            logger.error(f"未知错误: {str(e)}")
            raise

7. 总结：从实验到生产的完整路径

通过本文的实践，我们完成了从单机部署到生产级API服务的完整升级。让我们回顾一下关键收获：

7.1 核心成果总结

1. API服务化：将Phi-4-mini-reasoning封装成了标准的REST API，任何编程语言都能调用
2. 批量处理能力：实现了高效的批量推理，可以同时处理上百个任务
3. 压力测试方案：建立了完整的性能测试体系，知道系统能承受多大压力
4. 生产就绪：提供了监控、错误处理、优化建议等生产环境需要的功能

7.2 实际应用建议

根据你的使用场景，可以选择不同的部署方案：

场景一：个人或小团队使用

• 直接使用基础API服务
• 并发数控制在3-5个
• 适合：个人项目、小规模数据分析

场景二：中等规模应用

• 使用批量API + 连接池优化
• 考虑添加Redis缓存常见回答
• 适合：企业内部工具、中小型网站

场景三：大规模生产环境

• 部署多个Ollama实例，使用负载均衡
• 实现完整的监控告警系统
• 考虑使用消息队列进行任务调度
• 适合：高并发商业应用、SaaS服务

7.3 下一步探索方向

如果你还想进一步深入，可以考虑：

1. 模型微调：针对特定领域的数据微调Phi-4-mini-reasoning
2. 多模型路由：根据问题类型自动选择最合适的模型
3. 流式响应：实现实时的流式输出，提升用户体验
4. 成本优化：监控API使用情况，优化资源分配

7.4 最后的技术提醒

记住几个关键数字（根据你的测试结果调整）：

• 安全并发数：3-5个（避免系统过载）
• 超时设置：30-60秒（给复杂问题足够时间）
• 缓存策略：对常见问题缓存1-24小时
• 监控频率：至少每分钟检查一次服务状态

现在，你已经拥有了一个可以投入生产的Phi-4-mini-reasoning服务。无论是批量处理数据，还是构建AI应用，这套方案都能为你提供稳定可靠的支持。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。