本地大模型分离部署逻辑推理服务器实战（上）

1、前置准备：

Python环境：

Conda python3.10.20

Python包基础依赖：

fastapi==0.109.0 uvicorn==0.27.0 pika==1.3.2 pydantic==2.5.3 pydantic-settings==2.1.0

RabbitMQ 环境：

RabbitMQ 4.2.6 + Erlang 27.3.4.10

大模型环境：

Ollma+LangChain

2、代码准备：

接口层：

发送请求：

@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():
        while True:
            if task_id not in task_results:
                yield f"data: {json.dumps({'error': 'Task not found'})}\n\n"
                break

            data = task_results[task_id]
            status = data["status"]

            yield f"data: {json.dumps({'status': status, 'content': data.get('result', '')})}\n\n"

            if status in ["completed", "failed"]:
                break
            await asyncio.sleep(0.5)  

    return StreamingResponse(event_generator(), media_type="text/event-stream")

清洗服务器：

（数据预处理+数据进入MQ 削峰填谷）

STOP_WORDS_SET = load_stopwords(file_path="./stopwords.txt")

def clean_jieba(raw_text: str) -> str:
    cleaned = " ".join(raw_text.strip().split())
    words = jieba.lcut(cleaned, cut_all=False)
    filtered = [w for w in words if w not in STOP_WORDS_SET and w.strip()]
    return " ".join(filtered)

def clean_text_logic(raw_text: str) -> str:
    return " ".join(raw_text.strip().split())

def get_connection():
    credentials = pika.PlainCredentials(settings.RABBITMQ_USER, settings.RABBITMQ_PASSWORD)
    parameters = pika.ConnectionParameters(host=settings.RABBITMQ_HOST, credentials=credentials)
    return pika.BlockingConnection(parameters)

async def process_and_publish_task(request: CleanRequest) -> str:
    task_id = str(uuid.uuid4())
    cleaned = clean_text_logic(request.raw_text)
    cleaned = clean_jieba(cleaned)
    task_results[task_id] = {"status": "queued", "result": None}

    msg = {"task_id": task_id, "content": cleaned, "ts": time.time()}
    conn = get_connection()
    try:
        ch = conn.channel()
        ch.queue_declare(queue=settings.INFERENCE_QUEUE, durable=True)
        ch.basic_publish('', settings.INFERENCE_QUEUE, body=json.dumps(msg), properties=pika.BasicProperties(delivery_mode=2))
    finally:
        conn.close()
    return task_id

逻辑推理服务器：

logger = logging.getLogger(__name__)

# 初始化本地大模型
llm = Ollama(model="llama3", base_url="http://localhost:11434")


def process_with_langchain(input_text: str) -> str:
    """
    使用 LangChain 进行逻辑推理
    """
    try:
        # 构建提示词，引导模型进行清洗后的深度分析
        prompt = f"请对以下经过清洗的文本进行逻辑总结：\n{input_text}"
        response = llm.invoke(prompt)
        return str(response).strip()
    except Exception as e:
        logger.error(f"LangChain inference error: {e}")
        return "Inference failed due to model error."


def start_inference_consumer():
    """
    后台线程：监听队列并执行 LangChain 推理
    """
    while True:
        connection = None
        try:
            credentials = pika.PlainCredentials(settings.RABBITMQ_USER, settings.RABBITMQ_PASSWORD)
            parameters = pika.ConnectionParameters(
                host=settings.RABBITMQ_HOST,
                credentials=credentials,
                heartbeat=600
            )
            connection = pika.BlockingConnection(parameters)
            channel = connection.channel()
            channel.queue_declare(queue=settings.INFERENCE_QUEUE, durable=True)
            channel.basic_qos(prefetch_count=1)

            def callback(ch, method, properties, body):
                try:
                    data = json.loads(body)
                    task_id = data['task_id']
                    content = data['content']

                    task_results[task_id] = {"status": "processing", "result": "", "chunks": []}

                    # 关键：使用 stream 进行流式推理
                    full_response = ""
                    for chunk in llm.stream(f"请总结：{content}"):
                        full_response += chunk
                        # 将最新片段存入状态，供 SSE 接口读取
                        if task_id in task_results:
                            task_results[task_id]["result"] = full_response
                            task_results[task_id]["last_chunk"] = chunk  # 记录最后一片段用于推送

                    task_results[task_id]["status"] = "completed"
                    ch.basic_ack(delivery_tag=method.delivery_tag)
                except Exception as e:
                    logger.error(e)
                    ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

            channel.basic_consume(
                queue=settings.INFERENCE_QUEUE,
                on_message_callback=callback,
                auto_ack=False
            )

            logger.info("[Consumer] Waiting for messages from RabbitMQ...")
            channel.start_consuming()

        except Exception as e:
            logger.error(f"[Consumer] Connection error: {e}. Reconnecting in 5s...")
            time.sleep(5)
        finally:
            if connection and connection.is_open:
                connection.close()