pj0235-eai_agentplatform/docs/04_Backend/BE04_AI_Chat_Module.md

BE04 — AI PathCoach 模块设计

版本：V1.1 | httpx 适配器 + 配置链 + Fail Fast | 参考：pj006-zhilianyuan2 llm/openai_adapter.py 配置优先级：system_config 数据库表 → .env 文件

---

1. 模块职责

• 全局 AI 聊天框的后端支持
• 上下文注入（当前产品/课程信息自动带入）
• 知识检索（MySQL FULLTEXT 召回 → Prompt 注入）
• SSE 流式响应 + 非流式调用（快捷动作）
• 3 个快捷动作（情景演练/查佣金/产品对比）

2. 架构

用户消息 + 页面上下文
      │
      ▼
  ┌──────────────────┐
  │  知识检索          │
  │  MySQL FULLTEXT   │
  │  → 匹配段落       │
  └──────┬───────────┘
         │ 上下文片段
         ▼
  ┌──────────────────┐
  │  Prompt 组装      │
  │  System Prompt   │
  │  + 知识上下文     │
  │  + 对话历史       │
  └──────┬───────────┘
         │
         ▼
  ┌──────────────────────────────┐
  │  build_llm_adapter() 工厂    │
  │  → 读取配置（库→.env）       │
  │  → Fail Fast 缺配置抛 501   │
  │  → 返回 OpenAICompatible    │
  └──────┬───────────────────────┘
         │
         ▼
  ┌──────────────────┐
  │  httpx 调用       │
  │  /chat/completions│
  │  SSE 流式 / 非流式│
  └──────────────────┘

3. 配置优先级与 Fail Fast

# services/ai_service.py
from __future__ import annotations
import httpx
from typing import Generator
from dataclasses import dataclass, field
from app.core.config import Settings
from app.errors import AppError


class LLMNotConfiguredError(AppError):
    """LLM 未配置（缺 api_key / base_url / model）"""
    status_code = 501
    error_code = "llm_not_configured"


@dataclass
class LLMConfig:
    """LLM 连接所需的三项配置"""
    base_url: str
    api_key: str
    model: str


def resolve_llm_config(settings: Settings) -> LLMConfig:
    """按优先级链解析 LLM 配置。缺任何一项即抛 LLMNotConfiguredError。

    优先级（高 → 低）：
    1. settings.llm_*（来自 system_config 数据库表）
    2. settings 中从 .env 读取的默认值
    """
    base_url = getattr(settings, "llm_base_url", None) or ""
    api_key = getattr(settings, "llm_api_key", None) or ""
    model = getattr(settings, "llm_model", None) or ""

    missing = []
    if not base_url:
        missing.append("llm_base_url")
    if not api_key:
        missing.append("llm_api_key")
    if not model:
        missing.append("llm_model")

    if missing:
        raise LLMNotConfiguredError(
            f"LLM 服务未配置——缺失：{', '.join(missing)}。"
            f"请管理员在【系统参数配置】中补充。"
        )

    return LLMConfig(base_url=base_url, api_key=api_key, model=model)

4. LLM 适配器（httpx 实现，参考 zhilianyuan2 openai_adapter.py）

使用 httpx 替代 OpenAI SDK，减少依赖、更可控、支持 token usage 采集。

# services/ai_service.py

class LLMAdapter:
    """OpenAI 兼容接口适配器（httpx 实现，无 openai SDK 依赖）"""

    def __init__(self, *, config: LLMConfig, http_client: httpx.Client | None = None):
        self._base_url = config.base_url.rstrip("/")
        self._api_key = config.api_key
        self._model = config.model
        self._client = http_client or httpx.Client(timeout=60.0)
        # 最后一次流式调用的 token 用量（供日志埋点）
        self.last_stream_usage: dict[str, int] = {}

    def _headers(self) -> dict[str, str]:
        return {
            "Authorization": f"Bearer {self._api_key}",
            "Content-Type": "application/json",
        }

    def _payload(self, messages: list[dict], *, stream: bool, **kwargs) -> dict:
        payload = {
            "model": self._model,
            "messages": messages,
            "stream": stream,
            "temperature": kwargs.get("temperature", 0.7),
            "max_tokens": kwargs.get("max_tokens", 2048),
        }
        if stream:
            payload["stream_options"] = {"include_usage": True}
        return payload

    def generate(self, messages: list[dict], **kwargs) -> str:
        """非流式调用，返回完整正文。用于快捷动作等一次性请求。"""
        try:
            resp = self._client.post(
                f"{self._base_url}/chat/completions",
                headers=self._headers(),
                json=self._payload(messages, stream=False, **kwargs),
            )
            resp.raise_for_status()
            data = resp.json()
            content = data["choices"][0]["message"]["content"]
            if not content or not content.strip():
                raise LLMError("LLM 返回空正文")
            return content
        except httpx.HTTPError as e:
            raise LLMError(f"LLM 调用失败：{e}") from e
        except (KeyError, ValueError) as e:
            raise LLMError(f"LLM 响应解析失败：{e}") from e

    def generate_stream(self, messages: list[dict], **kwargs) -> Generator[str, None, None]:
        """SSE 流式调用。逐 chunk yield 文本，末包采集 usage。"""
        self.last_stream_usage = {}
        try:
            with self._client.stream(
                "POST",
                f"{self._base_url}/chat/completions",
                headers=self._headers(),
                json=self._payload(messages, stream=True, **kwargs),
            ) as resp:
                resp.raise_for_status()
                for line in resp.iter_lines():
                    if not line or not line.startswith("data: "):
                        continue
                    payload = line[6:].strip()
                    if payload == "[DONE]":
                        break
                    chunk = json.loads(payload)
                    # 末包采集 usage（include_usage=true）
                    usage = chunk.get("usage")
                    if usage:
                        self.last_stream_usage = {
                            "input_tokens": int(usage.get("prompt_tokens", 0) or 0),
                            "output_tokens": int(usage.get("completion_tokens", 0) or 0),
                        }
                    choices = chunk.get("choices", [])
                    if not choices:
                        continue
                    delta = choices[0].get("delta", {})
                    content = delta.get("content", "")
                    if content:
                        yield content
        except httpx.HTTPError as e:
            raise LLMError(f"LLM 流式调用失败：{e}") from e
        except (KeyError, ValueError) as e:
            raise LLMError(f"LLM 流式响应解析失败：{e}") from e


def build_llm_adapter(settings: Settings) -> LLMAdapter:
    """工厂方法：解析配置 → 构造适配器。配置不全即 Fail Fast。"""
    config = resolve_llm_config(settings)
    return LLMAdapter(config=config)

5. 知识检索

# services/ai_service.py

def retrieve_knowledge(keywords: str, db: Session, top_k: int = 5) -> list[str]:
    """MySQL 全文索引检索知识块"""
    results = db.execute(
        text(
            "SELECT content FROM knowledge_chunk "
            "WHERE MATCH(content) AGAINST(:keywords IN NATURAL LANGUAGE MODE) "
            "LIMIT :limit"
        ),
        {"keywords": keywords, "limit": top_k},
    ).fetchall()
    return [r[0] for r in results]

6. System Prompt

SYSTEM_PROMPT = """你是一个博昇内部培训平台的 AI 助教 PathCoach。

你的职责：
1. 解答公司介绍、产品知识、佣金规则、销售话术、业务规则相关的问题
2. 严格依赖已审批知识库的内容回答
3. 如果知识库中未找到相关资料，明确回答「未找到相关资料」，不得臆测

禁止行为：
1. 禁止闲聊
2. 禁止编造数据
3. 禁止回答超出业务范围的问题
4. 禁止泄露敏感信息

当前页面上下文：
{page_context}

知识库相关片段：
{knowledge_context}
"""

7. API 实现

# api/ai_chat.py
from fastapi.responses import StreamingResponse

router = APIRouter(prefix="/api/ai-chat", tags=["ai_chat"])

@router.post("/message")
def chat_message(
    request: ChatRequest,
    current_user: CurrentUser,
    db: Session = Depends(get_db),
    settings: Settings = Depends(get_settings),
):
    """SSE 流式对话"""
    # 1. 检索知识
    knowledge = retrieve_knowledge(request.message, db)

    # 2. 组装 Prompt
    system = SYSTEM_PROMPT.format(
        page_context=json.dumps(request.context or {}),
        knowledge_context="\n\n".join(knowledge),
    )
    messages = [
        {"role": "system", "content": system},
        *request.history,
        {"role": "user", "content": request.message},
    ]

    # 3. 构建 LLM 适配器（缺配置即抛 501）
    llm = build_llm_adapter(settings)

    def generate():
        for chunk in llm.generate_stream(messages):
            yield f"data: {json.dumps({'type': 'text', 'content': chunk})}\n\n"
        yield "data: {\"type\": \"done\"}\n\n"

    return StreamingResponse(
        generate(), media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


@router.get("/quick-actions")
def get_quick_actions():
    """获取 3 个快捷按钮"""
    return {"data": {"actions": [
        {"id": "scenario", "label": "客户情景演练"},
        {"id": "commission", "label": "查询佣金/规则"},
        {"id": "compare", "label": "产品对比"},
    ]}}


@router.post("/quick-action")
def trigger_quick_action(
    request: QuickActionRequest,
    settings: Settings = Depends(get_settings),
):
    """触发快捷动作（非流式 LLM 调用）"""
    llm = build_llm_adapter(settings)

    if request.action_id == "commission":
        # 查佣金：构建 prompt → 非流式调用
        prompt = f"查询产品佣金信息，产品参数：{request.params}"
        resp = llm.generate([{"role": "user", "content": prompt}])
        return {"data": {"result": resp}}

    elif request.action_id == "compare":
        prompt = f"对比以下产品：{request.params}"
        resp = llm.generate([{"role": "user", "content": prompt}])
        return {"data": {"result": resp}}

    elif request.action_id == "scenario":
        # 情景演练：返回初始话术，后续走流式对话
        prompt = f"开始销售情景演练，场景参数：{request.params}"
        resp = llm.generate([{"role": "user", "content": prompt}])
        return {"data": {"result": resp, "mode": "scenario"}}

8. 上下文注入规则

页面	自动注入上下文	说明
产品详情	product_id, product_name, product_code	自动带入当前产品
课程详情	course_id, course_name, related_product	自动带入当前课程及关联产品
公司介绍	page: "company_intro"	提示 AI 当前页为公司介绍

9. 配置项

配置键	来源	说明
llm_base_url	system_config 表 / .env	LLM 服务地址，如 http://192.168.1.100:11434/v1
llm_api_key	system_config 表 / .env	API Key，本地 Ollama 可填 ollama
llm_model	system_config 表 / .env	模型名，如 qwen2.5:7b

10. 错误处理

场景	HTTP 状态	响应
LLM 未配置（缺 base_url/key/model）	501	{"error": "llm_not_configured", "message": "请管理员在系统参数配置中补充..."}
LLM 调用超时/网络错误	502	{"error": "llm_request_failed", "message": "LLM 服务不可达，请检查网络连接"}
LLM 返回空正文	502	{"error": "llm_empty_response", "message": "LLM 返回空结果"}
LLM 响应格式异常	502	{"error": "llm_response_error", "message": "LLM 响应异常"}