diff --git a/README.md b/README.md index 7d16e61..81c6d33 100644 --- a/README.md +++ b/README.md @@ -55,7 +55,7 @@ Audio profiles can reduce single-sample false positives with decision aggregatio ## API -`POST /api/v1/analysis/run` accepts `requestId`, `analysisType`, `profileCode`, `mediaUrl`, `options`, and `context`. +`POST /api/v1/analysis/run` accepts `requestId`, `analysisType`, `profileCode`, `mediaUrl`, `mediaUrls`, `options`, and `context`. The deployed endpoint is `http://192.168.28.10:14080`. It is called by the platform backend and requires a bearer token. @@ -69,6 +69,21 @@ Video requests may set `options.analysisMode` to one of: Existing workflows without this option are treated as `AUTO`. +Video requests may also set `options.detailedOutput`. It defaults to `false` and +returns a compact verdict without an event timeline. Set it to `true` only when +time-ranged event evidence and warnings are required. + +Long videos are sampled uniformly across their full duration. At most 30 frames are +sent in one model request so visual encoding stays within the H100 dynamic-memory +budget; the first and final state remain represented. + +Image requests use `analysisType=IMAGE_ANALYSIS` and profile +`common.image_analysis.v1`. Supply one to twelve scene image URLs in `mediaUrls`. +Set `options.prompt` to the complete analysis instruction. The model follows the +requested content and format without adding pass/fail fields. The response result +contains `result` (parsed JSON, array, number, or text) and `resultText` (the exact +model text). An optional `options.referenceImageUrl` is appended as the final image. + ## Qwen3.8 video models The model server runs the video models as separate SGLang services: @@ -86,6 +101,5 @@ docker compose -f deploy/compose.sglang.yaml up -d docker compose -f deploy/compose.sglang.yaml ps ``` -The analysis service calls their OpenAI-compatible APIs. When either SGLang endpoint -is unavailable, it falls back to the existing Ollama model for the corresponding -mode, provided `VISION_OLLAMA_FALLBACK_ENABLED=true`. +The analysis service calls only their OpenAI-compatible APIs. An unavailable SGLang +endpoint returns an explicit error; there is no Ollama fallback. diff --git a/app/config.py b/app/config.py index db83b61..38927f7 100644 --- a/app/config.py +++ b/app/config.py @@ -10,19 +10,17 @@ class Settings(BaseSettings): profiles_dir: Path = Path("/data/profiles") artifacts_dir: Path = Path("/data/artifacts") jobs_dir: Path = Path("/data/jobs") - ollama_base_url: str = "http://host.docker.internal:11434" - ollama_model: str = "qwen3-vl:32b" media_timeout_seconds: int = 120 - ollama_timeout_seconds: int = 600 sglang_fast_base_url: str = "" sglang_accurate_base_url: str = "" sglang_api_key: str = "" sglang_timeout_seconds: int = 600 - vision_ollama_fallback_enabled: bool = True max_audio_bytes: int = 100 * 1024 * 1024 + max_image_bytes: int = 30 * 1024 * 1024 max_video_bytes: int = 2 * 1024 * 1024 * 1024 video_sampling_frame_limit: int = 100 - max_video_context_tokens: int = 262144 + video_model_frame_limit: int = 30 + max_video_context_tokens: int = 65536 settings = Settings() diff --git a/app/main.py b/app/main.py index e98d8e3..eaece2b 100644 --- a/app/main.py +++ b/app/main.py @@ -8,6 +8,7 @@ from fastapi import Depends, FastAPI, Header, HTTPException from app.audio import classify_audio from app.config import settings +from app.image import analyze_images from app.media import download_media from app.profile_store import profile_store from app.schemas import AnalysisRequest, AnalysisResponse, AnalysisType @@ -47,23 +48,19 @@ async def health() -> dict: "fast": settings.sglang_fast_base_url, "accurate": settings.sglang_accurate_base_url, } - checks = [probe(f"{settings.ollama_base_url.rstrip('/')}/api/version")] configured_names = [name for name, url in sglang_urls.items() if url] - checks.extend( + checks = [ probe(f"{sglang_urls[name].rstrip('/')}/models") for name in configured_names - ) + ] statuses = await asyncio.gather(*checks) - ollama = statuses[0] - configured_statuses = dict(zip(configured_names, statuses[1:])) + configured_statuses = dict(zip(configured_names, statuses)) sglang = { name: configured_statuses.get(name, "NOT_CONFIGURED") for name in sglang_urls } return { "status": "UP", - "ollama": ollama, "sglang": sglang, - "visionModel": settings.ollama_model, "profiles": profile_store.status(), } @@ -76,6 +73,7 @@ async def health() -> dict: async def run_analysis(request: AnalysisRequest) -> AnalysisResponse: started = time.monotonic() media_path = None + media_paths = [] try: profile = profile_store.get(request.profileCode, request.analysisType.value) if request.analysisType == AnalysisType.AUDIO_CLASSIFICATION: @@ -84,10 +82,16 @@ async def run_analysis(request: AnalysisRequest) -> AnalysisResponse: ) result = classify_audio(profile, media_path) model = {"provider": "CMVR", "name": "mfcc-dtw-audio-fingerprint-v2"} - else: + elif request.analysisType == AnalysisType.VIDEO_ANALYSIS: media_path = await download_media( str(request.mediaUrl), ".video", settings.max_video_bytes ) + detailed_option = request.options.get("detailedOutput", False) + detailed_output = ( + detailed_option + if isinstance(detailed_option, bool) + else str(detailed_option).strip().lower() == "true" + ) result, duration, video_metadata = await analyze_video( profile, media_path, @@ -95,6 +99,7 @@ async def run_analysis(request: AnalysisRequest) -> AnalysisResponse: str(request.options.get("analysisMode", "AUTO")), request.options.get("tuning"), str(request.options.get("decisionPolicy", "FAIL_CLOSED")), + detailed_output, ) evidence = result.get("evidence") if not isinstance(evidence, dict): @@ -105,6 +110,7 @@ async def run_analysis(request: AnalysisRequest) -> AnalysisResponse: "sampledFrameCount": video_metadata["sampledFrameCount"], "sampleFps": video_metadata["effectiveSampleFps"], "samplingFrameLimit": video_metadata["samplingFrameLimit"], + "configuredSamplingFrameLimit": video_metadata["configuredSamplingFrameLimit"], "samplingCapped": video_metadata["samplingCapped"], "numCtx": video_metadata["numCtx"], "maximumWidth": video_metadata["maximumWidth"], @@ -115,6 +121,7 @@ async def run_analysis(request: AnalysisRequest) -> AnalysisResponse: result["fallback"] = video_metadata["fallback"] result["fallbackReason"] = video_metadata["fallbackReason"] result["effectiveTuning"] = video_metadata["tuning"] + result["detailedOutput"] = video_metadata["detailedOutput"] model = { "provider": video_metadata["provider"], "name": video_metadata["model"], @@ -125,6 +132,51 @@ async def run_analysis(request: AnalysisRequest) -> AnalysisResponse: "usedMode": video_metadata["mode"], "fallback": video_metadata["fallback"], } + else: + image_urls = [str(url) for url in request.mediaUrls] + if not image_urls and request.mediaUrl is not None: + image_urls = [str(request.mediaUrl)] + for image_url in image_urls: + media_paths.append(await download_media( + image_url, ".image", settings.max_image_bytes + )) + reference_path = None + reference_url = str(request.options.get("referenceImageUrl", "")).strip() + if reference_url: + reference_path = await download_media( + reference_url, ".reference-image", settings.max_image_bytes + ) + media_paths.append(reference_path) + prompt = str(request.options.get("prompt", "")).strip() + if not prompt: + target = str(request.options.get("targetDescription", "")).strip() + instruction = str(request.options.get("instruction", "")).strip() + prompt = "\n".join(part for part in (target, instruction) if part) + if prompt: + prompt += "\n只返回分析结果，不输出通过、未通过或置信度。" + result, image_metadata = await analyze_images( + profile, + media_paths[:-1] if reference_path is not None else media_paths, + reference_path, + prompt, + str(request.options.get("analysisMode", "AUTO")), + request.options.get("tuning"), + str(request.options.get("analysisMethod", "AUTO")), + ) + result["analysisMode"] = image_metadata["mode"] + result["fallback"] = image_metadata["fallback"] + result["imageCount"] = image_metadata["imageCount"] + result["analysisMethod"] = image_metadata["analysisMethod"] + model = { + "provider": image_metadata["provider"], + "name": image_metadata["model"], + "primaryProvider": image_metadata["provider"], + "primaryModel": image_metadata["model"], + "providerFallback": False, + "requestedMode": image_metadata["requestedMode"], + "usedMode": image_metadata["mode"], + "fallback": image_metadata["fallback"], + } return AnalysisResponse( requestId=request.requestId, analysisType=request.analysisType, @@ -143,3 +195,5 @@ async def run_analysis(request: AnalysisRequest) -> AnalysisResponse: finally: if media_path is not None: media_path.unlink(missing_ok=True) + for path in media_paths: + path.unlink(missing_ok=True) diff --git a/app/schemas.py b/app/schemas.py index bbc42c0..c115e4d 100644 --- a/app/schemas.py +++ b/app/schemas.py @@ -1,22 +1,30 @@ from enum import Enum from typing import Any -from pydantic import BaseModel, Field, HttpUrl +from pydantic import BaseModel, Field, HttpUrl, model_validator class AnalysisType(str, Enum): AUDIO_CLASSIFICATION = "AUDIO_CLASSIFICATION" VIDEO_ANALYSIS = "VIDEO_ANALYSIS" + IMAGE_ANALYSIS = "IMAGE_ANALYSIS" class AnalysisRequest(BaseModel): requestId: str = Field(min_length=1, max_length=200) analysisType: AnalysisType profileCode: str = Field(min_length=1, max_length=200) - mediaUrl: HttpUrl + mediaUrl: HttpUrl | None = None + mediaUrls: list[HttpUrl] = Field(default_factory=list, max_length=12) options: dict[str, Any] = Field(default_factory=dict) context: dict[str, Any] = Field(default_factory=dict) + @model_validator(mode="after") + def require_media(self): + if self.mediaUrl is None and not self.mediaUrls: + raise ValueError("mediaUrl or mediaUrls is required") + return self + class AnalysisResponse(BaseModel): requestId: str diff --git a/app/video.py b/app/video.py index 2d0a02d..fdeb298 100644 --- a/app/video.py +++ b/app/video.py @@ -58,6 +58,20 @@ VIDEO_RESULT_SCHEMA = { ], } +VIDEO_COMPACT_RESULT_SCHEMA = { + "type": "object", + "properties": { + "passed": VIDEO_RESULT_SCHEMA["properties"]["passed"], + "evidenceSufficient": {"type": "boolean"}, + "confidence": {"type": "number", "minimum": 0, "maximum": 1}, + "conclusion": {"type": "string"}, + "summary": {"type": "string"}, + }, + "required": [ + "passed", "evidenceSufficient", "confidence", "conclusion", "summary", + ], +} + VIDEO_TUNING_LIMITS = { "sampleFps": (0.25, 6.0), "maxWidth": (640, 1280), @@ -103,6 +117,13 @@ def _frame_sampling_plan( return count, prefix_count, prefix_fps, tail_time, capped +def _model_frame_limit() -> int: + return max( + 2, + min(settings.video_sampling_frame_limit, settings.video_model_frame_limit), + ) + + def _extract_frames( path: Path, sample_fps: float, maximum_frames: int, maximum_width: int ) -> tuple[list[Path], float, Path]: @@ -229,19 +250,19 @@ def _parse_json(content: str) -> dict[str, Any]: def _fallback_reason( - result: dict[str, Any], confidence_threshold: float + result: dict[str, Any], confidence_threshold: float, detailed_output: bool = True ) -> str | None: if result.get("structured") is False: return "FAST_RESULT_NOT_STRUCTURED" - required = ( + required = [ "passed", "evidenceSufficient", "confidence", "conclusion", "summary", - "events", - "warnings", - ) + ] + if detailed_output: + required.extend(("events", "warnings")) missing = [field for field in required if field not in result] if missing: return "FAST_RESULT_MISSING_FIELDS:" + ",".join(missing) @@ -249,7 +270,10 @@ def _fallback_reason( return "FAST_RESULT_EMPTY_CONCLUSION" if not isinstance(result.get("summary"), str) or not result["summary"].strip(): return "FAST_RESULT_EMPTY_SUMMARY" - if not isinstance(result.get("events"), list) or not isinstance(result.get("warnings"), list): + if detailed_output and ( + not isinstance(result.get("events"), list) + or not isinstance(result.get("warnings"), list) + ): return "FAST_RESULT_INVALID_COLLECTIONS" if result.get("evidenceSufficient") is not True: return "FAST_RESULT_INSUFFICIENT" @@ -475,6 +499,7 @@ def _normalize_decision( normalized["confidenceThreshold"] = confidence_threshold normalized.setdefault("conclusion", "通过" if passed else "未通过") normalized.setdefault("summary", normalized["conclusion"]) + normalized["result"] = normalized["summary"] normalized.pop("structured", None) return normalized @@ -482,7 +507,8 @@ def _normalize_decision( def _strategy(profile: Profile, name: str) -> dict[str, Any]: defaults = { "FAST": { - "model": profile.config.get("model", settings.ollama_model), + "model": profile.config.get("model", "Qwen/Qwen3.8-27B-FP8"), + "provider": "sglang", "sampleFps": 1.0, "maxWidth": 896, "maxOutputTokens": 512, @@ -491,7 +517,8 @@ def _strategy(profile: Profile, name: str) -> dict[str, Any]: "maxRetryOutputTokens": 4096, }, "ACCURATE": { - "model": profile.config.get("model", settings.ollama_model), + "model": profile.config.get("model", "Qwen/Qwen3.8-27B"), + "provider": "sglang", "sampleFps": 3.0, "maxWidth": 1280, "maxOutputTokens": 768, @@ -502,13 +529,12 @@ def _strategy(profile: Profile, name: str) -> dict[str, Any]: } configured = profile.config.get("strategies", {}).get(name.lower(), {}) strategy = {**defaults[name], **configured} - strategy["provider"] = str(strategy.get("provider", "ollama")).strip().lower() - if strategy["provider"] == "sglang": - strategy["baseUrl"] = ( - settings.sglang_fast_base_url - if name == "FAST" - else settings.sglang_accurate_base_url - ) + strategy["provider"] = str(strategy.get("provider", "sglang")).strip().lower() + strategy["baseUrl"] = ( + settings.sglang_fast_base_url + if name == "FAST" + else settings.sglang_accurate_base_url + ) return strategy @@ -525,31 +551,31 @@ def _sglang_payload(payload: dict[str, Any]) -> dict[str, Any]: for image in message.get("images") or [] ) options = payload.get("options") or {} - return { + result = { "model": payload["model"], "messages": [{"role": "user", "content": content}], "stream": False, "temperature": float(options.get("temperature", 0.1)), "max_tokens": int(options.get("num_predict", 512)), - "response_format": { - "type": "json_schema", - "json_schema": { - "name": "video_analysis", - "strict": True, - "schema": VIDEO_RESULT_SCHEMA, - }, - }, "chat_template_kwargs": { "enable_thinking": bool(payload.get("think", False)), "preserve_thinking": False, }, } + if payload.get("format"): + result["response_format"] = { + "type": "json_schema", + "json_schema": { + "name": "media_analysis", + "strict": True, + "schema": payload["format"], + }, + } + return result def _request_image_count(payload: dict[str, Any], provider: str) -> int: message = payload.get("messages", [{}])[0] - if provider == "ollama": - return len(message.get("images") or []) content = message.get("content") or [] return sum( 1 for item in content @@ -560,22 +586,17 @@ def _request_image_count(payload: dict[str, Any], provider: str) -> int: async def _request_vision_model( payload: dict[str, Any], empty_response_retries: int = 1, max_retry_output_tokens: int = 8192, - provider: str = "ollama", + provider: str = "sglang", base_url: str | None = None, ) -> tuple[str, dict[str, Any]]: - provider = str(provider or "ollama").strip().lower() - if provider not in {"ollama", "sglang"}: - raise ValueError(f"Unsupported vision model provider: {provider}") - if provider == "sglang" and not str(base_url or "").strip(): + provider = str(provider or "sglang").strip().lower() + if provider != "sglang": + raise ValueError("Only the SGLang vision model provider is supported") + if not str(base_url or "").strip(): raise VisionModelError("SGLang vision model endpoint is not configured") last_metadata: dict[str, Any] = {} - timeout_seconds = ( - settings.sglang_timeout_seconds - if provider == "sglang" - else settings.ollama_timeout_seconds - ) - timeout = httpx.Timeout(timeout_seconds, connect=5.0) + timeout = httpx.Timeout(settings.sglang_timeout_seconds, connect=5.0) async with httpx.AsyncClient(timeout=timeout) as client: for attempt in range(empty_response_retries + 1): request_payload = copy.deepcopy(payload) @@ -592,19 +613,13 @@ async def _request_vision_model( request_payload["think"] = False request_payload["messages"][0]["content"] += ( "\n\n/no_think\nReturn only the final JSON now. Do not output reasoning." + if request_payload.get("format") + else "\n\n/no_think\nReturn only the final answer requested by the user." ) - provider_payload = ( - _sglang_payload(request_payload) - if provider == "sglang" - else request_payload - ) - url = ( - f"{str(base_url).rstrip('/')}/chat/completions" - if provider == "sglang" - else f"{settings.ollama_base_url.rstrip('/')}/api/chat" - ) + provider_payload = _sglang_payload(request_payload) + url = f"{str(base_url).rstrip('/')}/chat/completions" headers = {} - if provider == "sglang" and settings.sglang_api_key: + if settings.sglang_api_key: headers["Authorization"] = f"Bearer {settings.sglang_api_key}" response = await client.post(url, json=provider_payload, headers=headers) try: @@ -619,34 +634,29 @@ async def _request_vision_model( f"detail={detail})" ) from exc body = response.json() - if provider == "sglang": - choices = body.get("choices") or [] - choice = choices[0] if choices else {} - message = choice.get("message") or {} - usage = body.get("usage") or {} - content = str(message.get("content") or "").strip() - metadata = { - "doneReason": choice.get("finish_reason"), - "promptEvalCount": usage.get("prompt_tokens"), - "evalCount": usage.get("completion_tokens"), - "totalDurationNs": None, - "thinkingLength": len(str(message.get("reasoning_content") or "")), - "attempt": attempt + 1, - "requestedOutputTokens": provider_payload.get("max_tokens"), - } - else: - message = body.get("message") or {} - content = str(message.get("content") or "").strip() - metadata = { - "doneReason": body.get("done_reason"), - "promptEvalCount": body.get("prompt_eval_count"), - "evalCount": body.get("eval_count"), - "totalDurationNs": body.get("total_duration"), - "thinkingLength": len(str(message.get("thinking") or "")), - "attempt": attempt + 1, - "requestedOutputTokens": request_payload.get("options", {}).get("num_predict"), - } + choices = body.get("choices") or [] + choice = choices[0] if choices else {} + message = choice.get("message") or {} + usage = body.get("usage") or {} + content = str(message.get("content") or "").strip() + metadata = { + "doneReason": choice.get("finish_reason"), + "promptEvalCount": usage.get("prompt_tokens"), + "evalCount": usage.get("completion_tokens"), + "totalDurationNs": None, + "thinkingLength": len(str(message.get("reasoning_content") or "")), + "attempt": attempt + 1, + "requestedOutputTokens": provider_payload.get("max_tokens"), + } metadata["provider"] = provider + truncated_structured_result = ( + bool(request_payload.get("format")) + and str(metadata.get("doneReason") or "").lower() == "length" + and _parse_json(content).get("structured") is False + ) + if content and truncated_structured_result and attempt < empty_response_retries: + last_metadata = metadata + continue if content: return content, metadata last_metadata = metadata @@ -664,12 +674,14 @@ async def _analyze_frames( extracted_width: int, prompt: str, tuning: dict[str, Any] | None = None, + detailed_output: bool = False, ) -> tuple[dict[str, Any], dict[str, Any]]: strategy = _apply_tuning(_strategy(profile, strategy_name), tuning or {}) + model_frame_limit = _model_frame_limit() desired_count, _, _, _, sampling_capped = _frame_sampling_plan( duration, float(strategy["sampleFps"]), - settings.video_sampling_frame_limit, + model_frame_limit, ) selected = _select_frames(frames, desired_count) maximum_width = max(320, int(strategy["maxWidth"])) @@ -678,10 +690,22 @@ async def _analyze_frames( f"\n\nThe video duration is {duration:.3f} seconds. " f"The following {len(selected)} images are ordered frames sampled across the video." ) + if detailed_output: + strategy_prompt += ( + "\n\nDetailed output is enabled. Include a concise event timeline and warnings." + ) + else: + strategy_prompt += ( + "\n\nCompact output is enabled. Return only passed, evidenceSufficient, " + "confidence, conclusion, and a short summary. Do not return events, warnings, " + "frame-by-frame descriptions, or repeated evidence. Keep conclusion under 20 " + "Chinese characters and summary under 100 Chinese characters." + ) think = bool(strategy.get("think", profile.config.get("think", False))) if not think: strategy_prompt += "\n\n/no_think\nReturn only the requested JSON object without reasoning." - output_tokens = int(strategy["maxOutputTokens"]) + token_key = "detailedMaxOutputTokens" if detailed_output else "compactMaxOutputTokens" + output_tokens = int(strategy.get(token_key, strategy["maxOutputTokens"])) context_tokens = _video_context_size( int(strategy["numCtx"]), len(selected), output_tokens ) @@ -689,7 +713,7 @@ async def _analyze_frames( "model": strategy["model"], "stream": False, "think": think, - "format": VIDEO_RESULT_SCHEMA, + "format": VIDEO_RESULT_SCHEMA if detailed_output else VIDEO_COMPACT_RESULT_SCHEMA, "keep_alive": str(strategy.get("keepAlive", "30m")), "messages": [{"role": "user", "content": strategy_prompt, "images": images}], "options": { @@ -698,47 +722,29 @@ async def _analyze_frames( "num_ctx": context_tokens, }, } - provider = str(strategy.get("provider", "ollama")) - actual_model = str(strategy["model"]) - provider_fallback = False - try: - content, metrics = await _request_vision_model( - payload, - int(strategy.get("emptyResponseRetries", 0)), - int(strategy.get("maxRetryOutputTokens", 8192)), - provider, - strategy.get("baseUrl"), - ) - except (VisionModelError, httpx.HTTPError): - fallback_model = str(strategy.get("fallbackModel") or "").strip() - if ( - provider != "sglang" - or not settings.vision_ollama_fallback_enabled - or not fallback_model - ): - raise - fallback_payload = copy.deepcopy(payload) - fallback_payload["model"] = fallback_model - content, metrics = await _request_vision_model( - fallback_payload, - int(strategy.get("emptyResponseRetries", 0)), - int(strategy.get("maxRetryOutputTokens", 8192)), - ) - provider_fallback = True - actual_model = fallback_model + provider = str(strategy.get("provider", "sglang")) + content, metrics = await _request_vision_model( + payload, + int(strategy.get("emptyResponseRetries", 0)), + int(strategy.get("maxRetryOutputTokens", 8192)), + provider, + strategy.get("baseUrl"), + ) metadata = { "mode": strategy_name, - "model": actual_model, + "model": str(strategy["model"]), "provider": metrics.get("provider", provider), "primaryProvider": provider, "primaryModel": strategy["model"], - "providerFallback": provider_fallback, + "providerFallback": False, "sampledFrameCount": len(selected), "maximumWidth": maximum_width, - "effectiveSampleFps": float(strategy["sampleFps"]), - "samplingFrameLimit": settings.video_sampling_frame_limit, + "effectiveSampleFps": round(max(0.0, (len(selected) - 1) / duration), 4), + "samplingFrameLimit": model_frame_limit, + "configuredSamplingFrameLimit": settings.video_sampling_frame_limit, "samplingCapped": sampling_capped, "numCtx": context_tokens, + "detailedOutput": detailed_output, **metrics, } return _parse_json(content), metadata @@ -751,6 +757,7 @@ async def analyze_video( analysis_mode: str = "AUTO", tuning: dict[str, Any] | None = None, decision_policy: str = "FAIL_CLOSED", + detailed_output: bool = False, ) -> tuple[dict[str, Any], float, dict[str, Any]]: requested_mode = str(analysis_mode or "AUTO").strip().upper() if requested_mode not in {"AUTO", "FAST", "ACCURATE"}: @@ -767,10 +774,11 @@ async def analyze_video( extraction_strategy = fast if requested_mode == "FAST" else accurate extraction_sample_fps = float(extraction_strategy["sampleFps"]) extraction_width = max(320, int(extraction_strategy["maxWidth"])) + model_frame_limit = _model_frame_limit() frames, duration, frame_dir = _extract_frames( media_path, extraction_sample_fps, - settings.video_sampling_frame_limit, + model_frame_limit, extraction_width, ) try: @@ -807,6 +815,7 @@ async def analyze_video( extraction_width, focused_prompt, effective_tuning, + detailed_output, ) verified = _normalize_decision( verified, confidence_threshold, extra_instruction @@ -833,7 +842,7 @@ async def analyze_video( if requested_mode in {"FAST", "ACCURATE"}: result, metadata = await _analyze_frames( profile, requested_mode, frames, duration, extraction_width, prompt, - effective_tuning, + effective_tuning, detailed_output, ) if requested_mode == "FAST": result, metadata = await finalize_fast_result( @@ -857,9 +866,11 @@ async def analyze_video( try: result, metadata = await _analyze_frames( profile, "FAST", frames, duration, extraction_width, prompt, - effective_tuning, + effective_tuning, detailed_output, + ) + fallback_reason = _fallback_reason( + result, confidence_threshold, detailed_output ) - fallback_reason = _fallback_reason(result, confidence_threshold) if fallback_reason is None or not fallback_to_accurate: result, metadata = await finalize_fast_result( result, metadata, "AUTO", fallback_reason @@ -870,7 +881,7 @@ async def analyze_video( result, metadata = await _analyze_frames( profile, "ACCURATE", frames, duration, extraction_width, prompt, - effective_tuning, + effective_tuning, detailed_output, ) result = _normalize_decision(result, confidence_threshold, extra_instruction) metadata.update( diff --git a/data/profiles/aima/power-video/v1/profile.json b/data/profiles/aima/power-video/v1/profile.json index 250f40c..2c918a3 100644 --- a/data/profiles/aima/power-video/v1/profile.json +++ b/data/profiles/aima/power-video/v1/profile.json @@ -2,7 +2,6 @@ "code": "aima.power_video.v1", "name": "爱玛测试视频分析", "analysisType": "VIDEO_ANALYSIS", - "model": "qwen3-vl:32b", "promptFile": "prompt.txt", "think": false, "temperature": 0.1, @@ -10,22 +9,24 @@ "fast": { "provider": "sglang", "model": "Qwen/Qwen3.8-27B-FP8", - "fallbackModel": "qwen3-vl:8b-instruct", "sampleFps": 1.0, "maxWidth": 896, "maxOutputTokens": 512, + "compactMaxOutputTokens": 256, + "detailedMaxOutputTokens": 1024, "numCtx": 16384, - "emptyResponseRetries": 0, + "emptyResponseRetries": 1, "maxRetryOutputTokens": 4096, "keepAlive": "30m" }, "accurate": { "provider": "sglang", "model": "Qwen/Qwen3.8-27B", - "fallbackModel": "qwen3-vl:32b", "sampleFps": 3.0, "maxWidth": 1120, "maxOutputTokens": 768, + "compactMaxOutputTokens": 384, + "detailedMaxOutputTokens": 1536, "numCtx": 32768, "emptyResponseRetries": 2, "maxRetryOutputTokens": 8192, diff --git a/deploy/.env.example b/deploy/.env.example index 4b85cc9..fb0b001 100644 --- a/deploy/.env.example +++ b/deploy/.env.example @@ -1,12 +1,11 @@ API_KEY=replace-with-a-strong-random-token MEDIA_TIMEOUT_SECONDS=120 -OLLAMA_TIMEOUT_SECONDS=600 SGLANG_FAST_BASE_URL=http://192.168.28.10:14081/v1 SGLANG_ACCURATE_BASE_URL=http://192.168.28.10:14082/v1 SGLANG_API_KEY= SGLANG_TIMEOUT_SECONDS=600 -VISION_OLLAMA_FALLBACK_ENABLED=true MAX_AUDIO_BYTES=104857600 MAX_VIDEO_BYTES=2147483648 VIDEO_SAMPLING_FRAME_LIMIT=100 -MAX_VIDEO_CONTEXT_TOKENS=262144 +VIDEO_MODEL_FRAME_LIMIT=30 +MAX_VIDEO_CONTEXT_TOKENS=65536 diff --git a/deploy/compose.sglang.yaml b/deploy/compose.sglang.yaml index dcc472d..d17c0ee 100644 --- a/deploy/compose.sglang.yaml +++ b/deploy/compose.sglang.yaml @@ -32,11 +32,11 @@ services: - --tp-size - "1" - --context-length - - "262144" + - "65536" - --kv-cache-dtype - fp8_e4m3 - --mem-fraction-static - - "0.88" + - "0.80" - --reasoning-parser - qwen3 @@ -73,10 +73,10 @@ services: - --tp-size - "1" - --context-length - - "262144" + - "65536" - --kv-cache-dtype - fp8_e4m3 - --mem-fraction-static - - "0.92" + - "0.84" - --reasoning-parser - qwen3 diff --git a/deploy/compose.yaml b/deploy/compose.yaml index 2d7f470..7e0cbde 100644 --- a/deploy/compose.yaml +++ b/deploy/compose.yaml @@ -3,7 +3,7 @@ name: cmvr-ai-analysis-qwen38 services: analysis-service: container_name: cmvr-ai-analysis - image: cmvr-ai-analysis:1.2.0-qwen38 + image: cmvr-ai-analysis:1.5.4-multi-match build: context: .. dockerfile: Dockerfile @@ -17,12 +17,11 @@ services: PROFILES_DIR: /data/profiles ARTIFACTS_DIR: /data/artifacts JOBS_DIR: /data/jobs - OLLAMA_BASE_URL: http://host.docker.internal:11434 - OLLAMA_MODEL: qwen3-vl:32b SGLANG_FAST_BASE_URL: http://192.168.28.10:14081/v1 SGLANG_ACCURATE_BASE_URL: http://192.168.28.10:14082/v1 SGLANG_TIMEOUT_SECONDS: "600" - VISION_OLLAMA_FALLBACK_ENABLED: "true" + VIDEO_MODEL_FRAME_LIMIT: "30" + MAX_VIDEO_CONTEXT_TOKENS: "65536" extra_hosts: - "host.docker.internal:host-gateway" volumes: diff --git a/requirements.txt b/requirements.txt index 645a1a1..6d9dc25 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,7 @@ fastapi==0.116.1 httpx==0.28.1 numpy==2.2.6 +opencv-python-headless==4.12.0.88 pydantic-settings==2.10.1 python-multipart==0.0.20 scipy==1.15.3 diff --git a/tests/test_video.py b/tests/test_video.py index 3778761..5be6257 100644 --- a/tests/test_video.py +++ b/tests/test_video.py @@ -37,68 +37,85 @@ class _FakeClient: def test_vision_request_retries_empty_content(monkeypatch): _FakeClient.responses = [ { - "done_reason": "length", - "eval_count": 1024, - "message": {"content": "", "thinking": "reasoning"}, + "choices": [{ + "finish_reason": "length", + "message": {"content": "", "reasoning_content": "reasoning"}, + }], + "usage": {"completion_tokens": 1024}, }, { - "done_reason": "stop", - "eval_count": 20, - "message": {"content": '{"passed": true}', "thinking": ""}, + "choices": [{ + "finish_reason": "stop", + "message": {"content": '{"passed": true}', "reasoning_content": ""}, + }], + "usage": {"completion_tokens": 20}, }, ] _FakeClient.payloads = [] monkeypatch.setattr(httpx, "AsyncClient", _FakeClient) payload = { + "model": "Qwen/Qwen3.8-27B-FP8", "think": False, "messages": [{"role": "user", "content": "Analyze"}], "options": {"num_predict": 1024}, } - content, metadata = asyncio.run(video._request_vision_model(payload)) + content, metadata = asyncio.run(video._request_vision_model( + payload, base_url="http://sglang-fast:30000/v1" + )) assert content == '{"passed": true}' assert metadata["doneReason"] == "stop" assert len(_FakeClient.payloads) == 2 - assert _FakeClient.payloads[0]["think"] is False - assert _FakeClient.payloads[1]["options"]["num_predict"] == 4096 - assert _FakeClient.payloads[1]["think"] is False - assert "/no_think" in _FakeClient.payloads[1]["messages"][0]["content"] + assert _FakeClient.payloads[0]["chat_template_kwargs"]["enable_thinking"] is False + assert _FakeClient.payloads[1]["max_tokens"] == 4096 + assert _FakeClient.payloads[1]["chat_template_kwargs"]["enable_thinking"] is False + assert "/no_think" in _FakeClient.payloads[1]["messages"][0]["content"][0]["text"] assert metadata["requestedOutputTokens"] == 4096 def test_vision_request_escalates_budget_after_repeated_length_cutoff(monkeypatch): _FakeClient.responses = [ { - "done_reason": "length", - "eval_count": 768, - "message": {"content": "", "thinking": "first reasoning"}, + "choices": [{ + "finish_reason": "length", + "message": {"content": "", "reasoning_content": "first reasoning"}, + }], + "usage": {"completion_tokens": 768}, }, { - "done_reason": "length", - "eval_count": 4096, - "message": {"content": "", "thinking": "more reasoning"}, + "choices": [{ + "finish_reason": "length", + "message": {"content": "", "reasoning_content": "more reasoning"}, + }], + "usage": {"completion_tokens": 4096}, }, { - "done_reason": "stop", - "eval_count": 30, - "message": {"content": '{"passed": true}', "thinking": ""}, + "choices": [{ + "finish_reason": "stop", + "message": {"content": '{"passed": true}', "reasoning_content": ""}, + }], + "usage": {"completion_tokens": 30}, }, ] _FakeClient.payloads = [] monkeypatch.setattr(httpx, "AsyncClient", _FakeClient) payload = { + "model": "Qwen/Qwen3.8-27B-FP8", "think": False, "messages": [{"role": "user", "content": "Analyze /no_think"}], "options": {"num_predict": 768}, } content, metadata = asyncio.run( - video._request_vision_model(payload, empty_response_retries=2) + video._request_vision_model( + payload, empty_response_retries=2, + base_url="http://sglang-fast:30000/v1", + ) ) assert content == '{"passed": true}' - assert [_payload["options"]["num_predict"] for _payload in _FakeClient.payloads] == [ + assert [_payload["max_tokens"] for _payload in _FakeClient.payloads] == [ 768, 4096, 8192, @@ -106,6 +123,45 @@ def test_vision_request_escalates_budget_after_repeated_length_cutoff(monkeypatc assert metadata["attempt"] == 3 +def test_vision_request_retries_truncated_structured_content(monkeypatch): + _FakeClient.responses = [ + { + "choices": [{ + "finish_reason": "length", + "message": {"content": '{"passed": true, "events": [', "reasoning_content": ""}, + }], + "usage": {"completion_tokens": 512}, + }, + { + "choices": [{ + "finish_reason": "stop", + "message": {"content": '{"passed": true}', "reasoning_content": ""}, + }], + "usage": {"completion_tokens": 18}, + }, + ] + _FakeClient.payloads = [] + monkeypatch.setattr(httpx, "AsyncClient", _FakeClient) + payload = { + "model": "Qwen/Qwen3.8-27B-FP8", + "think": False, + "format": video.VIDEO_RESULT_SCHEMA, + "messages": [{"role": "user", "content": "Analyze"}], + "options": {"num_predict": 512}, + } + + content, metadata = asyncio.run(video._request_vision_model( + payload, + empty_response_retries=1, + base_url="http://sglang-fast:30000/v1", + )) + + assert content == '{"passed": true}' + assert len(_FakeClient.payloads) == 2 + assert _FakeClient.payloads[1]["max_tokens"] == 4096 + assert metadata["attempt"] == 2 + + def test_sglang_payload_uses_openai_multimodal_format(): payload = video._sglang_payload( { @@ -115,6 +171,7 @@ def test_sglang_payload_uses_openai_multimodal_format(): {"role": "user", "content": "Analyze", "images": ["abc", "def"]} ], "options": {"temperature": 0.1, "num_predict": 512}, + "format": video.VIDEO_RESULT_SCHEMA, } ) @@ -191,6 +248,21 @@ def test_fast_result_quality_controls_accurate_fallback(): ) +def test_compact_result_does_not_require_event_evidence_arrays(): + compact = { + "passed": True, + "evidenceSufficient": True, + "confidence": 0.9, + "conclusion": "通过", + "summary": "状态变化符合标准", + } + + assert video._fallback_reason(compact, 0.75, detailed_output=False) is None + assert video._fallback_reason(compact, 0.75, detailed_output=True).startswith( + "FAST_RESULT_MISSING_FIELDS:" + ) + + def test_decision_is_fail_closed_when_evidence_is_insufficient(): result = video._normalize_decision( { @@ -208,6 +280,7 @@ def test_decision_is_fail_closed_when_evidence_is_insufficient(): assert result["passed"] is False assert result["decisionReason"] == "INSUFFICIENT_EVIDENCE" assert result["conclusion"].startswith("未通过：视频证据不足") + assert result["result"] == "没有拍到完整过程" def test_negative_acceptance_criterion_corrects_inconsistent_model_boolean(): @@ -315,10 +388,21 @@ def test_frame_sampling_always_includes_video_tail(): assert capped is True -def test_video_context_expands_with_frame_count(): +def test_video_context_is_capped_to_safe_model_budget(): assert video._video_context_size(16384, 7, 512) == 16384 + assert video._video_context_size(16384, 18, 1536) == 32768 assert video._video_context_size(16384, 27, 512) == 65536 - assert video._video_context_size(32768, 100, 768) == 262144 + assert video._video_context_size(32768, 100, 768) == 65536 + + +def test_model_frame_limit_preserves_global_limit(monkeypatch): + monkeypatch.setattr(video.settings, "video_sampling_frame_limit", 100) + monkeypatch.setattr(video.settings, "video_model_frame_limit", 30) + assert video._model_frame_limit() == 30 + + count, _, _, _, capped = video._frame_sampling_plan(20.0, 3.0, 30) + assert count == 30 + assert capped is True def test_auto_mode_falls_back_to_accurate_result(monkeypatch, tmp_path):