refactor(fam-edge): 问答链路抽离到独立 ai-gateway 服务，fam-edge 改为转发客户端

原本嵌在 fam-edge 里的问答模型降级链（NVIDIA 文字模型 -> Gemini 非 flash 文字
模型 -> 本地 Ollama 兜底，含 key 轮换/熔断）跟视频分析业务无关，是通用能力，
抽成独立 ai-gateway 服务（OpenAI 兼容协议），除了 fam-edge 自己，别的项目也能
直接接入。

- qa.py 重写为 HTTP 转发客户端，调 ai-gateway 的 /v1/chat/completions，翻译回
  原有 run_qa/run_qa_stream 契约，api_gateway.py 和 fam-core 调用方零改动
- 删除 model_adapters/ollama_adapter.py 及其测试（问答专用，视频分析不需要本地模型）
- gemini_adapter.py / nvidia_adapter.py 移除 chat()/chat_stream() 及问答专用超时
  （只保留视频分析用的 analyze_video）
- app.py 移除 Ollama 预热逻辑（现在由 ai-gateway 自己负责）
- config.yaml 移除 3 个问答专用 model 条目，新增 ai_gateway 客户端配置块

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

fam-edge/config/config.yaml

@@ -103,7 +103,16 @@ person_identifier:
     max_retries: 2
     retry_backoff_sec: 3
 
-# 智能问答降级链（与视频分析独立）：Gemini -> NVIDIA -> 本地 Ollama
+# 智能问答（2026-08-23 抽离到独立 ai-gateway 服务，OpenAI 兼容协议）：
+# fam-edge 这边只是转发客户端，问答本体的模型降级链/key 轮换/熔断都在
+# ai-gateway 自己的 config.yaml 里配置，这里只填怎么连它。
+# token 走 .env AI_GATEWAY_TOKEN，跟 ai-gateway 侧配置的值必须一致。
+ai_gateway:
+  base_url: "http://127.0.0.1:5100"   # 同机部署，走本地回环，不走公网
+  token: "${AI_GATEWAY_TOKEN}"
+  timeout: 60
+
+# 视频分析模型链
 models:
   - provider: "gemini"
     role: "vision"
@@ -127,9 +136,6 @@ models:
       - "智能摄像头-3"
       - "智能摄像头-4"
     timeout: 600
-    # 问答专用超时（跟上面视频分析的 timeout 分开）：用户在等交互式回答，一个
-    # key/模型卡住不该等 10 分钟，超时要短，快速降级到下一个 key/模型/provider
-    chat_timeout: 20
     # 模型级独立超时（最终值，不参与编排层 ×2 放大）
     # gemini-flash-lite 实测 ~22-34s，按用户要求放宽至 8 分钟（480s），避免大视频/排队时过早切断
     model_timeouts:
@@ -159,7 +165,6 @@ models:
     base_url: "https://integrate.api.nvidia.com/v1"
     api_key: "${NVIDIA_API_KEY}"
     timeout: 600
-    chat_timeout: 20             # 问答专用超时，跟视频分析的 timeout 分开
     max_base64_mb: 20           # 超过此大小直接跳过 NVIDIA，不做注定失败的编码+上传
     switch_interval_sec: 5      # 模型切换间隔：一个失败后等待再试下一个（未来加模型时用）
     model_timeouts:             # 模型级独立超时（最终值，不参与 ×2）
@@ -169,75 +174,3 @@ models:
       threshold: 5
       cooldown: 300
 
-  # 问答专用 NVIDIA 文字模型链（2026-08-23 新增，跟上面视频分析用的 omni 模型
-  # 完全独立）：用户要求问答不用 flash/omni，优先找 NVIDIA 免费文字模型里上下文
-  # 最大的几个。实测(2026-08-23)在当前账号可用、非 deprecated 的候选里：
-  #   nemotron-3-ultra-550b-a55b: 1M 上下文，561B，最强，free endpoint 已验证可调
-  #   nemotron-3-super-120b-a12b: 1M 上下文，124B，同为 Nemotron-3 系列备份
-  #   openai/gpt-oss-120b: 131K 上下文，117B，不同厂商备份（Nemotron 系列整体
-  #     出问题时的多样性兜底）
-  # 淘汰原因记录：meta/llama-3.1-70b-instruct、nvidia/llama-3.3-nemotron-
-  # super-49b-v1.5 均已收到 "will be deprecated on 08/25/2026" 通知，不用；
-  # nvidia/llama-3.1-nemotron-ultra-253b-v1、mistralai/mistral-large-2-instruct、
-  # nvidia/nemotron-4-340b-instruct、moonshotai/kimi-k2.6 在 /v1/models 目录
-  # 里能看到，但实测调用 chat.completions 返回 404 "Not found for account"
-  # （免费层没有这些模型的调用权限，文档列出不代表能调）。
-  - provider: "nvidia"
-    role: "text"
-    usage: "qa_primary"
-    enabled: true
-    model_name: "nvidia/nemotron-3-ultra-550b-a55b"
-    fallback_models:
-      - "nvidia/nemotron-3-super-120b-a12b"
-      - "openai/gpt-oss-120b"
-    base_url: "https://integrate.api.nvidia.com/v1"
-    api_key: "${NVIDIA_API_KEY}"
-    timeout: 600
-    # 这几个都是"推理"模型，回答前会先输出一段思考过程再给最终答案，比普通模型
-    # 更费 token/更慢，问答超时给宽松一点（不是简单文字模型那种 20s 就该出结果）
-    chat_timeout: 45
-    switch_interval_sec: 3
-    circuit_breaker:
-      enabled: true
-      threshold: 5
-      cooldown: 300
-
-  # 问答专用 Gemini 非 flash 文字模型（2026-08-23 新增）：用户明确要求问答不用
-  # flash，这里走 gemini-pro-latest（1M 上下文，跟 gemini-flash-latest 同样的
-  # "-latest" 别名习惯，自动跟最新 pro 版本），失败再退 gemini-2.5-pro。
-  # key 复用视频分析同一批（各自独立 Google Cloud 项目，配额互不影响）。
-  - provider: "gemini"
-    role: "text"
-    usage: "qa_primary"
-    enabled: true
-    model_name: "gemini-pro-latest"
-    fallback_models:
-      - "gemini-2.5-pro"
-    api_key: "${GEMINI_API_KEY}"
-    extra_api_keys:
-      - "${GEMINI_API_KEY_2}"
-      - "${GEMINI_API_KEY_3}"
-      - "${GEMINI_API_KEY_4}"
-    key_labels:
-      - "智能摄像头-1"
-      - "智能摄像头-2"
-      - "智能摄像头-3"
-      - "智能摄像头-4"
-    timeout: 60
-    chat_timeout: 30
-    circuit_breaker:
-      enabled: true
-      threshold: 5
-      cooldown: 300
-
-  # 本地模型：纯文本 qwen2.5:7b，仅参与智能问答兜底
-  - provider: "ollama"
-    role: "text"
-    usage: "qa_fallback"
-    enabled: true
-    model_name: "qwen2.5:7b"
-    base_url: "http://localhost:11434"
-    timeout: 120
-    num_predict: 512
-    circuit_breaker:
-      enabled: false

fam-edge/src/fam_edge/app.py

@@ -8,7 +8,6 @@ FAM-Edge 主应用 - Flask 单进程（新架构 v2.1）
 """
 import os
 import sys
-import threading
 from flask import Flask, jsonify
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
@@ -19,7 +18,6 @@ from .api_gateway.api_gateway import api_bp
 from . import state
 from .video_queue import VideoQueue
 from .person_service import PersonService
-from .model_adapters.ollama_adapter import OllamaAdapter
 
 logger = setup_logger('fam-edge.app')
 
@@ -49,21 +47,6 @@ try:
 except Exception as e:
     logger.error(f"后台服务启动失败: {e}", exc_info=True)
 
-# 启动时后台预热 Ollama（问答链路最末位兜底，实测从未被自然触发过，
-# OLLAMA_KEEP_ALIVE=-1 只保证加载后不换出、不负责主动预加载）。后台线程跑，
-# 不阻塞 gunicorn worker 启动；找不到 ollama 配置项或预热失败都只记警告。
-try:
-    _ollama_cfg = next(
-        (m for m in load_config().get('models', []) if m.get('provider') == 'ollama'),
-        None)
-    if _ollama_cfg and _ollama_cfg.get('enabled', False):
-        threading.Thread(
-            target=lambda: OllamaAdapter(_ollama_cfg).warm_up(),
-            daemon=True, name='ollama-warmup').start()
-        logger.info("Ollama 预热任务已在后台启动")
-except Exception as e:
-    logger.warning(f"Ollama 预热任务启动失败（不影响主服务）: {e}")
-
 
 if __name__ == '__main__':
     cfg = load_config()

fam-edge/src/fam_edge/model_adapters/__init__.py

@@ -1,14 +1,12 @@
 """模型适配器包"""
 from .base_adapter import BaseModelAdapter
 from .circuit_breaker import CircuitBreaker
-from .ollama_adapter import OllamaAdapter
 from .gemini_adapter import GeminiAdapter
 from .adapter_factory import build_adapter, build_adapters, register_adapter
 
 __all__ = [
     "BaseModelAdapter",
     "CircuitBreaker",
-    "OllamaAdapter",
     "GeminiAdapter",
     "build_adapter",
     "build_adapters",

fam-edge/src/fam_edge/model_adapters/adapter_factory.py

@@ -9,15 +9,16 @@
 from typing import List
 
 from .base_adapter import BaseModelAdapter
-from .ollama_adapter import OllamaAdapter
 from .gemini_adapter import GeminiAdapter
 from .nvidia_adapter import NvidiaVisionAdapter
 from ..logger import setup_logger
 
 logger = setup_logger('fam-edge.adapter_factory')
 
+# ollama 已于 2026-08-23 移除：本地模型只在问答链路里当兜底用，问答已经整个
+# 抽离到独立的 ai-gateway 服务（含它自己的 ollama 适配器），fam-edge 这边
+# 只剩视频分析（vision 角色），不再需要注册纯文本本地模型。
 _ADAPTER_REGISTRY = {
-    "ollama": OllamaAdapter,
     "gemini": GeminiAdapter,
     "nvidia": NvidiaVisionAdapter,
 }

fam-edge/src/fam_edge/model_adapters/base_adapter.py

@@ -37,11 +37,9 @@ class BaseModelAdapter(ABC):
     def __init__(self, provider_name: str, config: dict):
         self.provider_name = provider_name  # 如 "ollama", "gemini"
         self.config = config
-        # 角色: vision=视觉分析, text=智能问答; 默认 vision
+        # 角色: vision=视觉分析, text=智能问答（问答链路已抽离到 ai-gateway，
+        # 这里目前只有 vision 在用；text 角色留给尚未清理的旧 person_service）
         self.role = config.get('role', 'vision')
-        # usage: 纯文档/编排层筛选用的标记（如 "qa_primary"/"qa_fallback"），
-        # 不影响本适配器自身行为；QAOrchestrator 据此挑选参与问答链路的适配器。
-        self.usage = config.get('usage', '')
         # 模型调用统计回调（由编排层注入）:
         # hook(provider, model, started_at, duration_sec, success, error)
         self.model_call_hook = None

fam-edge/src/fam_edge/model_adapters/gemini_adapter.py

@@ -3,8 +3,12 @@ GeminiAdapter - Google Gemini 云端 VLM 适配器
 
 provider_name = "gemini"
 模型: gemini-flash-latest
-角色: vision (整视频直出结构化 JSON) + 智能问答
+角色: vision (整视频直出结构化 JSON)
 健康检查: GET /v1beta/models?key=...
+
+问答（chat/chat_stream）已于 2026-08-23 抽离到独立的 ai-gateway 服务
+（跟视频分析业务无关，是通用能力），这里不再实现，fam-edge 自己的问答请求
+转发给 ai-gateway（见 qa.py）。
 熔断器: 启用
 整视频分析: 用 Files API 上传完整视频 -> generateContent 直出结构化 JSON
   （本地不切片、不抽帧；Gemini 原生支持长视频）
@@ -50,7 +54,7 @@ logger = setup_logger('fam-edge.gemini_adapter')
 
 
 class GeminiAdapter(BaseModelAdapter):
-    """Gemini 云端 VLM 适配器 (整视频直出结构化 JSON + 文本问答)"""
+    """Gemini 云端 VLM 适配器 (整视频直出结构化 JSON，不参与问答)"""
 
     def __init__(self, config: dict):
         super().__init__("gemini", config)
@@ -74,10 +78,6 @@ class GeminiAdapter(BaseModelAdapter):
                 self.key_labels.append(str(label))
         self.api_key = self.api_keys[0] if self.api_keys else ''  # 向后兼容单 key 用法
         self.timeout = config.get('timeout', 600)
-        # 问答（chat）专用超时——跟视频分析的 timeout 分开，不能共用 600s：
-        # 智能问答是同步等待用户看结果的交互场景，一个 key/模型卡住不该让用户等
-        # 10 分钟，超时应该短、快速降级到下一个 key/模型/provider
-        self.chat_timeout = config.get('chat_timeout', 20)
         # 模型级独立超时（最终值，不参与编排层 ×N 放大）: {model_name: seconds}
         # 例: {"gemini-flash-lite-latest": 90}（按实测耗时 ×4 配置）
         self.model_timeouts = {
@@ -401,128 +401,6 @@ class GeminiAdapter(BaseModelAdapter):
         camera = load_config().get('gdrive_sync', {}).get('camera_name', '')
         return build_video_prompt(known_members, event_start_time, camera)
 
-    # ------------------------------------------------------------------
-    # 智能问答：纯文本
-    # ------------------------------------------------------------------
-    def chat(self, prompt: str, max_tokens: int = 512) -> Optional[str]:
-        if self._cb.is_open():
-            logger.warning("Gemini 熔断器 OPEN，跳过问答")
-            return None
-        if not self.api_keys:
-            logger.warning("Gemini API Key 未配置，跳过问答")
-            return None
-        try:
-            result = self._generate_text(prompt, max_tokens=max_tokens, temperature=0.3)
-        except Exception as e:
-            logger.error(f"Gemini 问答异常: {e}")
-            result = None
-        if result:
-            self._cb.record_success()
-        else:
-            self._cb.record_failure()
-        return result
-
-    def _generate_text(self, text: str, max_tokens: int, temperature: float) -> Optional[str]:
-        """纯文本 generateContent，按 key 轮换（同 analyze_video 共用一套轮转起点）
-        × 模型 fallback 链依次尝试。"""
-        for idx, api_key in self._rotated_keys():
-            key_label = self.key_labels[idx]
-            for model in self.model_chain:
-                try:
-                    resp = requests.post(
-                        f"{self._base_url}/models/{model}:generateContent?key={api_key}",
-                        json={"contents": [{"parts": [{"text": text}]}],
-                              "generationConfig": {
-                                  "temperature": temperature,
-                                  "maxOutputTokens": max_tokens}},
-                        timeout=self.chat_timeout
-                    )
-                except requests.Timeout:
-                    logger.warning(f"Gemini {key_label} [{model}] 问答超时({self.chat_timeout}s)")
-                    continue
-                except Exception as e:
-                    logger.error(f"Gemini {key_label} [{model}] 问答异常: {e}")
-                    continue
-                if resp.status_code == 200:
-                    cands = resp.json().get('candidates', [])
-                    out = ''.join(
-                        p.get('text', '')
-                        for p in (cands[0].get('content', {}) if cands else {}).get('parts', [])
-                    ).strip() if cands else ''
-                    if out:
-                        return out
-                elif resp.status_code == 429:
-                    logger.warning(f"Gemini {key_label} [{model}] 429，切换下一模型/Key")
-                    continue
-        return None
-
-    def chat_stream(self, prompt: str, max_tokens: int = 512):
-        """流式问答：逐块 yield 文本增量。用于聊天界面边生成边显示，不用等全量
-        返回再展示——之前整段等待是"卡住没反馈"体验差的根源之一。
-
-        按 key 轮换 × 模型链依次尝试，但只在"这次尝试还没吐出任何文本"时才允许
-        换下一个 key/模型；一旦已经开始吐字给用户看了，中途出错就直接结束这次
-        生成（不再悄悄换 provider 接着写，否则会出现两段风格/内容不连贯的回答
-        拼在一起，比直接告知"生成中断"更让人困惑）。
-        """
-        if self._cb.is_open():
-            logger.warning("Gemini 熔断器 OPEN，跳过问答（流式）")
-            return
-        if not self.api_keys:
-            logger.warning("Gemini API Key 未配置，跳过问答（流式）")
-            return
-        got_any = False
-        for idx, api_key in self._rotated_keys():
-            key_label = self.key_labels[idx]
-            for model in self.model_chain:
-                try:
-                    resp = requests.post(
-                        f"{self._base_url}/models/{model}:streamGenerateContent"
-                        f"?alt=sse&key={api_key}",
-                        json={"contents": [{"parts": [{"text": prompt}]}],
-                              "generationConfig": {
-                                  "temperature": 0.3, "maxOutputTokens": max_tokens}},
-                        timeout=self.chat_timeout, stream=True,
-                    )
-                except requests.Timeout:
-                    logger.warning(f"Gemini {key_label} [{model}] 流式问答超时({self.chat_timeout}s)")
-                    continue
-                except Exception as e:
-                    logger.error(f"Gemini {key_label} [{model}] 流式问答异常: {e}")
-                    continue
-                if resp.status_code != 200:
-                    logger.warning(f"Gemini {key_label} [{model}] 流式问答 HTTP {resp.status_code}")
-                    resp.close()
-                    continue
-                # Gemini 响应体固定是 UTF-8，但 Content-Type 没带 charset 参数，
-                # requests 会自己猜编码（猜错会把中文变成乱码）——强制指定，
-                # 不依赖 requests 的自动嗅探。
-                resp.encoding = 'utf-8'
-                try:
-                    for line in resp.iter_lines(decode_unicode=True):
-                        if not line or not line.startswith('data: '):
-                            continue
-                        chunk = line[len('data: '):]
-                        try:
-                            obj = json.loads(chunk)
-                        except ValueError:
-                            continue
-                        cands = obj.get('candidates', [])
-                        text = ''.join(
-                            p.get('text', '')
-                            for p in (cands[0].get('content', {}) if cands else {}).get('parts', []))
-                        if text:
-                            got_any = True
-                            yield text
-                except Exception as e:
-                    logger.warning(f"Gemini {key_label} [{model}] 流式读取中断: {e}")
-                finally:
-                    resp.close()
-                if got_any:
-                    self._cb.record_success()
-                    return  # 已经开始吐字，不管这次是否读完都不再换 provider
-        self._cb.record_failure()
-
     def get_timeout(self) -> int:
         return self.timeout
 

fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py

@@ -3,9 +3,12 @@ NvidiaVisionAdapter - NVIDIA NIM 云端 VLM 适配器
 
 provider_name = "nvidia"
 模型: nvidia/nemotron-3-nano-omni-30b-a3b-reasoning（唯一实测确认可用的视频理解模型）
-角色: vision (整视频直出结构化 JSON) + 智能问答
+角色: vision (整视频直出结构化 JSON)
 SDK: openai (NIM 兼容 OpenAI API 规范)
 
+问答（chat/chat_stream）已于 2026-08-23 抽离到独立的 ai-gateway 服务
+（跟视频分析业务无关，是通用能力），这里不再实现。
+
 整视频分析实测结论（2026-08-21 用真实短视频逐个探测）:
   - nemotron-3-nano-omni-30b-a3b-reasoning: video_url 只认 base64 data URI
     （`data:video/mp4;base64,<...>`），Assets API 的 asset_id 引用方式对它直接 500
@@ -44,7 +47,7 @@ except ImportError:
 
 
 class NvidiaVisionAdapter(BaseModelAdapter):
-    """NVIDIA NIM 云端 VLM 适配器 (整视频单次调用; 文本问答)
+    """NVIDIA NIM 云端 VLM 适配器 (整视频单次调用，不参与问答)
 
     多模型降级链（类似 Gemini flash -> flash-lite）:
       - model_chain = [model_name] + fallback_models
@@ -62,8 +65,6 @@ class NvidiaVisionAdapter(BaseModelAdapter):
         self.api_key = self._resolve_key(config.get('api_key', ''))
         self.base_url = config.get('base_url', 'https://integrate.api.nvidia.com/v1')
         self.timeout = config.get('timeout', 600)
-        # 问答专用超时，跟视频分析分开——交互式问答不该等到跟视频分析一样久
-        self.chat_timeout = config.get('chat_timeout', 20)
         self.max_base64_mb = float(config.get('max_base64_mb', 20))
         # 模型级独立超时（最终值，不参与编排层 ×N 放大）: {model_name: seconds}
         self.model_timeouts = {
@@ -194,34 +195,6 @@ class NvidiaVisionAdapter(BaseModelAdapter):
         camera = load_config().get('gdrive_sync', {}).get('camera_name', '')
         return build_video_prompt(known_members, event_start_time, camera)
 
-    # ------------------------------------------------------------------
-    # 智能问答：纯文本
-    # ------------------------------------------------------------------
-    def chat(self, prompt: str, max_tokens: int = 2048) -> Optional[str]:
-        if self._cb.is_open():
-            logger.warning("NVIDIA 熔断器 OPEN，跳过问答")
-            return None
-        if self._client is None:
-            logger.warning("NVIDIA 客户端未初始化，跳过问答")
-            return None
-        for model in self.model_chain:
-            try:
-                resp = self._client.chat.completions.create(
-                    model=model,
-                    messages=[{"role": "user", "content": prompt}],
-                    temperature=0.3,
-                    max_tokens=max_tokens,
-                    timeout=self.chat_timeout
-                )
-                content = resp.choices[0].message.content
-                if content:
-                    self._cb.record_success()
-                    return content.strip()
-            except Exception as e:
-                logger.warning(f"NVIDIA [{model}] 问答异常: {e}")
-        self._cb.record_failure()
-        return None
-
     def get_timeout(self) -> int:
         return self.timeout
 

fam-edge/src/fam_edge/model_adapters/ollama_adapter.py

@@ -1,128 +0,0 @@
-"""
-OllamaAdapter - 本地模型适配器（仅智能问答兜底）
-
-provider_name = "ollama"
-模型: qwen2.5:7b（纯文本）
-角色: text（智能问答兜底；Gemini 与 NVIDIA 均失败时启用）
-健康检查: GET /api/tags
-不参与视觉分析、不参与视频结构化输出（云端 VLM 直出）
-"""
-import requests
-from typing import Dict, Optional
-
-from .base_adapter import BaseModelAdapter
-from .circuit_breaker import CircuitBreaker
-from ..logger import setup_logger
-
-logger = setup_logger('fam-edge.ollama_adapter')
-
-
-class OllamaAdapter(BaseModelAdapter):
-    """Ollama 本地 VLM 适配器"""
-
-    def __init__(self, config: dict):
-        super().__init__("ollama", config)
-        self.base_url = config.get('base_url', 'http://localhost:11434')
-        self.model_name = config.get('model_name', 'llava-phi3')
-        self.timeout = config.get('timeout', 240)
-        self.num_predict = config.get('num_predict', 500)
-        cb_cfg = config.get('circuit_breaker', {})
-        self._cb = CircuitBreaker(
-            threshold=cb_cfg.get('threshold', 5),
-            cooldown=cb_cfg.get('cooldown', 900),
-            enabled=cb_cfg.get('enabled', False)  # 本地模型默认不启用
-        )
-
-    def warm_up(self) -> bool:
-        """启动时主动送一次最小请求，把模型强制加载进内存。
-
-        背景：OLLAMA_KEEP_ALIVE=-1（systemd 环境变量已配置）只保证"一旦加载过
-        就不再因为空闲被换出"，但不会在服务启动时主动预加载——Ollama 现在只在
-        问答链路最末位兜底（前面 NVIDIA/Gemini 一直成功的话永远轮不到它），
-        实测 model_calls 表里从来没有一条 ollama 记录，说明模型从未被加载过。
-        真正需要兜底的那一刻才现加载，用户会等上首次冷启动的 ~1-2 分钟
-        （见 README 6.2 冷启动实测数据）。启动时主动预热一次，之后就一直
-        常驻内存，兜底真正触发时不再有冷启动延迟。
-        """
-        try:
-            resp = requests.post(
-                f"{self.base_url}/api/generate",
-                json={"model": self.model_name, "prompt": "hi", "stream": False,
-                      "options": {"num_predict": 1}},
-                timeout=180,  # 冷启动可能到 1-2 分钟，给足时间
-            )
-            if resp.status_code == 200:
-                logger.info(f"Ollama 模型预热完成: {self.model_name}")
-                return True
-            logger.warning(f"Ollama 预热失败: HTTP {resp.status_code} {resp.text[:200]}")
-        except Exception as e:
-            logger.warning(f"Ollama 预热异常（不影响服务启动，问答兜底时会正常现加载）: {e}")
-        return False
-
-    def health_check(self) -> bool:
-        """GET /api/tags，检查模型是否可用"""
-        try:
-            resp = requests.get(f"{self.base_url}/api/tags", timeout=10)
-            if resp.status_code == 200:
-                models = resp.json().get('models', [])
-                model_names = [m.get('name', '') for m in models]
-                # 兼容 llava-phi3:latest 等后缀
-                has_model = any(self.model_name in name for name in model_names)
-                if has_model:
-                    logger.info(f"Ollama 健康检查通过: 模型 {self.model_name} 可用")
-                    return True
-                else:
-                    logger.warning(f"Ollama 健康检查失败: 模型 {self.model_name} 未找到，可用模型: {model_names}")
-                    return False
-            return False
-        except Exception as e:
-            logger.error(f"Ollama 健康检查异常: {e}")
-            return False
-
-    def analyze_video(self, video_path: str,
-                     known_members_context: str,
-                     event_start_time: str = '') -> Optional[Dict]:
-        """Ollama 为纯文本模型，不参与视频分析，返回 None（降级链不会选它做视频）。"""
-        logger.info("Ollama 为纯文本模型，跳过视频分析")
-        return None
-
-    def get_timeout(self) -> int:
-        return self.timeout
-
-    def get_circuit_breaker(self) -> CircuitBreaker:
-        return self._cb
-
-    # ------------------------------------------------------------------
-    # 智能问答：纯文本（本地模型，仅作 Gemini/NVIDIA 全失败时的兜底）
-    # ------------------------------------------------------------------
-    def chat(self, prompt: str, max_tokens: int = 512) -> Optional[str]:
-        if self._cb.is_open():
-            logger.warning("Ollama 熔断器 OPEN，跳过问答")
-            return None
-        try:
-            resp = requests.post(
-                f"{self.base_url}/api/generate",
-                json={
-                    "model": self.model_name,
-                    "prompt": prompt,
-                    "stream": False,
-                    "options": {"temperature": 0.3, "num_predict": max_tokens}
-                },
-                timeout=self.timeout
-            )
-            if resp.status_code == 200:
-                output = resp.json().get('response', '').strip()
-                if output:
-                    self._cb.record_success()
-                    return output
-                self._cb.record_failure()
-            else:
-                logger.error(f"Ollama 问答失败: {resp.status_code} {resp.text[:200]}")
-                self._cb.record_failure()
-        except requests.Timeout:
-            logger.error(f"Ollama 问答超时 ({self.timeout}s)")
-            self._cb.record_failure()
-        except Exception as e:
-            logger.error(f"Ollama 问答异常: {e}")
-            self._cb.record_failure()
-        return None

fam-edge/src/fam_edge/qa.py

@@ -1,79 +1,136 @@
 """
-QA - 智能问答编排
+QA - 智能问答代理客户端（2026-08-23 问答链路整体抽离到独立 ai-gateway 服务后重写）
 
-问答链路（2026-08-23 重构）跟视频分析链路完全独立，不再复用视频分析用的
-Gemini flash / NVIDIA omni 模型：
-  role='text' 的适配器才参与问答，按 config.yaml 里 models 数组的出现顺序
-  依次尝试 chat()/chat_stream()，首个成功即用。role='vision' 的适配器
-  （视频分析用的 Gemini flash-latest、NVIDIA omni）完全不参与问答。
+原来的问答编排本体（NVIDIA 文字模型链 -> Gemini 非 flash 文字模型 -> 本地
+Ollama 兜底，含 key 轮换/熔断/降级）已经整个搬到独立的 ai-gateway 服务
+（OpenAI 兼容协议 /v1/chat/completions），跟视频分析业务解耦，别的项目也能
+直接用 OpenAI SDK 接入。fam-edge 这边现在只是一个转发客户端：调 ai-gateway，
+把它的 OpenAI 格式响应翻译回 fam-edge 原有的 (answer, provider) / 流式事件
+字典契约，上层 api_gateway.py 的 /api/edge/chat/ask(/stream) 端点和 fam-core
+的调用方完全不用改。
 
-当前链路（config.yaml 里对应 usage 标记，仅供人读，编排逻辑只看 role+顺序）:
-  1. NVIDIA 文字模型链（usage=qa_primary）：nemotron-3-ultra-550b-a55b ->
-     nemotron-3-super-120b-a12b -> gpt-oss-120b（同一个 NvidiaVisionAdapter
-     实例内部 model_chain 依次降级，见该适配器 chat()）
-  2. Gemini 非 flash 文字模型（usage=qa_primary）：gemini-pro-latest ->
-     gemini-2.5-pro
-  3. 本地 Ollama qwen2.5:7b（usage=qa_fallback，兜底）
+多 provider 之间失败降级（一个模型没吐出任何内容才换下一个、已经开始吐字后
+中途失败不悄悄换源）现在整个发生在 ai-gateway 内部，对这个客户端不可见——
+本客户端只会看到最终成功 provider 的分块流，或者全部失败时的空流。
 """
+import json
+import os
 from typing import Optional, Tuple
 
+import requests
+
 from .logger import setup_logger
 from .config_loader import load_config
-from .model_adapters.adapter_factory import build_adapters
 
 logger = setup_logger('fam-edge.qa')
 
 
 class QAOrchestrator:
     def __init__(self):
-        self.config = load_config()
-        all_adapters = build_adapters(self.config.get('models', []))
-        # 只有 role='text' 的适配器参与问答；按 config.yaml 里的出现顺序决定
-        # 降级顺序，不需要额外的 qa_order 配置——顺序即优先级。
-        self.adapters = [a for a in all_adapters if a.get_role() == 'text']
+        cfg = load_config().get('ai_gateway', {})
+        self.base_url = (cfg.get('base_url') or 'http://127.0.0.1:5100').rstrip('/')
+        self.token = self._resolve_token(cfg.get('token', ''))
+        self.timeout = cfg.get('timeout', 60)
+
+    def _resolve_token(self, raw: str) -> str:
+        if raw.startswith('${') and raw.endswith('}'):
+            return os.environ.get(raw[2:-1], '')
+        return raw
+
+    def _headers(self) -> dict:
+        headers = {"Content-Type": "application/json"}
+        if self.token:
+            headers["Authorization"] = f"Bearer {self.token}"
+        return headers
 
     def run_qa(self, prompt: str,
                max_tokens: int = 1024) -> Tuple[Optional[str], Optional[str]]:
-        """依次尝试各适配器的 chat()，返回 (answer, provider)。"""
-        for adapter in self.adapters:
-            try:
-                answer = adapter.chat(prompt, max_tokens=max_tokens)
-            except Exception as e:
-                logger.warning(f"QA {adapter.provider_name} 异常: {e}")
-                continue
-            if answer:
-                logger.info(f"QA 命中 provider={adapter.provider_name}")
-                return answer, adapter.provider_name
-            logger.info(f"QA {adapter.provider_name} 无返回，降级下一模型")
-        return None, None
+        """调 ai-gateway 非流式接口，返回 (answer, provider)。"""
+        try:
+            resp = requests.post(
+                f"{self.base_url}/v1/chat/completions",
+                headers=self._headers(),
+                json={"messages": [{"role": "user", "content": prompt}],
+                      "max_tokens": max_tokens, "stream": False},
+                timeout=self.timeout)
+        except Exception as e:
+            logger.warning(f"QA ai-gateway 请求异常: {e}")
+            return None, None
+        if resp.status_code != 200:
+            logger.warning(f"QA ai-gateway 返回 {resp.status_code}: {resp.text[:200]}")
+            return None, None
+        try:
+            data = resp.json()
+            answer = data["choices"][0]["message"]["content"]
+        except Exception as e:
+            logger.warning(f"QA ai-gateway 响应解析失败: {e}")
+            return None, None
+        if not answer:
+            return None, None
+        provider = data.get("provider")
+        logger.info(f"QA 命中 provider={provider}")
+        return answer, provider
 
     def run_qa_stream(self, prompt: str, max_tokens: int = 1024):
-        """流式版：依次尝试各适配器的 chat_stream()，yield 结构化事件字典。
+        """流式版：转发 ai-gateway 的 SSE 分块，翻译回原有事件字典契约。
 
         事件类型:
-          {"type":"provider_trying","provider":p}  开始尝试这个 provider
-          {"type":"chunk","provider":p,"text":t}    这个 provider 吐出的文本增量
-          {"type":"provider_failed","provider":p}   这个 provider 一个字都没吐出就失败，换下一个
-          {"type":"done","provider":p}              成功结束（这个 provider 至少吐出过一块）
-          {"type":"all_failed"}                     所有 provider 都失败
-
-        跟 run_qa 一样"仅在还没吐出任何文本时才允许换下一个 provider"——一旦
-        开始给用户看字了，中途失败就结束这次生成，不再悄悄换源接着写。
+          {"type":"provider_trying","provider":p}  流里第一次看到这个 provider
+          {"type":"chunk","provider":p,"text":t}    文本增量
+          {"type":"done","provider":p}              成功结束（至少吐出过一块）
+          {"type":"all_failed"}                     请求失败或没有任何文本产出
         """
-        for adapter in self.adapters:
-            yield {"type": "provider_trying", "provider": adapter.provider_name}
-            got_any = False
-            try:
-                for chunk in adapter.chat_stream(prompt, max_tokens=max_tokens):
-                    if chunk:
-                        got_any = True
-                        yield {"type": "chunk", "provider": adapter.provider_name, "text": chunk}
-            except Exception as e:
-                logger.warning(f"QA {adapter.provider_name} 流式异常: {e}")
-            if got_any:
-                logger.info(f"QA 流式命中 provider={adapter.provider_name}")
-                yield {"type": "done", "provider": adapter.provider_name}
-                return
-            logger.info(f"QA {adapter.provider_name} 流式无返回，降级下一模型")
-            yield {"type": "provider_failed", "provider": adapter.provider_name}
-        yield {"type": "all_failed"}
+        try:
+            resp = requests.post(
+                f"{self.base_url}/v1/chat/completions",
+                headers=self._headers(),
+                json={"messages": [{"role": "user", "content": prompt}],
+                      "max_tokens": max_tokens, "stream": True},
+                timeout=self.timeout, stream=True)
+        except Exception as e:
+            logger.warning(f"QA ai-gateway 流式请求异常: {e}")
+            yield {"type": "all_failed"}
+            return
+        if resp.status_code != 200:
+            logger.warning(f"QA ai-gateway 流式返回 {resp.status_code}: {resp.text[:200]}")
+            yield {"type": "all_failed"}
+            return
+        # 响应体固定 UTF-8，但 Content-Type 不一定带 charset，requests 会自己猜
+        # 编码——猜错就是中文乱码，强制指定跳过嗅探（同源坑见 gemini_adapter 历史修复）。
+        resp.encoding = 'utf-8'
+
+        current_provider = None
+        got_any = False
+        try:
+            for line in resp.iter_lines(decode_unicode=True):
+                if not line or not line.startswith('data: '):
+                    continue
+                payload = line[len('data: '):]
+                if payload == '[DONE]':
+                    break
+                try:
+                    chunk = json.loads(payload)
+                except ValueError:
+                    continue
+                if 'error' in chunk:
+                    logger.warning(f"QA ai-gateway 流式错误: {chunk['error']}")
+                    break
+                provider = chunk.get('provider')
+                if provider and provider != current_provider:
+                    current_provider = provider
+                    yield {"type": "provider_trying", "provider": provider}
+                choices = chunk.get('choices') or []
+                if not choices:
+                    continue
+                text = (choices[0].get('delta') or {}).get('content')
+                if text:
+                    got_any = True
+                    yield {"type": "chunk", "provider": current_provider, "text": text}
+        except Exception as e:
+            logger.warning(f"QA ai-gateway 流式读取异常: {e}")
+
+        if got_any:
+            logger.info(f"QA 流式命中 provider={current_provider}")
+            yield {"type": "done", "provider": current_provider}
+        else:
+            yield {"type": "all_failed"}

fam-edge/tests/test_gemini_adapter.py

@@ -108,17 +108,3 @@ def test_rotated_keys_single_key_never_errors():
     a = GeminiAdapter(_cfg())
     for _ in range(3):
         assert a._rotated_keys() == [(0, "key-primary")]
-
-
-def test_chat_timeout_defaults_short_not_shared_with_video_timeout():
-    """核心诉求: 问答是交互场景，不能沿用视频分析的 600s 超时——否则一个卡住
-    的 key/模型会让用户在聊天界面一直等，这正是"一直卡着"这个 bug 的根因。"""
-    a = GeminiAdapter(_cfg(timeout=600))
-    assert a.timeout == 600
-    assert a.chat_timeout == 20
-    assert a.chat_timeout != a.timeout
-
-
-def test_chat_timeout_configurable():
-    a = GeminiAdapter(_cfg(chat_timeout=8))
-    assert a.chat_timeout == 8

fam-edge/tests/test_ollama_adapter.py

@@ -1,73 +0,0 @@
-from fam_edge.model_adapters.ollama_adapter import OllamaAdapter
-
-
-def _cfg(**overrides):
-    base = {
-        "provider": "ollama",
-        "role": "text",
-        "model_name": "qwen2.5:7b",
-        "base_url": "http://localhost:11434",
-        "circuit_breaker": {"enabled": False},
-    }
-    base.update(overrides)
-    return base
-
-
-class _FakeResp:
-    def __init__(self, status_code=200, text=""):
-        self.status_code = status_code
-        self.text = text
-
-    def json(self):
-        return {"response": "ok"}
-
-
-def test_warm_up_success(monkeypatch):
-    calls = {}
-
-    def fake_post(url, json=None, timeout=None):
-        calls["url"] = url
-        calls["json"] = json
-        calls["timeout"] = timeout
-        return _FakeResp(200)
-    monkeypatch.setattr(
-        "fam_edge.model_adapters.ollama_adapter.requests.post", fake_post)
-    a = OllamaAdapter(_cfg())
-    assert a.warm_up() is True
-    assert calls["url"] == "http://localhost:11434/api/generate"
-    assert calls["json"]["model"] == "qwen2.5:7b"
-    # 只为触发加载，不需要真的生成长文本
-    assert calls["json"]["options"]["num_predict"] == 1
-
-
-def test_warm_up_http_error_returns_false(monkeypatch):
-    monkeypatch.setattr(
-        "fam_edge.model_adapters.ollama_adapter.requests.post",
-        lambda url, json=None, timeout=None: _FakeResp(500, "boom"))
-    a = OllamaAdapter(_cfg())
-    assert a.warm_up() is False
-
-
-def test_warm_up_exception_does_not_raise(monkeypatch):
-    """核心诉求: 预热失败(比如 Ollama 服务当时没起来)不能抛异常影响主服务
-    启动，只应该记警告日志、返回 False。"""
-    def raise_err(url, json=None, timeout=None):
-        raise ConnectionError("refused")
-    monkeypatch.setattr(
-        "fam_edge.model_adapters.ollama_adapter.requests.post", raise_err)
-    a = OllamaAdapter(_cfg())
-    assert a.warm_up() is False
-
-
-def test_warm_up_uses_generous_timeout_for_cold_start(monkeypatch):
-    """核心诉求: 冷启动实测能到 1-2 分钟，预热请求的超时不能沿用问答的短超时。"""
-    captured = {}
-
-    def fake_post(url, json=None, timeout=None):
-        captured["timeout"] = timeout
-        return _FakeResp(200)
-    monkeypatch.setattr(
-        "fam_edge.model_adapters.ollama_adapter.requests.post", fake_post)
-    a = OllamaAdapter(_cfg(timeout=20))  # chat() 用的短超时
-    a.warm_up()
-    assert captured["timeout"] >= 120

fam-edge/tests/test_qa.py

@@ -1,109 +1,199 @@
+import json
+
+import pytest
+
 from fam_edge.qa import QAOrchestrator
 
 
-class _FakeRoleAdapter:
-    """用于 __init__ 过滤逻辑测试，只需要 get_role()，不需要真的能 chat。"""
-    def __init__(self, provider_name, role):
-        self.provider_name = provider_name
-        self.role = role
-
-    def get_role(self):
-        return self.role
+def _orchestrator(monkeypatch, cfg=None, token='tok'):
+    ai_gateway_cfg = {"base_url": "http://127.0.0.1:5100", "token": token, "timeout": 5}
+    if cfg:
+        ai_gateway_cfg.update(cfg)
+    monkeypatch.setattr(
+        "fam_edge.qa.load_config", lambda: {"ai_gateway": ai_gateway_cfg})
+    return QAOrchestrator()
 
 
-def test_init_only_keeps_text_role_adapters_in_config_order(monkeypatch):
-    """核心诉求: 问答链路只用 role='text' 的适配器（跟视频分析用的
-    role='vision' 完全隔离），且顺序沿用 config.yaml 里 models 数组的出现
-    顺序，不需要额外的 qa_order 配置。"""
-    fake_adapters = [
-        _FakeRoleAdapter("gemini", "vision"),   # 视频分析用的 gemini-flash，不该出现
-        _FakeRoleAdapter("nvidia", "vision"),    # 视频分析用的 nvidia omni，不该出现
-        _FakeRoleAdapter("nvidia", "text"),      # 新的问答专用 nvidia 文字模型链
-        _FakeRoleAdapter("gemini", "text"),      # 新的问答专用 gemini 非 flash 文字模型
-        _FakeRoleAdapter("ollama", "text"),      # 本地兜底
-    ]
-    monkeypatch.setattr("fam_edge.qa.load_config", lambda: {"models": []})
-    monkeypatch.setattr("fam_edge.qa.build_adapters", lambda models: fake_adapters)
+def test_init_reads_base_url_and_token_from_config(monkeypatch):
+    qa = _orchestrator(monkeypatch, {"base_url": "http://example:5100/"}, token='secret')
+    assert qa.base_url == "http://example:5100"
+    assert qa.token == 'secret'
+
+
+def test_init_resolves_token_from_env_var(monkeypatch):
+    monkeypatch.setenv("MY_GATEWAY_TOKEN", "resolved-secret")
+    qa = _orchestrator(monkeypatch, token='${MY_GATEWAY_TOKEN}')
+    assert qa.token == 'resolved-secret'
+
+
+def test_init_defaults_base_url_when_unconfigured(monkeypatch):
+    monkeypatch.setattr("fam_edge.qa.load_config", lambda: {})
     qa = QAOrchestrator()
-    assert [a.role for a in qa.adapters] == ["text", "text", "text"]
-    assert len(qa.adapters) == 3
+    assert qa.base_url == "http://127.0.0.1:5100"
 
 
-class _FakeAdapter:
-    def __init__(self, provider_name, chunks=None, raises=False):
-        self.provider_name = provider_name
-        self._chunks = chunks or []
-        self._raises = raises
+class _FakeResp:
+    """模拟 requests.Response：非流式用 status_code/json()/text，
+    流式额外提供 iter_lines()（逐行 yield，跟真实 SSE 消费方式一致）。"""
 
-    def chat_stream(self, prompt, max_tokens=512):
-        if self._raises:
-            raise RuntimeError("boom")
-        for c in self._chunks:
-            yield c
+    def __init__(self, status_code=200, payload=None, text='', lines=None):
+        self.status_code = status_code
+        self._payload = payload
+        self.text = text
+        self._lines = lines if lines is not None else []
+        self.encoding = None
 
-    def chat(self, prompt, max_tokens=512):
-        return ''.join(self._chunks) or None
+    def json(self):
+        return self._payload
+
+    def iter_lines(self, decode_unicode=True):
+        for line in self._lines:
+            yield line
 
 
-def _orchestrator(adapters):
-    qa = QAOrchestrator.__new__(QAOrchestrator)  # 跳过 __init__（不需要真实 config/adapters）
-    qa.adapters = adapters
-    return qa
+def _capture_post(monkeypatch, resp):
+    calls = []
+
+    def fake_post(url, headers=None, json=None, timeout=None, stream=False):
+        calls.append({"url": url, "headers": headers, "json": json,
+                      "timeout": timeout, "stream": stream})
+        return resp
+
+    monkeypatch.setattr("fam_edge.qa.requests.post", fake_post)
+    return calls
 
 
-def test_run_qa_stream_first_provider_success():
-    qa = _orchestrator([_FakeAdapter("gemini", chunks=["你", "好"])])
+def test_run_qa_success(monkeypatch):
+    qa = _orchestrator(monkeypatch)
+    resp = _FakeResp(payload={"choices": [{"message": {"content": "你好"}}],
+                              "provider": "nvidia"})
+    calls = _capture_post(monkeypatch, resp)
+    answer, provider = qa.run_qa("hi", max_tokens=100)
+    assert answer == "你好"
+    assert provider == "nvidia"
+    assert calls[0]["json"] == {"messages": [{"role": "user", "content": "hi"}],
+                                "max_tokens": 100, "stream": False}
+    assert calls[0]["headers"]["Authorization"] == "Bearer tok"
+
+
+def test_run_qa_non_200_returns_none(monkeypatch):
+    qa = _orchestrator(monkeypatch)
+    resp = _FakeResp(status_code=503, text='{"error":{"message":"所有模型均不可用"}}')
+    _capture_post(monkeypatch, resp)
+    answer, provider = qa.run_qa("hi")
+    assert answer is None
+    assert provider is None
+
+
+def test_run_qa_empty_answer_returns_none(monkeypatch):
+    qa = _orchestrator(monkeypatch)
+    resp = _FakeResp(payload={"choices": [{"message": {"content": ""}}], "provider": "gemini"})
+    _capture_post(monkeypatch, resp)
+    answer, provider = qa.run_qa("hi")
+    assert answer is None
+    assert provider is None
+
+
+def test_run_qa_connection_error_returns_none(monkeypatch):
+    qa = _orchestrator(monkeypatch)
+
+    def _raise(*args, **kwargs):
+        raise ConnectionError("boom")
+
+    monkeypatch.setattr("fam_edge.qa.requests.post", _raise)
+    answer, provider = qa.run_qa("hi")
+    assert answer is None
+    assert provider is None
+
+
+def _sse_lines(events):
+    lines = []
+    for e in events:
+        lines.append(f"data: {json.dumps(e, ensure_ascii=False)}")
+    lines.append("data: [DONE]")
+    return lines
+
+
+def test_run_qa_stream_single_provider_success(monkeypatch):
+    qa = _orchestrator(monkeypatch)
+    lines = _sse_lines([
+        {"provider": "nvidia", "choices": [{"delta": {"content": "你"}}]},
+        {"provider": "nvidia", "choices": [{"delta": {"content": "好"}}]},
+        {"provider": "nvidia", "choices": [{"delta": {}}]},
+    ])
+    resp = _FakeResp(lines=lines)
+    _capture_post(monkeypatch, resp)
     events = list(qa.run_qa_stream("hi"))
     types = [e["type"] for e in events]
     assert types == ["provider_trying", "chunk", "chunk", "done"]
     assert events[1]["text"] == "你"
     assert events[2]["text"] == "好"
-    assert events[-1]["provider"] == "gemini"
-
-
-def test_run_qa_stream_falls_back_when_first_yields_nothing():
-    """核心诉求: 第一个 provider 一个字都没吐出来才允许换下一个——不是失败就切，
-    是"完全没有产出"才切。"""
-    qa = _orchestrator([
-        _FakeAdapter("gemini", chunks=[]),
-        _FakeAdapter("nvidia", chunks=["答案"]),
-    ])
-    events = list(qa.run_qa_stream("hi"))
-    types = [e["type"] for e in events]
-    assert types == ["provider_trying", "provider_failed", "provider_trying", "chunk", "done"]
     assert events[-1]["provider"] == "nvidia"
 
 
-def test_run_qa_stream_does_not_switch_after_partial_output():
-    """核心诉求: 已经开始吐字之后中途失败，不能悄悄换下一个 provider 接着写
-    （会出现两段风格/内容不连贯的回答拼在一起）——直接结束这次生成。"""
-    class _PartialThenRaise:
-        provider_name = "gemini"
-        def chat_stream(self, prompt, max_tokens=512):
-            yield "先吐"
-            raise RuntimeError("connection reset")
+def test_run_qa_stream_emits_provider_trying_once_per_change(monkeypatch):
+    """provider 字段没变化时不该重复吐 provider_trying。"""
+    qa = _orchestrator(monkeypatch)
+    lines = _sse_lines([
+        {"provider": "nvidia", "choices": [{"delta": {"content": "a"}}]},
+        {"provider": "nvidia", "choices": [{"delta": {"content": "b"}}]},
+    ])
+    resp = _FakeResp(lines=lines)
+    _capture_post(monkeypatch, resp)
+    events = list(qa.run_qa_stream("hi"))
+    trying = [e for e in events if e["type"] == "provider_trying"]
+    assert len(trying) == 1
+    assert trying[0]["provider"] == "nvidia"
 
-    qa = _orchestrator([_PartialThenRaise(), _FakeAdapter("nvidia", chunks=["不该被用到"])])
+
+def test_run_qa_stream_no_chunks_yields_all_failed(monkeypatch):
+    qa = _orchestrator(monkeypatch)
+    resp = _FakeResp(lines=["data: [DONE]"])
+    _capture_post(monkeypatch, resp)
+    events = list(qa.run_qa_stream("hi"))
+    assert events == [{"type": "all_failed"}]
+
+
+def test_run_qa_stream_non_200_yields_all_failed(monkeypatch):
+    qa = _orchestrator(monkeypatch)
+    resp = _FakeResp(status_code=503, text='{"error":{"message":"所有模型均不可用"}}')
+    _capture_post(monkeypatch, resp)
+    events = list(qa.run_qa_stream("hi"))
+    assert events == [{"type": "all_failed"}]
+
+
+def test_run_qa_stream_connection_error_yields_all_failed(monkeypatch):
+    qa = _orchestrator(monkeypatch)
+
+    def _raise(*args, **kwargs):
+        raise ConnectionError("boom")
+
+    monkeypatch.setattr("fam_edge.qa.requests.post", _raise)
+    events = list(qa.run_qa_stream("hi"))
+    assert events == [{"type": "all_failed"}]
+
+
+def test_run_qa_stream_error_chunk_stops_and_uses_partial_output(monkeypatch):
+    """已经吐出过内容后遇到错误块：按"至少吐出过一块就算 done"处理，不是 all_failed。"""
+    qa = _orchestrator(monkeypatch)
+    lines = [
+        f"data: {json.dumps({'provider': 'gemini', 'choices': [{'delta': {'content': '先吐'}}]}, ensure_ascii=False)}",
+        f"data: {json.dumps({'error': {'message': 'boom'}}, ensure_ascii=False)}",
+    ]
+    resp = _FakeResp(lines=lines)
+    _capture_post(monkeypatch, resp)
     events = list(qa.run_qa_stream("hi"))
     types = [e["type"] for e in events]
     assert types == ["provider_trying", "chunk", "done"]
-    assert events[1]["text"] == "先吐"
     assert events[-1]["provider"] == "gemini"
 
 
-def test_run_qa_stream_all_providers_fail():
-    qa = _orchestrator([
-        _FakeAdapter("gemini", chunks=[]),
-        _FakeAdapter("nvidia", chunks=[], raises=True),
-    ])
-    events = list(qa.run_qa_stream("hi"))
-    assert events[-1]["type"] == "all_failed"
-    assert "provider_failed" in [e["type"] for e in events]
-
-
-def test_run_qa_stream_exception_treated_as_no_output():
-    qa = _orchestrator([_FakeAdapter("gemini", raises=True), _FakeAdapter("nvidia", chunks=["ok"])])
-    events = list(qa.run_qa_stream("hi"))
-    assert events[0] == {"type": "provider_trying", "provider": "gemini"}
-    assert events[1] == {"type": "provider_failed", "provider": "gemini"}
-    assert events[-1]["provider"] == "nvidia"
+def test_run_qa_stream_sets_stream_true_and_utf8_encoding(monkeypatch):
+    qa = _orchestrator(monkeypatch)
+    resp = _FakeResp(lines=["data: [DONE]"])
+    calls = _capture_post(monkeypatch, resp)
+    list(qa.run_qa_stream("hi", max_tokens=222))
+    assert calls[0]["json"]["stream"] is True
+    assert calls[0]["json"]["max_tokens"] == 222
+    assert calls[0]["stream"] is True
+    assert resp.encoding == 'utf-8'