LLM 正在拆掉云原生架构的地基

任意请求 → 任意服务器 → 唯一数据库

特定请求 → 必须落到特定服务器 → 上下文状态在显存里

# session_router.py — 最小可运行的会话亲和路由示例
# 依赖: pip install fastapi uvicorn redis httpx
# 运行: uvicorn session_router:app --port 8000
# 前置: 本地起一个 Redis (docker run -d -p 6379:6379 redis)
#        以及两个模拟推理节点 (见下方 inference_node.py)

import hashlib
import httpx
from fastapi import FastAPI, Request
from redis import Redis

app = FastAPI()
redis = Redis(host="localhost", port=6379, decode_responses=True)

# 模拟两个推理节点
INFERENCE_NODES = [
    "http://localhost:8001",  # 推理节点 A
    "http://localhost:8002",  # 推理节点 B
]


def pick_node(session_id: str) -> str:
    """根据 session_id 做确定性路由，同一会话始终落到同一节点"""
    # 先查缓存：之前绑定了哪个节点？
    cached = redis.get(f"session:{session_id}:node")
    if cached:
        return cached

    # 没有缓存，用一致性哈希分配
    idx = int(hashlib.md5(session_id.encode()).hexdigest(), 16) % len(INFERENCE_NODES)
    node = INFERENCE_NODES[idx]
    # 写入缓存，TTL 30 分钟（会话超时后释放绑定）
    redis.setex(f"session:{session_id}:node", 1800, node)
    return node


@app.post("/chat/{session_id}")
async def chat(session_id: str, request: Request):
    body = await request.json()
    node_url = pick_node(session_id)

    try:
        resp = await httpx.AsyncClient(timeout=30.0).post(
            f"{node_url}/infer", json={"session_id": session_id, **body}
        )
        return resp.json()
    except httpx.ConnectError:
        # 节点挂了：清除绑定，重新分配
        redis.delete(f"session:{session_id}:node")
        new_node = pick_node(session_id)
        resp = await httpx.AsyncClient(timeout=30.0).post(
            f"{new_node}/infer", json={"session_id": session_id, **body}
        )
        # 新节点没有 KV Cache，返回中标记需要重新 prefill
        result = resp.json()
        result["prefill_required"] = True
        return result

# inference_node.py — 模拟推理节点，启动两个实例
# 运行: uvicorn inference_node:app --port 8001
#       uvicorn inference_node:app --port 8002

from fastapi import FastAPI, Request
from collections import defaultdict

app = FastAPI()

# 模拟每个节点本地维护的 KV Cache 状态
# key=session_id, value=已缓存的 token 数
kv_cache: dict[str, int] = defaultdict(int)


@app.post("/infer")
async def infer(request: Request):
    body = await request.json()
    session_id = body["session_id"]
    new_tokens = len(body.get("prompt", "").split())

    cached_tokens = kv_cache[session_id]
    kv_cache[session_id] += new_tokens

    # 如果有缓存，只处理新增 token（模拟增量推理）
    # 如果没有缓存，需要全量 prefill
    prefill = cached_tokens == 0

    return {
        "response": f"[节点 {app.port}] 回复: ...",
        "cached_tokens": cached_tokens,
        "new_tokens": new_tokens,
        "prefill": prefill,  # True = 慢，False = 快（复用了 KV Cache）
    }

# hack: 让响应里能区分节点
@app.middleware("http")
async def add_port(request: Request, call_next):
    import starlette.responses as r
    response = await call_next(request)
    app.port = request.url.port  # 不严谨，仅用于演示区分节点
    return response