feat(ai): 完善AI网关服务并添加LLM降级模式

config.py 加openai_api_key/base_url/dev_mode

新建llm_client.py httpx异步调OpenAI REST API

main.py 业务路由加/ai前缀+降级模式+readyz端点

Gateway添加/notifications和/ai路由

docs: known-issues记录P5三服务经验

services/ai/src/ai/config.py

@@ -7,12 +7,27 @@ class Settings(BaseSettings):
     """应用配置."""
 
     port: int = 3008
+    # LLM 配置（可选，为空时降级返回骨架响应）
     openai_api_key: str = ""
+    openai_base_url: str = "https://api.openai.com/v1"
     anthropic_api_key: str = ""
+    # 开发模式：true 时跳过 OTel exporter 初始化，避免本地无 collector 时报错
+    dev_mode: str = "false"
+    # 可观测性
     otel_endpoint: str = "http://localhost:4318"
     log_level: str = "info"
 
     model_config = {"env_file": ".env", "env_prefix": ""}
 
+    @property
+    def is_dev(self) -> bool:
+        """是否处于开发模式."""
+        return self.dev_mode.lower() == "true"
+
+    @property
+    def llm_available(self) -> bool:
+        """LLM 是否可用（至少一个 provider 配置了 API key）."""
+        return bool(self.openai_api_key or self.anthropic_api_key)
+
 
 settings = Settings()

services/ai/src/ai/llm_client.py

@@ -0,0 +1,137 @@
+"""LLM 客户端 - 使用 httpx 直接调用 OpenAI 兼容 REST API。
+
+设计要点：
+- 不依赖 openai SDK，纯 httpx 异步调用
+- api_key 为空或调用失败时返回 None / yield 降级骨架数据
+- 调用方据此决定是否进入降级路径
+"""
+
+from collections.abc import AsyncGenerator
+from typing import Any
+
+import httpx
+import structlog
+
+logger = structlog.get_logger()
+
+# 非流式请求默认超时（秒）
+DEFAULT_TIMEOUT: float = 30.0
+# 流式请求建立连接超时（秒）；读取通过迭代器控制
+STREAM_CONNECT_TIMEOUT: float = 30.0
+# 流式读取单次 chunk 超时（秒）
+STREAM_READ_TIMEOUT: float = 60.0
+
+
+def _build_url(base_url: str) -> str:
+    """拼接 chat completions 端点 URL."""
+    return f"{base_url.rstrip('/')}/chat/completions"
+
+
+def _build_headers(api_key: str) -> dict[str, str]:
+    """构建请求头."""
+    return {
+        "Authorization": f"Bearer {api_key}",
+        "Content-Type": "application/json",
+    }
+
+
+async def chat_completion(
+    messages: list[dict[str, Any]],
+    model: str,
+    temperature: float,
+    api_key: str,
+    base_url: str,
+) -> dict[str, Any] | None:
+    """非流式调用 LLM。
+
+    Returns:
+        OpenAI 兼容的响应 dict；api_key 为空或调用失败时返回 None（由调用方降级）。
+    """
+    if not api_key:
+        logger.warning("llm_chat_completion_no_api_key_degraded")
+        return None
+
+    url = _build_url(base_url)
+    headers = _build_headers(api_key)
+    payload = {
+        "model": model,
+        "messages": messages,
+        "temperature": temperature,
+        "stream": False,
+    }
+
+    try:
+        async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as client:
+            resp = await client.post(url, json=payload, headers=headers)
+            resp.raise_for_status()
+            return resp.json()
+    except httpx.HTTPStatusError as exc:
+        logger.error(
+            "llm_chat_completion_http_error",
+            status_code=exc.response.status_code,
+            body=exc.response.text[:500],
+        )
+        return None
+    except Exception as exc:  # noqa: BLE001 - 顶层兜底，所有异常均降级
+        logger.error("llm_chat_completion_failed", error=str(exc))
+        return None
+
+
+async def chat_completion_stream(
+    messages: list[dict[str, Any]],
+    model: str,
+    temperature: float,
+    api_key: str,
+    base_url: str,
+) -> AsyncGenerator[str, None]:
+    """流式调用 LLM，以 SSE 格式（``data: <chunk>\\n\\n``）yield。
+
+    api_key 为空或调用失败时 yield 降级骨架数据，保证下游始终能消费。
+    """
+    if not api_key:
+        logger.warning("llm_stream_no_api_key_degraded")
+        yield (
+            'data: {"choices":[{"delta":{"content":"[degraded] LLM API key not configured"}}]}\n\n'
+        )
+        yield "data: [DONE]\n\n"
+        return
+
+    url = _build_url(base_url)
+    headers = _build_headers(api_key)
+    payload = {
+        "model": model,
+        "messages": messages,
+        "temperature": temperature,
+        "stream": True,
+    }
+
+    timeout = httpx.Timeout(
+        connect=STREAM_CONNECT_TIMEOUT,
+        read=STREAM_READ_TIMEOUT,
+        write=STREAM_CONNECT_TIMEOUT,
+        pool=STREAM_CONNECT_TIMEOUT,
+    )
+
+    try:
+        async with (
+            httpx.AsyncClient(timeout=timeout) as client,
+            client.stream("POST", url, json=payload, headers=headers) as resp,
+        ):
+            resp.raise_for_status()
+            async for line in resp.aiter_lines():
+                if not line or not line.startswith("data: "):
+                    continue
+                yield f"{line}\n\n"
+                if line.strip() == "data: [DONE]":
+                    return
+    except httpx.HTTPStatusError as exc:
+        logger.error(
+            "llm_stream_http_error_degraded",
+            status_code=exc.response.status_code,
+        )
+        yield 'data: {"choices":[{"delta":{"content":"[degraded] LLM stream HTTP error"}}]}\n\n'
+        yield "data: [DONE]\n\n"
+    except Exception as exc:  # noqa: BLE001 - 顶层兜底，所有异常均降级
+        logger.error("llm_stream_failed_degraded", error=str(exc))
+        yield 'data: {"choices":[{"delta":{"content":"[degraded] LLM stream error"}}]}\n\n'
+        yield "data: [DONE]\n\n"

services/ai/src/ai/main.py

@@ -1,9 +1,11 @@
 """AI 网关服务入口."""
 
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
+from typing import Any
 
 import structlog
-from fastapi import FastAPI
+from fastapi import APIRouter, FastAPI
 from fastapi.responses import StreamingResponse
 from opentelemetry import trace
 from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
@@ -12,25 +14,45 @@ from opentelemetry.sdk.trace.export import BatchSpanProcessor
 from prometheus_client import make_asgi_app
 from pydantic import BaseModel
 
+from .config import settings
+from .llm_client import chat_completion, chat_completion_stream
+
 logger = structlog.get_logger()
 tracer = trace.get_tracer(__name__)
 
 
 def init_tracer() -> None:
-    """初始化 OpenTelemetry."""
+    """初始化 OpenTelemetry.
+
+    endpoint 从 settings.otel_endpoint 读取；dev_mode=true 时跳过 exporter
+    初始化，避免本地无 collector 时报错。
+    """
+    if settings.is_dev:
+        logger.info("dev_mode_tracer_skipped", dev_mode=settings.dev_mode)
+        return
+
     provider = TracerProvider()
-    exporter = OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces")
+    endpoint = f"{settings.otel_endpoint.rstrip('/')}/v1/traces"
+    exporter = OTLPSpanExporter(endpoint=endpoint)
     provider.add_span_processor(BatchSpanProcessor(exporter))
     trace.set_tracer_provider(provider)
+    logger.info("tracer_initialized", otel_endpoint=endpoint)
 
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     """应用生命周期."""
     init_tracer()
-    logger.info("ai service starting")
+    logger.info(
+        "ai_service_starting",
+        llm_available=settings.llm_available,
+        dev_mode=settings.is_dev,
+        openai_base_url=settings.openai_base_url,
+    )
+    if not settings.llm_available:
+        logger.warning("ai_service_llm_degraded_no_api_key")
     yield
-    logger.info("ai service stopping")
+    logger.info("ai_service_stopping")
 
 
 app = FastAPI(
@@ -41,11 +63,14 @@ app = FastAPI(
 
 app.mount("/metrics", make_asgi_app())
 
+# 业务路由加 /ai 前缀，Gateway 代理 /api/v1/ai/* → /ai/*
+router = APIRouter(prefix="/ai")
+
 
 class ChatRequest(BaseModel):
     """聊天请求."""
 
-    messages: list[dict]
+    messages: list[dict[str, Any]]
     model: str = "gpt-4o-mini"
     temperature: float = 0.7
     stream: bool = False
@@ -56,56 +81,157 @@ class ChatResponse(BaseModel):
 
     content: str
     model: str
-    usage: dict
+    usage: dict[str, Any]
+    degraded: bool = False
+
+
+def _extract_content(result: dict[str, Any] | None) -> tuple[str, str, dict[str, Any]]:
+    """从 OpenAI 响应中抽取 (content, model, usage)。"""
+    if result is None:
+        return "", "", {}
+    choices = result.get("choices", [])
+    content = ""
+    if choices:
+        content = choices[0].get("message", {}).get("content", "") or ""
+    model = result.get("model", "") or ""
+    usage = result.get("usage", {}) or {}
+    return content, model, usage
 
 
 @app.get("/healthz")
-async def healthz():
-    """健康检查."""
+async def healthz() -> dict[str, Any]:
+    """健康检查（liveness）."""
     return {"status": "ok", "service": "ai"}
 
 
-@app.post("/chat", response_model=ChatResponse)
-async def chat(req: ChatRequest):
-    """LLM 聊天接口."""
+@app.get("/readyz")
+async def readyz() -> dict[str, Any]:
+    """就绪检查（readiness）.
+
+    LLM 未配置时仍返回 200，但标记 degraded=true，调用方可据此判断是否路由流量。
+    """
+    llm_configured = settings.llm_available
+    return {
+        "status": "ok",
+        "service": "ai",
+        "llm_configured": llm_configured,
+        "degraded": not llm_configured,
+        "openai_base_url": settings.openai_base_url,
+    }
+
+
+@router.post("/chat", response_model=ChatResponse)
+async def chat(req: ChatRequest) -> ChatResponse:
+    """LLM 聊天接口（无 API key 时降级返回骨架响应）."""
     with tracer.start_as_current_span("ai_chat"):
-        # P5 骨架：实际调用 OpenAI/Anthropic API
-        # 需要从环境变量获取 API key
-        return {
-            "content": "P5 skeleton - LLM integration pending",
-            "model": req.model,
-            "usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
-        }
+        result = await chat_completion(
+            messages=req.messages,
+            model=req.model,
+            temperature=req.temperature,
+            api_key=settings.openai_api_key,
+            base_url=settings.openai_base_url,
+        )
+        if result is None:
+            logger.warning("chat_degraded", model=req.model)
+            return ChatResponse(
+                content="[degraded] LLM unavailable - returning skeleton response",
+                model=req.model,
+                usage={"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
+                degraded=True,
+            )
+        content, model, usage = _extract_content(result)
+        return ChatResponse(
+            content=content,
+            model=model or req.model,
+            usage=usage,
+            degraded=False,
+        )
 
 
-@app.post("/chat/stream")
-async def chat_stream(req: ChatRequest):
-    """流式聊天（SSE）."""
+@router.post("/chat/stream")
+async def chat_stream(req: ChatRequest) -> StreamingResponse:
+    """流式聊天（SSE，无 API key 时降级返回骨架 SSE）."""
 
-    async def generate():
+    async def generate() -> AsyncGenerator[str, None]:
         with tracer.start_as_current_span("ai_chat_stream"):
-            # P5 骨架：流式调用 LLM
-            yield "data: P5 skeleton\n\n"
-            yield "data: [DONE]\n\n"
+            async for chunk in chat_completion_stream(
+                messages=req.messages,
+                model=req.model,
+                temperature=req.temperature,
+                api_key=settings.openai_api_key,
+                base_url=settings.openai_base_url,
+            ):
+                yield chunk
 
     return StreamingResponse(generate(), media_type="text/event-stream")
 
 
-@app.post("/generate/question")
-async def generate_question(prompt: str):
-    """生成题目."""
+@router.post("/generate/question")
+async def generate_question(prompt: str) -> dict[str, Any]:
+    """生成题目（无 API key 时降级返回骨架）."""
     with tracer.start_as_current_span("generate_question"):
+        messages = [
+            {
+                "role": "system",
+                "content": "You are an educational question generator. "
+                "Generate a clear, concise question based on the user's prompt.",
+            },
+            {"role": "user", "content": prompt},
+        ]
+        result = await chat_completion(
+            messages=messages,
+            model="gpt-4o-mini",
+            temperature=0.7,
+            api_key=settings.openai_api_key,
+            base_url=settings.openai_base_url,
+        )
+        if result is None:
+            logger.warning("generate_question_degraded", prompt=prompt[:100])
+            return {
+                "success": True,
+                "data": {"question": "[degraded] question generation skeleton"},
+                "degraded": True,
+            }
+        content, _, _ = _extract_content(result)
         return {
             "success": True,
-            "data": {"question": "P5 skeleton - question generation pending"},
+            "data": {"question": content},
+            "degraded": False,
         }
 
 
-@app.post("/optimize/expression")
-async def optimize_expression(text: str):
-    """优化表达."""
+@router.post("/optimize/expression")
+async def optimize_expression(text: str) -> dict[str, Any]:
+    """优化表达（无 API key 时降级返回骨架）."""
     with tracer.start_as_current_span("optimize_expression"):
+        messages = [
+            {
+                "role": "system",
+                "content": "You are a writing assistant. "
+                "Optimize the user's text for clarity, conciseness, and tone.",
+            },
+            {"role": "user", "content": text},
+        ]
+        result = await chat_completion(
+            messages=messages,
+            model="gpt-4o-mini",
+            temperature=0.5,
+            api_key=settings.openai_api_key,
+            base_url=settings.openai_base_url,
+        )
+        if result is None:
+            logger.warning("optimize_expression_degraded", text=text[:100])
+            return {
+                "success": True,
+                "data": {"optimized": "[degraded] expression optimization skeleton"},
+                "degraded": True,
+            }
+        content, _, _ = _extract_content(result)
         return {
             "success": True,
-            "data": {"optimized": "P5 skeleton - expression optimization pending"},
+            "data": {"optimized": content},
+            "degraded": False,
         }
+
+
+app.include_router(router)