外部客户端（浏览器/SDK）              内部服务（Control Layer + Data Layer）
      │                                     │
      │  REST/JSON（易于调试、通用）           │  gRPC/Protobuf（高性能、类型安全）
      │                                     │
      ▼                                     ▼
┌─────────────────────────────────────────────┐
│           Intelligence Layer (Python)        │
│                                              │
│  ┌──────────────┐    ┌──────────────────┐   │
│  │  FastAPI      │    │  gRPC Server     │   │
│  │  Port 8000    │    │  Port 50051      │   │
│  │  REST/JSON    │    │  Protobuf        │   │
│  └──────┬───────┘    └──────┬───────────┘   │
│         │                    │               │
│         └────────┬───────────┘               │
│                  │                           │
│         ┌────────┴────────┐                  │
│         │  Service Layer  │                  │
│         │  (共享业务逻辑)  │                  │
│         └─────────────────┘                  │
└──────────────────────────────────────────────┘

# coomia-dip Intelligence Layer 入口
from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时初始化
    await init_grpc_clients()
    await init_reasoning_engine()
    yield
    # 关闭时清理
    await shutdown_grpc_clients()

app = FastAPI(
    title="coomia-dip Intelligence Layer",
    version="1.0.0",
    lifespan=lifespan,
)

from fastapi import APIRouter, Depends, Query, Path
from pydantic import BaseModel, Field

router = APIRouter(prefix="/api/v1/reasoning", tags=["reasoning"])

class ReasoningRequest(BaseModel):
    """推理请求"""
    world_id: str = Field(..., description="World ID")
    query: str = Field(..., description="推理查询")
    context: dict[str, Any] = Field(default_factory=dict)
    max_depth: int = Field(default=5, ge=1, le=20)

class ReasoningResponse(BaseModel):
    """推理响应"""
    result: dict[str, Any]
    confidence: float = Field(ge=0.0, le=1.0)
    reasoning_chain: list[str]
    execution_time_ms: float

@router.post("/execute", response_model=ReasoningResponse)
async def execute_reasoning(
    request: ReasoningRequest,
    world_context: WorldContext = Depends(get_world_context),
) -> ReasoningResponse:
    result = await reasoning_service.execute(
        world_id=request.world_id,
        query=request.query,
        context=request.context,
        max_depth=request.max_depth,
    )
    return ReasoningResponse(
        result=result.data,
        confidence=result.confidence,
        reasoning_chain=result.chain,
        execution_time_ms=result.duration_ms,
    )

from fastapi import Depends, Header, HTTPException

async def get_world_context(
    x_world_id: str = Header(..., alias="X-World-Id"),
    x_user_id: str | None = Header(None, alias="X-User-Id"),
) -> WorldContext:
    """从请求头提取 World Context"""
    if not x_world_id:
        raise HTTPException(status_code=400, detail="X-World-Id header required")
    return WorldContext(world_id=x_world_id, user_id=x_user_id)

async def get_reasoning_service(
    world_ctx: WorldContext = Depends(get_world_context),
) -> ReasoningService:
    """获取绑定了 World Context 的推理服务"""
    return ReasoningService(world_context=world_ctx)

// reasoning_service.proto
syntax = "proto3";
package onto.intelligence.v1;

service ReasoningService {
    rpc ExecuteReasoning (ReasoningRequest) returns (ReasoningResponse);
    rpc StreamReasoning (ReasoningRequest) returns (stream ReasoningStep);
    rpc BatchReasoning (stream ReasoningRequest) returns (stream ReasoningResponse);
}

message ReasoningRequest {
    string world_id = 1;
    string query = 2;
    map<string, string> context = 3;
    int32 max_depth = 4;
}

message ReasoningResponse {
    map<string, string> result = 1;
    double confidence = 2;
    repeated string reasoning_chain = 3;
    double execution_time_ms = 4;
}

message ReasoningStep {
    int32 step_number = 1;
    string description = 2;
    string status = 3;
    map<string, string> intermediate_result = 4;
}

import grpc
from concurrent import futures
from onto.intelligence.v1 import reasoning_service_pb2_grpc as reasoning_grpc
from onto.intelligence.v1 import reasoning_service_pb2 as reasoning_pb

class ReasoningGrpcServicer(reasoning_grpc.ReasoningServiceServicer):
    """gRPC 推理服务实现"""

    def __init__(self, reasoning_service: ReasoningService):
        self.service = reasoning_service

    async def ExecuteReasoning(
        self,
        request: reasoning_pb.ReasoningRequest,
        context: grpc.aio.ServicerContext,
    ) -> reasoning_pb.ReasoningResponse:
        # 从 gRPC metadata 获取 World Context
        metadata = dict(context.invocation_metadata())
        world_id = metadata.get("x-world-id", request.world_id)

        try:
            result = await self.service.execute(
                world_id=world_id,
                query=request.query,
                context=dict(request.context),
                max_depth=request.max_depth,
            )
            return reasoning_pb.ReasoningResponse(
                result=result.data,
                confidence=result.confidence,
                reasoning_chain=result.chain,
                execution_time_ms=result.duration_ms,
            )
        except ValidationError as e:
            await context.abort(grpc.StatusCode.INVALID_ARGUMENT, str(e))
        except PermissionError as e:
            await context.abort(grpc.StatusCode.PERMISSION_DENIED, str(e))

    async def StreamReasoning(
        self,
        request: reasoning_pb.ReasoningRequest,
        context: grpc.aio.ServicerContext,
    ):
        """服务端流式推理 — 逐步返回推理过程"""
        async for step in self.service.stream_execute(
            world_id=request.world_id,
            query=request.query,
        ):
            yield reasoning_pb.ReasoningStep(
                step_number=step.number,
                description=step.description,
                status=step.status,
                intermediate_result=step.data,
            )

import asyncio
import uvicorn

async def start_grpc_server() -> grpc.aio.Server:
    """启动 gRPC Server"""
    server = grpc.aio.server(
        futures.ThreadPoolExecutor(max_workers=10),
        options=[
            ("grpc.max_send_message_length", 50 * 1024 * 1024),
            ("grpc.max_receive_message_length", 50 * 1024 * 1024),
            ("grpc.keepalive_time_ms", 30000),
            ("grpc.keepalive_timeout_ms", 5000),
        ],
    )
    reasoning_grpc.add_ReasoningServiceServicer_to_server(
        ReasoningGrpcServicer(reasoning_service), server
    )
    server.add_insecure_port("[::]:50051")
    await server.start()
    return server

async def main():
    # 并行启动 FastAPI 和 gRPC
    grpc_server = await start_grpc_server()

    config = uvicorn.Config(
        app="main:app",
        host="0.0.0.0",
        port=8000,
        workers=1,
        loop="asyncio",
    )
    uvicorn_server = uvicorn.Server(config)

    await asyncio.gather(
        uvicorn_server.serve(),
        grpc_server.wait_for_termination(),
    )

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

from pydantic import BaseModel
from google.protobuf.message import Message
from typing import TypeVar, Type

T = TypeVar("T", bound=BaseModel)
P = TypeVar("P", bound=Message)

class ProtoConverter:
    """Pydantic v2 与 Protobuf 之间的自动转换"""

    @staticmethod
    def to_pydantic(proto_msg: P, pydantic_class: Type[T]) -> T:
        """Protobuf Message → Pydantic Model"""
        from google.protobuf.json_format import MessageToDict
        data = MessageToDict(proto_msg, preserving_proto_field_name=True)
        return pydantic_class.model_validate(data)

    @staticmethod
    def to_proto(pydantic_obj: T, proto_class: Type[P]) -> P:
        """Pydantic Model → Protobuf Message"""
        from google.protobuf.json_format import ParseDict
        data = pydantic_obj.model_dump(mode="json")
        return ParseDict(data, proto_class())

# 使用示例
pydantic_req = ProtoConverter.to_pydantic(grpc_request, ReasoningRequest)
proto_resp = ProtoConverter.to_proto(pydantic_response, reasoning_pb.ReasoningResponse)

from pydantic import BaseModel, Field, ConfigDict

class OntologyInstanceModel(BaseModel):
    """同时服务 REST 和 gRPC 的共享模型"""
    model_config = ConfigDict(from_attributes=True)

    id: str
    object_type: str = Field(alias="objectType")
    object_id: str = Field(alias="objectId")
    world_id: str = Field(alias="worldId")
    properties: dict[str, Any] = Field(default_factory=dict)
    version: int = 0
    created_at: datetime | None = None
    updated_at: datetime | None = None

from enum import IntEnum
import grpc
from fastapi import HTTPException

class ErrorCode(IntEnum):
    VALIDATION_ERROR = 1001
    NOT_FOUND = 1002
    PERMISSION_DENIED = 1003
    CONFLICT = 1004
    INTERNAL_ERROR = 1005
    TIMEOUT = 1006

# REST HTTP 状态码映射
ERROR_TO_HTTP = {
    ErrorCode.VALIDATION_ERROR: 400,
    ErrorCode.NOT_FOUND: 404,
    ErrorCode.PERMISSION_DENIED: 403,
    ErrorCode.CONFLICT: 409,
    ErrorCode.INTERNAL_ERROR: 500,
    ErrorCode.TIMEOUT: 504,
}

# gRPC 状态码映射
ERROR_TO_GRPC = {
    ErrorCode.VALIDATION_ERROR: grpc.StatusCode.INVALID_ARGUMENT,
    ErrorCode.NOT_FOUND: grpc.StatusCode.NOT_FOUND,
    ErrorCode.PERMISSION_DENIED: grpc.StatusCode.PERMISSION_DENIED,
    ErrorCode.CONFLICT: grpc.StatusCode.ALREADY_EXISTS,
    ErrorCode.INTERNAL_ERROR: grpc.StatusCode.INTERNAL,
    ErrorCode.TIMEOUT: grpc.StatusCode.DEADLINE_EXCEEDED,
}

class CoomiaDipError(Exception):
    """coomia-dip 统一异常基类"""
    def __init__(self, code: ErrorCode, message: str, details: dict | None = None):
        self.code = code
        self.message = message
        self.details = details or {}
        super().__init__(message)

    def to_http_exception(self) -> HTTPException:
        return HTTPException(
            status_code=ERROR_TO_HTTP[self.code],
            detail={"code": self.code, "message": self.message, "details": self.details},
        )

    async def abort_grpc(self, context: grpc.aio.ServicerContext) -> None:
        await context.abort(
            ERROR_TO_GRPC[self.code],
            f"[{self.code}] {self.message}",
        )

from starlette.middleware.base import BaseHTTPMiddleware

class RequestTracingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        request_id = request.headers.get("X-Request-Id", str(uuid4()))
        start_time = time.monotonic()

        # 注入 request context
        token = request_context.set({"request_id": request_id})

        try:
            response = await call_next(request)
            duration = time.monotonic() - start_time

            response.headers["X-Request-Id"] = request_id
            response.headers["X-Response-Time"] = f"{duration:.3f}s"

            logger.info(
                "request completed",
                request_id=request_id,
                method=request.method,
                path=request.url.path,
                status=response.status_code,
                duration_ms=duration * 1000,
            )
            return response
        finally:
            request_context.reset(token)

app.add_middleware(RequestTracingMiddleware)

class TracingInterceptor(grpc.aio.ServerInterceptor):
    async def intercept_service(self, continuation, handler_call_details):
        request_id = None
        for key, value in handler_call_details.invocation_metadata:
            if key == "x-request-id":
                request_id = value
                break

        if not request_id:
            request_id = str(uuid4())

        start_time = time.monotonic()
        method = handler_call_details.method

        try:
            handler = await continuation(handler_call_details)
            duration = time.monotonic() - start_time
            logger.info(
                "gRPC call completed",
                request_id=request_id,
                method=method,
                duration_ms=duration * 1000,
            )
            return handler
        except Exception as e:
            duration = time.monotonic() - start_time
            logger.error(
                "gRPC call failed",
                request_id=request_id,
                method=method,
                error=str(e),
                duration_ms=duration * 1000,
            )
            raise

# 注册拦截器
server = grpc.aio.server(interceptors=[TracingInterceptor()])

# FastAPI 健康检查
@app.get("/health")
async def health_check():
    checks = {
        "grpc_server": await check_grpc_server(),
        "database": await check_database(),
        "redis": await check_redis(),
    }
    all_healthy = all(checks.values())
    return JSONResponse(
        status_code=200 if all_healthy else 503,
        content={"status": "healthy" if all_healthy else "unhealthy", "checks": checks},
    )

# gRPC 健康检查（标准协议）
from grpc_health.v1 import health_pb2_grpc, health_pb2

class HealthServicer(health_pb2_grpc.HealthServicer):
    async def Check(self, request, context):
        service = request.service
        if service == "" or service == "onto.intelligence.v1.ReasoningService":
            return health_pb2.HealthCheckResponse(
                status=health_pb2.HealthCheckResponse.SERVING
            )
        return health_pb2.HealthCheckResponse(
            status=health_pb2.HealthCheckResponse.NOT_SERVING
        )

import asyncio
from concurrent.futures import ThreadPoolExecutor

# CPU 密集型任务使用线程池
executor = ThreadPoolExecutor(max_workers=4)

async def cpu_intensive_reasoning(query: str) -> dict:
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        executor,
        _sync_reasoning,  # 同步的 CPU 密集型函数
        query,
    )
    return result

def _sync_reasoning(query: str) -> dict:
    # OR-Tools 求解等 CPU 密集型操作
    return solver.solve(query)

测试环境：4 核 8GB
工具：wrk2, ghz

REST (FastAPI):
  简单查询: 12,000 RPS, P99 = 5ms
  推理请求: 2,000 RPS, P99 = 45ms

gRPC:
  简单查询: 18,000 RPS, P99 = 2ms
  推理请求: 3,500 RPS, P99 = 30ms
  流式传输: 50,000 msg/s

gRPC 比 REST 快 50-75%（序列化 + 传输）

FROM python:3.12-slim

WORKDIR /app
COPY pyproject.toml .
RUN pip install --no-cache-dir .

COPY . .
EXPOSE 8000 50051

CMD ["python", "-m", "coomia-dip.intelligence.main"]

apiVersion: apps/v1
kind: Deployment
metadata:
  name: intelligence-Layer
spec:
  replicas: 3
  template:
    spec:
      containers:
        - name: intelligence
          image: coomia-dip/intelligence-Layer:latest
          ports:
            - containerPort: 8000
              name: http
            - containerPort: 50051
              name: grpc
          resources:
            requests:
              cpu: "2"
              memory: "4Gi"
            limits:
              cpu: "4"
              memory: "8Gi"
          livenessProbe:
            httpGet:
              path: /health
              port: 8000
            initialDelaySeconds: 10
          readinessProbe:
            grpc:
              port: 50051
            initialDelaySeconds: 5

from fastapi.testclient import TestClient

def test_execute_reasoning():
    client = TestClient(app)
    response = client.post(
        "/api/v1/reasoning/execute",
        json={"world_id": "test", "query": "optimize route", "max_depth": 3},
        headers={"X-World-Id": "test"},
    )
    assert response.status_code == 200
    data = response.json()
    assert "result" in data
    assert 0 <= data["confidence"] <= 1

import grpc
import pytest

@pytest.fixture
async def grpc_channel():
    async with grpc.aio.insecure_channel("localhost:50051") as channel:
        yield channel

@pytest.mark.asyncio
async def test_grpc_reasoning(grpc_channel):
    stub = reasoning_grpc.ReasoningServiceStub(grpc_channel)
    response = await stub.ExecuteReasoning(
        reasoning_pb.ReasoningRequest(
            world_id="test",
            query="optimize route",
            max_depth=3,
        ),
        metadata=[("x-world-id", "test")],
    )
    assert response.confidence > 0
    assert len(response.reasoning_chain) > 0