diff --git a/fam-edge/config/config.yaml b/fam-edge/config/config.yaml index b7514b4..e839d04 100644 --- a/fam-edge/config/config.yaml +++ b/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 diff --git a/fam-edge/src/fam_edge/app.py b/fam-edge/src/fam_edge/app.py index 9b925b4..3d4136f 100644 --- a/fam-edge/src/fam_edge/app.py +++ b/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() diff --git a/fam-edge/src/fam_edge/model_adapters/__init__.py b/fam-edge/src/fam_edge/model_adapters/__init__.py index 57bc87a..bb800b1 100644 --- a/fam-edge/src/fam_edge/model_adapters/__init__.py +++ b/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", diff --git a/fam-edge/src/fam_edge/model_adapters/adapter_factory.py b/fam-edge/src/fam_edge/model_adapters/adapter_factory.py index 7134d01..993aa87 100644 --- a/fam-edge/src/fam_edge/model_adapters/adapter_factory.py +++ b/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, } diff --git a/fam-edge/src/fam_edge/model_adapters/base_adapter.py b/fam-edge/src/fam_edge/model_adapters/base_adapter.py index ee00222..379a543 100644 --- a/fam-edge/src/fam_edge/model_adapters/base_adapter.py +++ b/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 diff --git a/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py b/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py index 2efb7da..f6e0912 100644 --- a/fam-edge/src/fam_edge/model_adapters/gemini_adapter.py +++ b/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 diff --git a/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py b/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py index 9166c97..0cb03ef 100644 --- a/fam-edge/src/fam_edge/model_adapters/nvidia_adapter.py +++ b/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 diff --git a/fam-edge/src/fam_edge/model_adapters/ollama_adapter.py b/fam-edge/src/fam_edge/model_adapters/ollama_adapter.py deleted file mode 100644 index 7c3f1d3..0000000 --- a/fam-edge/src/fam_edge/model_adapters/ollama_adapter.py +++ /dev/null @@ -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 diff --git a/fam-edge/src/fam_edge/qa.py b/fam-edge/src/fam_edge/qa.py index 49aa535..4533b2e 100644 --- a/fam-edge/src/fam_edge/qa.py +++ b/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"} diff --git a/fam-edge/tests/test_gemini_adapter.py b/fam-edge/tests/test_gemini_adapter.py index 6c86410..60171e9 100644 --- a/fam-edge/tests/test_gemini_adapter.py +++ b/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 diff --git a/fam-edge/tests/test_ollama_adapter.py b/fam-edge/tests/test_ollama_adapter.py deleted file mode 100644 index 52d5a92..0000000 --- a/fam-edge/tests/test_ollama_adapter.py +++ /dev/null @@ -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 diff --git a/fam-edge/tests/test_qa.py b/fam-edge/tests/test_qa.py index 28b9923..e2d43de 100644 --- a/fam-edge/tests/test_qa.py +++ b/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'