本地大模型分离部署逻辑推理服务器实战(下)

3、异步升级思路：

Python环境加入Redis：

pip install redis

使用AioRedis消除线程阻塞，配合FastAPI 实现高并发，Nginx禁用缓冲及超长等待，Python加入心跳保活，前端防止SSE响应并加入重连

2、代码升级：

异步服务层：

发送请求：

import redis.asyncio as redis
import json
import logging

logger = logging.getLogger(__name__)

class AsyncRedisStreamService:
    def __init__(self, host='localhost', port=6379, db=0):
        self.redis_client = redis.Redis(host=host, port=port, db=db, decode_responses=True)

    async def publish_chunk(self, task_id: str, chunk: str, status: str = "processing"):
        """异步发布推理片段"""
        channel = f"stream:{task_id}"
        message = json.dumps({"task_id": task_id, "chunk": chunk, "status": status})
        await self.redis_client.publish(channel, message)

    async def get_pubsub(self):
        """获取异步 PubSub 对象"""
        return self.redis_client.pubsub()

    async def close(self):
        await self.redis_client.aclose()

利用 aio-pika 和 aioredis 实现全异步链路：

import aio_pika
import json
import asyncio
from langchain_community.llms import Ollama
from app.services.async_redis_service import AsyncRedisStreamService

llm = Ollama(model="llama3")
redis_service = AsyncRedisStreamService()


async def process_task(message: aio_pika.IncomingMessage):
    async with message.process():
        body = json.loads(message.body.decode())
        task_id = body['task_id']

        try:
            # LangChain 异步流式推理
            async for chunk in llm.astream(f"请总结：{body['cleaned_content']}"):
                await redis_service.publish_chunk(task_id, chunk)

            await redis_service.publish_chunk(task_id, "", status="completed")
        except Exception as e:
            await redis_service.publish_chunk(task_id, str(e), status="failed")


async def main():
    connection = await aio_pika.connect("amqp://guest:guest@localhost/")
    channel = await connection.channel()
    queue = await channel.declare_queue("llm_inference_queue", durable=True)
    await queue.consume(process_task)
    print("Async Consumer Started...")
    await asyncio.Future()


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

异步接口层：

import json
from fastapi import APIRouter, BackgroundTasks, HTTPException
from app.schemas.models import CleanRequest, InferenceResult
from app.services.producer import process_and_publish_task
from app.services.state_store import task_results
from fastapi.responses import StreamingResponse
from app.services.async_redis_service import AsyncRedisStreamService

import asyncio
router = APIRouter()

@router.post("/clean-and-submit")
async def submit_task(request: CleanRequest, background_tasks: BackgroundTasks):
    try:
        task_id = await process_and_publish_task(request)
        return {"task_id": task_id, "status": "submitted"}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@router.get("/result/{task_id}", response_model=InferenceResult)
def get_result(task_id: str):
    if task_id not in task_results:
        raise HTTPException(status_code=404, detail="Task not found")
    data = task_results[task_id]
    return InferenceResult(task_id=task_id, status=data["status"], result=data.get("result"))


@router.get("/stream/{task_id}")
async def stream_result(task_id: str):
    async def event_generator():
        service = AsyncRedisStreamService()
        pubsub = await service.get_pubsub()
        await pubsub.subscribe(f"stream:{task_id}")
        last_heartbeat = asyncio.get_event_loop().time()

        try:
            async for message in pubsub.listen():
                if message['type'] == 'message':
                    data = json.loads(message['data'])
                    yield f"data: {json.dumps(data)}\n\n"
                    last_heartbeat = asyncio.get_event_loop().time()

                    if data.get('status') in ['completed', 'failed']:
                        break

                # 心跳保活：若 15 秒无数据，发送注释行防止连接断开
                elif asyncio.get_event_loop().time() - last_heartbeat > 15:
                    yield ": heartbeat\n\n"
                    last_heartbeat = asyncio.get_event_loop().time()

        except asyncio.CancelledError:
            pass
        finally:
            await pubsub.unsubscribe(f"stream:{task_id}")
            await service.close()

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"  # 关键：禁用 Nginx 缓冲
        }
    )

Nginx配置文件优化：

location /api/v1/stream/ {

    proxy_pass http://127.0.0.1:8000;

    
    # 关键配置：禁用代理缓冲

    proxy_buffering off;

    
    # 确保 HTTP 版本支持长连接

    proxy_http_version 1.1;

    proxy_set_header Connection "";
  

    # 增加超时时间，适应长时推理任务

    proxy_read_timeout 3600s;

    proxy_send_timeout 3600s;

    
    # 禁用 gzip 压缩，避免分块传输编码冲突

    gzip off;

}

前端优化：

在HTTP Header加入

Cache-Control: no-cache

前端处理SSE时，加入自动重连：

const eventSource = new EventSource('/api/v1/stream/123');

// 设置重连时间为 3 秒

eventSource.onopen = () => {

    console.log('Connection opened');

};

eventSource.onmessage = (event) => {

    const data = JSON.parse(event.data);

    updateUI(data);

};

eventSource.onerror = (err) => {

    console.error('Connection lost, reconnecting...', err);

};

后端指定重连时间：

yield "retry: 3000\n\n"