class CgVisionService:
def __init__(self, owner: Any) -> None:
self.owner = owner
# ---- configuration / transport -------------------------------------
@property
def config(self) -> Any:
return self.owner.config
def _require_configured(self) -> None:
if not self.config.vision_configured:
raise CgVisionError(
VISION_NOT_CONFIGURED_CODE,
"视觉识别未配置：请设置 CG_RAG_VISION_ENDPOINT 与 CG_RAG_VISION_MODEL。",
)
def _headers(self) -> dict[str, str]:
headers = {"Content-Type": "application/json"}
key = self.config.effective_vision_api_key
if key:
headers["Authorization"] = f"Bearer {key}"
return headers
def _decode_images(self, request: CgVisionAnswerRequest) -> tuple[VisionImagePart, ...]:
max_images = int(self.config.vision_max_images)
if len(request.images) > max_images:
raise CgVisionError(
VISION_INVALID_IMAGE_CODE,
f"最多支持 {max_images} 张图片，本次收到 {len(request.images)} 张。",
)
max_bytes = int(self.config.vision_max_image_bytes)
parts: list[VisionImagePart] = []
for ordinal, image in enumerate(request.images, start=1):
try:
mime_type, payload = image.resolved()
except ValueError as exc:
raise CgVisionError(VISION_INVALID_IMAGE_CODE, str(exc)) from None
try:
data = base64.b64decode(payload, validate=True)
except (binascii.Error, ValueError):
raise CgVisionError(
VISION_INVALID_IMAGE_CODE,
f"第 {ordinal} 张图片不是合法的 base64 数据。",
) from None
if not data:
raise CgVisionError(
VISION_INVALID_IMAGE_CODE, f"第 {ordinal} 张图片为空。"
)
if len(data) > max_bytes:
raise CgVisionError(
VISION_INVALID_IMAGE_CODE,
f"第 {ordinal} 张图片为 {len(data)} 字节，超过上限 {max_bytes} 字节。",
)
try:
normalized = normalize_vision_image_bytes(
data,
declared_mime_type=mime_type,
max_pixels=int(self.config.vision_max_pixels),
max_dimension=int(self.config.vision_max_dimension),
normalized_max_dimension=int(self.config.vision_normalized_max_dimension),
jpeg_quality=int(self.config.vision_jpeg_quality),
preprocessed=bool(image.preprocessed),
)
except ImageValidationError as exc:
raise CgVisionError(VISION_INVALID_IMAGE_CODE, str(exc)) from None
parts.append(
VisionImagePart(
attachment_id=f"inline-{ordinal}",
ordinal=ordinal,
mime_type=normalized.mime_type,
data=normalized.content,
width=normalized.width,
height=normalized.height,
source_turn_index=image.source_turn_index,
)
)
return tuple(parts)
def _vision_request(self, request: CgVisionAnswerRequest) -> VisionModelRequest:
return VisionModelRequest(
profile=load_construction_hazard_profile(),
original_text=request.question,
images=self._decode_images(request),
task_mode=request.task_mode,
history=tuple(VisionHistoryTurn(query=turn.query, answer=turn.answer, turn_index=turn.turn_index) for turn in request.history),
)
def _thinking_enabled(self, request: CgVisionAnswerRequest) -> bool:
if request.enable_thinking is None:
return bool(self.config.vision_enable_thinking)
return bool(request.enable_thinking)
def _post(
self,
payload: dict[str, Any],
*,
streaming: bool,
on_token: Callable[[str, str], None] | None = None,
final_answer_guard: FinalAnswerStreamObserver | None = None,
cancel_token: Any = None,
) -> dict[str, Any]:
attempts = VISION_UPSTREAM_RETRY_COUNT + 1
last_exc: BaseException | None = None
# Streaming: retry only before the first forwarded token. Replaying
# after tokens were already sent would duplicate text on the client (B8).
streamed_any_token = False
def _tracking_on_token(channel: str, text: str) -> None:
nonlocal streamed_any_token
if text:
streamed_any_token = True
if on_token is not None:
on_token(channel, text)
for attempt in range(1, attempts + 1):
_raise_if_cancelled(cancel_token)
try:
with self.owner.capacity_limiters.generation.acquire(
cancel_token=cancel_token
):
if streaming:
return self._post_streaming(
payload,
on_token=_tracking_on_token,
final_answer_guard=final_answer_guard,
cancel_token=cancel_token,
)
response = self.owner.generation_session.post(
self.config.vision_endpoint,
headers=self._headers(),
json=payload,
timeout=self.config.timeout_seconds,
)
response.raise_for_status()
body = response.json()
except requests.RequestException as exc:
last_exc = exc
# Tokens already handed to the caller are final. Retrying would
# replay the model call and duplicate them. Validation failures
# are not transient upstream outages either.
if streaming and streamed_any_token:
raise _transport_vision_error(exc) from exc
if not _transport_is_retryable(exc) or attempt >= attempts:
raise _transport_vision_error(exc) from exc
time.sleep(VISION_UPSTREAM_RETRY_SLEEP_SECONDS * attempt)
continue
if not isinstance(body, dict):
raise CgVisionError(
VISION_RESPONSE_INVALID_CODE, "视觉模型返回的响应体不是对象。"
)
return body
raise CgVisionError(
VISION_UPSTREAM_ERROR_CODE, "视觉模型调用失败。"
) from last_exc
def _post_streaming(
self,
payload: dict[str, Any],
*,
on_token: Callable[[str, str], None] | None,
final_answer_guard: FinalAnswerStreamObserver | None = None,
cancel_token: Any = None,
) -> dict[str, Any]:
streaming_payload = {**payload, "stream": True}
content_parts: list[str] = []
reasoning_parts: list[str] = []
# The agent decides to search by emitting tool calls, which arrive as
# deltas. Dropping them makes every round look like the model declined
# to retrieve, so the answer silently loses its citations.
tool_calls = StreamingToolCallAccumulator()
usage: dict[str, Any] = {}
response = self.owner.generation_session.post(
self.config.vision_endpoint,
headers=self._headers(),
json=streaming_payload,
timeout=self.config.timeout_seconds,
stream=True,
)
try:
response.raise_for_status()
# parse_chat_completion_sse already flattens each chunk into
# {"content"|"reasoning_content"|"usage"|"done"}; it does not hand
# back raw OpenAI chunks, so there is no choices/delta to walk.
for event in self.owner._parse_chat_completion_sse(
response.iter_lines(decode_unicode=True)
):
_raise_if_cancelled(cancel_token)
if not isinstance(event, dict):
continue
if isinstance(event.get("usage"), dict):
usage = event["usage"]
tool_calls.push(event)
text = event.get("content")
if isinstance(text, str) and text:
content_parts.append(text)
visible_text = (
final_answer_guard.push(text, text)
if final_answer_guard is not None
else text
)
if on_token is not None and visible_text:
on_token("answer", visible_text)
reasoning = event.get("reasoning_content")
if isinstance(reasoning, str) and reasoning:
reasoning_parts.append(reasoning)
if on_token is not None:
on_token("reasoning", reasoning)
finally:
response.close()
if final_answer_guard is not None:
visible_text = final_answer_guard.finalize()
if on_token is not None and visible_text:
on_token("answer", visible_text)
message: dict[str, Any] = {
"role": "assistant",
"content": "".join(content_parts),
"reasoning_content": "".join(reasoning_parts),
}
tool_calls.apply_to(message)
return {"choices": [{"message": message}], "usage": usage}
# ---- 自由生成 -------------------------------------------------------
def vision_answer(
self,
request: CgVisionAnswerRequest,
*,
on_token: Callable[[str, str], None] | None = None,
progress_callback: Callable[[dict[str, Any]], None] | None = None,
cancel_token: Any = None,
) -> dict[str, Any]:
self._require_configured()
started_at = time.perf_counter()
vision_request = self._vision_request(request)
thinking_enabled = self._thinking_enabled(request)
adapter = get_provider_adapter("qwen_openai_compatible")
payload: dict[str, Any] = {
"model": self.config.vision_model,
"messages": build_qwen_direct_vqa_messages(vision_request),
"temperature": VISION_AGENT_TEMPERATURE,
"max_tokens": int(self.config.vision_max_tokens),
}
adapter.apply_payload_options(
payload, response_format_json=False, enable_thinking=thinking_enabled
)
streaming = bool(self.config.vision_streaming)
def forward_token(channel: str, text: str) -> None:
if on_token is not None:
on_token(channel, text)
if progress_callback is not None:
progress_callback(
{"_event": "token", "channel": channel, "text": text}
)
usage_attempts: list[dict[str, Any]] = []
category_correction_count = 0
answer = ""
body: dict[str, Any] = {}
hazard_categories: tuple[str, ...] = ()
format_fallback = False
for attempt in range(MAX_FREE_GENERATION_FORMAT_ATTEMPTS):
body = self._post(
payload,
streaming=streaming,
on_token=forward_token,
cancel_token=cancel_token,
)
usage_attempts.append(self.owner._extract_usage(body))
message = self.owner._extract_response_message(body)
answer = strip_think_blocks(
self.owner._extract_message_content(message), final=True
).strip()
if not answer:
raise CgVisionError(
VISION_RESPONSE_INVALID_CODE,
"视觉模型没有返回可用的回答内容。",
)
try:
hazard_categories = tuple(
extract_hazard_categories(
answer,
image_count=len(vision_request.images),
)
or ()
)
break
except HazardCategoryOutputInvalidError:
if attempt >= MAX_FREE_GENERATION_FORMAT_ATTEMPTS - 1:
answer = server_fallback_free_answer(answer)
hazard_categories = ()
format_fallback = True
break
category_correction_count = 1
if streaming and progress_callback is not None:
progress_callback({"_event": "answer_reset"})
previous: dict[str, Any] = {
"role": "assistant",
"content": answer,
}
previous_reasoning = self.owner._extract_reasoning_content(body)
if thinking_enabled and previous_reasoning:
previous["reasoning_content"] = previous_reasoning
payload = {
**payload,
"messages": [
*payload["messages"],
previous,
{
"role": "user",
"content": DIRECT_HAZARD_CATEGORY_CORRECTION_MESSAGE,
},
],
}
elapsed_ms = (time.perf_counter() - started_at) * 1000.0
usage = _sum_usage_dicts(*usage_attempts)
return _with_request_id(
{
"ok": True,
"status": "degraded" if format_fallback else "ok",
"degraded": format_fallback,
"degraded_reason": (
"free_generation_format_fallback" if format_fallback else ""
),
"validation_failures": (
["hazard_category"] if format_fallback else []
),
"mode": "free_form",
"question": request.question,
"answer": answer,
"reasoning": (
self.owner._extract_reasoning_content(body) if thinking_enabled else ""
),
"image_count": len(vision_request.images),
**({"history_turn_count": len(request.history)} if request.history else {}),
"model": self.config.vision_model,
"thinking_enabled": thinking_enabled,
"task_mode": request.task_mode,
"citations": [],
"retrieval_docs": [],
"hazard_categories": list(hazard_categories),
"hazard_taxonomy_version": HAZARD_TAXONOMY_VERSION,
"usage": usage,
"runtime_stats": {
"total_time_ms": elapsed_ms,
"hazard_category_correction_count": category_correction_count,
},
}
)
# ---- 观察 -----------------------------------------------------------
def vision_observe(
self,
request: CgVisionAnswerRequest,
*,
cancel_token: Any = None,
) -> dict[str, Any]:
"""One structured pass over the images.
Unlike the two report modes this validates the model's output against
the hazard-label vocabulary and reports the parse outcome, so a caller
can turn the photo into a retrieval query and fall back cleanly when
the model does not produce a usable observation.
"""
if request.observation_format == "facts":
return self._vision_facts(request, cancel_token=cancel_token)
self._require_configured()
started_at = time.perf_counter()
vision_request = self._vision_request(request)
adapter = get_provider_adapter("qwen_openai_compatible")
payload: dict[str, Any] = {
"model": self.config.vision_model,
"messages": build_qwen_vision_messages(vision_request),
"temperature": VISION_AGENT_TEMPERATURE,
"max_tokens": int(self.config.vision_max_tokens),
}
adapter.apply_payload_options(
payload, response_format_json=False, enable_thinking=False
)
body = self._post(
payload,
streaming=bool(self.config.vision_streaming),
cancel_token=cancel_token,
)
message = self.owner._extract_response_message(body)
raw_content = self.owner._extract_message_content(message)
safe_content = extract_safe_fallback_text(raw_content)
if not safe_content:
raise CgVisionError(
VISION_RESPONSE_INVALID_CODE, "视觉观察没有返回可安全展示的内容。"
)
profile = vision_request.profile
base = {
"ok": True,
"mode": "observe",
"text": safe_content,
"hazard_label": "",
"is_non_scene": False,
"parse_status": "failed",
"parse_error": "",
"profile_id": profile.id,
"system_prompt_sha256": profile.system_prompt_sha256,
"image_count": len(vision_request.images),
**({"history_turn_count": len(request.history)} if request.history else {}),
"model": self.config.vision_model,
"usage": self.owner._extract_usage(body),
}
try:
report = parse_hazard_report(safe_content, allowed_labels=profile.labels)
except VlmOutputInvalidError as exc:
# A schema miss is a reportable outcome, not a failed call: the
# caller decides whether to fall back or abort.
base["parse_error"] = str(exc)
base["elapsed_ms"] = (time.perf_counter() - started_at) * 1000.0
return _with_request_id(base)
base.update(
{
"text": report.text,
"hazard_label": report.label,
"is_non_scene": report.is_non_scene,
"parse_status": "succeeded",
"elapsed_ms": (time.perf_counter() - started_at) * 1000.0,
}
)
return _with_request_id(base)
def _vision_facts(self, request: CgVisionAnswerRequest, *, cancel_token: Any = None) -> dict[str, Any]:
from dataclasses import replace
import hashlib
import json
from ..agentic.visual_facts import VisualFacts, VISUAL_FACTS_PROMPT, parse_visual_facts
from ..generation.qwen_provider import build_qwen_image_messages
from ..agentic.hybrid_prompt_sets import active_prompt_set, select_hybrid_prompt, prompt_sha256
from ..agentic.hazard_categories import HAZARD_CATEGORIES
from ..agentic.native_visual_adapter import OBSERVATION_MODE_ENV
import os
observation_mode = os.environ.get(OBSERVATION_MODE_ENV, "whole")
if observation_mode == "native":
from ..agentic.native_visual_adapter import observe_native
self._require_configured()
return _with_request_id(observe_native(self,self._vision_request(request),cancel_token))
if observation_mode == "expert":
from ..agentic.expert_visual import observe_expert
self._require_configured()
return _with_request_id(observe_expert(self,self._vision_request(request),cancel_token))
if observation_mode != "whole":
raise ValueError("unknown hybrid observation mode")
visual_prompt = select_hybrid_prompt("visual", VISUAL_FACTS_PROMPT)
diagnostics: list[dict[str, Any]] = []
self._require_configured()
started_at = time.perf_counter()
vision_request = self._vision_request(request)
adapter = get_provider_adapter("qwen_openai_compatible")
observations: list[dict[str, Any]] = []
usages: list[dict[str, Any]] = []
retries = 0
for image_index, image in enumerate(vision_request.images, 1):
one_image = replace(vision_request, images=(image,), history=())
messages = build_qwen_image_messages(
one_image, system_prompt=visual_prompt,
user_text=f"观察重点：{request.question}。只观察当前这张图，输出观察 JSON。",
)
parsed = None
for attempt in range(2):
_raise_if_cancelled(cancel_token)
payload = {
"model": self.config.vision_model,
"messages": messages + (
[{"role": "user", "content": "上一条输出未通过观察 JSON 校验。仅输出符合字段定义的 JSON。"}]
if attempt else []
),
"temperature": 0.0,
"max_tokens": min(int(self.config.vision_max_tokens), 2048),
}
adapter.apply_payload_options(
payload, response_format_json=True, enable_thinking=False,
response_json_schema=VisualFacts.model_json_schema(),
)
body = self._post(payload, streaming=bool(self.config.vision_streaming), cancel_token=cancel_token)
usages.append(self.owner._extract_usage(body))
raw = self.owner._extract_message_content(self.owner._extract_response_message(body))
try:
parsed = parse_visual_facts(raw)
diagnostics.append({"image_index": image_index, "attempt": attempt + 1, "parse_status": "succeeded"})
break
except (ValueError, TypeError) as exc:
from ..agentic.vision_agentic import sanitize_untrusted_text
errors = []
if hasattr(exc, "errors"):
errors = [{"loc": list(e.get("loc", ())), "type": e.get("type")} for e in exc.errors(include_input=False, include_url=False)]
diagnostics.append({
"image_index": image_index, "attempt": attempt + 1, "parse_status": "failed",
"error_type": type(exc).__name__, "schema_errors": errors,
"safe_response_text": sanitize_untrusted_text(raw)[:4000],
})
retries += 1
if parsed is None:
return _with_request_id({
"ok": True, "mode": "observe", "observation_format": "facts",
"parse_status": "failed", "parse_error": "visual_facts_invalid",
"profile_id": "hybrid-visual-facts-v1", "prompt_set": active_prompt_set(),
"system_prompt_sha256": prompt_sha256(visual_prompt), "observation_diagnostics": diagnostics,
"observations": [], "text": "", "model": self.config.vision_model,
"image_count": len(vision_request.images),
"elapsed_ms": (time.perf_counter() - started_at) * 1000,
"usage": {key: sum(int(u.get(key) or 0) for u in usages) for key in ("prompt_tokens", "completion_tokens", "total_tokens")},
})
observations.append({"image_index": image_index, **parsed.model_dump(mode="json")})
return _with_request_id({
"ok": True, "mode": "observe", "observation_format": "facts",
"parse_status": "succeeded", "parse_error": "", "observations": observations,
"text": json.dumps(observations, ensure_ascii=False),
"image_count": len(vision_request.images), "model": self.config.vision_model,
**({"history_turn_count": len(request.history)} if request.history else {}),
"profile_id": "hybrid-visual-facts-v1",
"system_prompt_sha256": prompt_sha256(visual_prompt),
"elapsed_ms": (time.perf_counter() - started_at) * 1000,
"usage": {key: sum(int(u.get(key) or 0) for u in usages) for key in ("prompt_tokens", "completion_tokens", "total_tokens")},
"unmapped_category_hints": [
{"image_index": item["image_index"], "hint": hazard["candidate_category"]}
for item in observations for hazard in item["hazards"]
if hazard["candidate_category"] and hazard["candidate_category"] not in HAZARD_CATEGORIES
],
"structured_retry_count": retries, "prompt_set": active_prompt_set(), "observation_diagnostics": diagnostics,
})
# ---- 智能生成 -------------------------------------------------------
def _build_agent_call(
self,
*,
thinking_enabled_default: bool,
progress_callback: Callable[[dict[str, Any]], None] | None,
) -> Callable[..., ModelPhaseResponse]:
adapter = get_provider_adapter("qwen_openai_compatible")
def call_agent(
*,
messages: list[dict[str, Any]],
phase: str,
include_tools: bool,
thinking_enabled: bool,
streaming: bool,
emit_answer_tokens: bool,
emit_reasoning_tokens: bool,
final_answer_guard: Any = None,
cancel_token: Any = None,
temperature: float | None = None,
) -> ModelPhaseResponse:
_raise_if_cancelled(cancel_token)
phase_started_at = time.perf_counter()
payload: dict[str, Any] = {
"model": self.config.vision_model,
"messages": list(messages),
"temperature": (
VISION_AGENT_TEMPERATURE if temperature is None else float(temperature)
),
"max_tokens": int(self.config.vision_max_tokens),
}
if include_tools:
payload["tools"] = [build_vision_retrieval_tool_schema()]
payload["tool_choice"] = "auto"
# Keep private thinking for search, but disable it for schema-bound
# observe/final phases. This gateway routes JSON-schema output into
# reasoning_content when thinking is enabled, emptying message.content.
enforce_final_schema = phase in {"final", "controller_final"}
enforce_observe_schema = phase == "observe"
provider_thinking_enabled = (
thinking_enabled
and not enforce_final_schema
and not enforce_observe_schema
)
json_schema = None
if phase == "final":
json_schema = build_public_envelope_json_schema()
elif phase == "controller_final":
json_schema = build_answer_only_json_schema()
elif enforce_observe_schema:
json_schema = build_controller_observe_json_schema()
if enforce_observe_schema:
payload["max_tokens"] = min(int(self.config.vision_max_tokens), 1024)
adapter.apply_payload_options(
payload,
response_format_json=bool(json_schema),
enable_thinking=provider_thinking_enabled,
response_json_schema=json_schema,
)
def forward(channel: str, text: str) -> None:
if progress_callback is None:
return
if channel == "answer" and not emit_answer_tokens:
return
if channel == "reasoning" and not emit_reasoning_tokens:
return
progress_callback({"_event": "token", "channel": channel, "text": text})
body = self._post(
payload,
streaming=streaming and bool(self.config.vision_streaming),
on_token=forward,
final_answer_guard=final_answer_guard,
cancel_token=cancel_token,
)
_raise_if_cancelled(cancel_token)
message = self.owner._extract_response_message(body)
content = strip_think_blocks(
self.owner._extract_message_content(message),
final=phase in {"final", "controller_final"},
)
if phase in {"final", "controller_final"}:
content = strip_answer_tags(content).strip()
return ModelPhaseResponse(
content=content,
reasoning=(
self.owner._extract_reasoning_content(body)
if provider_thinking_enabled
else ""
),
usage=self.owner._extract_usage(body),
duration_ms=(time.perf_counter() - phase_started_at) * 1000.0,
message=dict(message),
)
return call_agent
def _build_retrieval_batch_call(
self, *, scope: str, topk: int
) -> Callable[..., list[dict[str, Any]]]:
"""In-process retrieval, shaped exactly like the MCP tool result the
runtime already knows how to read."""
def call_retrieval_batch(
*,
calls: Sequence[tuple[str, dict[str, Any]]],
cancel_token: Any = None,
) -> list[dict[str, Any]]:
results: list[dict[str, Any]] = []
for name, arguments in calls:
_raise_if_cancelled(cancel_token)
started_at = time.perf_counter()
if name != CG_RETRIEVE_RERANK_TOOL_NAME:
results.append(
compact_retrieve_rerank_tool_result(
{"ok": False, "error": {"code": "unsupported_tool", "message": name}}
)
)
continue
try:
raw = self.owner.retrieve_rerank(
CgRetrieveRerankRequest(
query=str(arguments.get("query") or ""),
scope=scope,
topk=topk,
),
cancel_token=cancel_token,
)
compact = compact_retrieve_rerank_tool_result(raw)
except Exception as exc: # surfaced to the agent as a tool error
if cancel_token is not None and getattr(cancel_token, "cancelled", False):
raise
compact = compact_retrieve_rerank_tool_result(
{
"ok": False,
"error": {"code": "service_error", "message": str(exc)[:200]},
}
)
compact["tool_duration_ms"] = (time.perf_counter() - started_at) * 1000.0
results.append(compact)
return results
return call_retrieval_batch
def vision_agentic(
self,
request: CgVisionAgenticRequest,
*,
progress_callback: Callable[[dict[str, Any]], None] | None = None,
cancel_token: Any = None,
) -> dict[str, Any]:
self._require_configured()
started_at = time.perf_counter()
vision_request = self._vision_request(request)
thinking_enabled = self._thinking_enabled(request)
topk = int(request.topk or self.config.vision_agentic_retrieval_topk)
rounds = int(
request.max_search_rounds or self.config.vision_agentic_max_search_rounds
)
profile = resolve_vision_profile(
profile_name=getattr(self.config, "vision_profile", None),
workflow=self.config.vision_agentic_workflow,
)
strict_raw = str(
getattr(self.config, "vision_strict_retrieval_failure", "") or ""
).strip().lower()
strict_override = (
True
if strict_raw in {"1", "true", "yes", "y", "on"}
else False
if strict_raw in {"0", "false", "no", "n", "off"}
else None
)
budget = float(getattr(self.config, "vision_total_budget_seconds", 0.0) or 0.0)
heartbeat = float(
getattr(self.config, "vision_sse_heartbeat_seconds", -1.0)
)
retry_count = int(getattr(self.config, "vision_upstream_retry_count", 0) or 0)
retry_sleep = float(
getattr(self.config, "vision_upstream_retry_sleep_seconds", -1.0)
)
profile = apply_vision_config_overrides(
profile,
total_budget_seconds=budget if budget > 0 else None,
sse_heartbeat_seconds=heartbeat if heartbeat >= 0 else None,
strict_retrieval_failure=strict_override,
vision_upstream_retry_count=retry_count if retry_count > 0 else None,
vision_upstream_retry_sleep_seconds=(
retry_sleep if retry_sleep >= 0 else None
),
)
workflow = profile.workflow
# Keep module-level retry knobs aligned with the active profile
# so a single source drives both continuous and controller paths.
global VISION_UPSTREAM_RETRY_COUNT, VISION_UPSTREAM_RETRY_SLEEP_SECONDS
VISION_UPSTREAM_RETRY_COUNT = int(profile.vision_upstream_retry_count)
VISION_UPSTREAM_RETRY_SLEEP_SECONDS = float(
profile.vision_upstream_retry_sleep_seconds
)
runtime = VisionAgenticSearchRuntime(
agent_call=self._build_agent_call(
thinking_enabled_default=thinking_enabled,
progress_callback=progress_callback,
),
retrieval_batch_call=self._build_retrieval_batch_call(
scope=request.scope, topk=topk
),
max_search_rounds=rounds,
retrieval_topk=topk,
workflow=workflow,
max_evidence_documents=profile.max_evidence_documents,
max_queries_per_round=profile.max_queries_per_round,
behaviour_profile=profile,
)
try:
result = runtime.run(
request=vision_request,
scope=request.scope,
final_thinking_enabled=thinking_enabled,
final_streaming=bool(self.config.vision_streaming),
progress_callback=progress_callback,
cancel_token=cancel_token,
)
except VisionAgenticStructuredOutputError as exc:
raise CgVisionError(
VISION_RESPONSE_INVALID_CODE,
"视觉模型未返回可用的结构化观察结果。",
details={
"reason": getattr(exc, "validation_code", "") or "observe_protocol"
},
) from exc
except VisionAgenticSynthesisError as exc:
raise CgVisionError(
VISION_RESPONSE_INVALID_CODE,
"视觉模型未能按要求生成最终报告，请稍后重试。",
details={
"reason": "final_report_protocol",
"validation_category": exc.validation_category,
"validation_code": exc.validation_code,
"validation_codes": list(exc.validation_codes),
"validation_failures": [
{"attempt": attempt, "category": category, "code": code}
for attempt, category, code in exc.safe_validation_failures
],
},
) from exc
except VisionAgenticBudgetExceededError as exc:
raise CgVisionError(
VISION_BUDGET_EXCEEDED_CODE,
"视觉识别超过整单时间预算，请稍后重试。",
details={"stage": getattr(exc, "stage", "") or ""},
) from exc
except VisionAgenticRetrievalError as exc:
raise CgVisionError(
VISION_RETRIEVAL_FAILED_CODE,
"视觉法规检索失败，请稍后重试。",
details={"reason": str(exc)[:200]},
) from exc
elapsed_ms = (time.perf_counter() - started_at) * 1000.0
runtime_stats = dict(result.runtime_stats)
runtime_stats.setdefault("total_time_ms", elapsed_ms)
public_validation = dict(result.metadata.get("public_envelope_validation") or {})
degraded = bool(public_validation.get("degraded"))
degraded_reason = "final_protocol_exhausted" if degraded else ""
warnings = list(result.warnings)
metadata = dict(result.metadata)
declared_hazards = any(
str(category).strip() != "无"
for category in (metadata.get("hazard_categories") or [])
)
# A valid JSON envelope cannot verify a normative report without
# usable citations. Preserve the report but mark it for human review.
if (
request.task_mode != "visual_qa"
and result.scene_status in {"assessable", "not_assessed"}
and declared_hazards
and not result.citations
):
failures = list(public_validation.get("failures") or [])
if "missing_required_citation" not in failures:
failures.append("missing_required_citation")
if not result.retrieval_docs and "missing_retrieval_evidence" not in failures:
failures.append("missing_retrieval_evidence")
public_validation.update(valid=False, degraded=True, failures=failures)
metadata["public_envelope_validation"] = public_validation
if not degraded:
degraded_reason = "missing_required_citation"
degraded = True
warning = "本次智能识别报告缺少可核对的条文引用，请人工复核。"
if warning not in warnings:
warnings.append(warning)
if (
request.task_mode != "visual_qa"
and result.scene_status == "assessable"
and result.retrieval_docs
and not metadata.get("hazard_categories")
and not degraded
):
failures = list(public_validation.get("failures") or [])
if "missing_inspection_report" not in failures:
failures.append("missing_inspection_report")
public_validation.update(valid=False, degraded=True, failures=failures)
metadata["public_envelope_validation"] = public_validation
degraded_reason = "missing_inspection_report"
degraded = True
warnings.append("本次返回了观察内容，但未完成隐患报告，请人工复核。")
return _with_request_id(
{
"ok": True,
"status": "degraded" if degraded else "ok",
"degraded": degraded,
"degraded_reason": degraded_reason,
"validation_failures": list(public_validation.get("failures") or []),
"mode": "agentic",
"question": request.question,
"answer": result.assistant_text,
"reasoning": result.reasoning if thinking_enabled else "",
"reasoning_stages": [dict(stage) for stage in result.reasoning_stages],
# Raw [cite^n] markers, deliberately: an API consumer needs the
# machine-readable form, and rendering them as 〔n〕 is the
# caller's presentation choice.
"citations": list(result.citations),
"retrieval_docs": [dict(doc) for doc in result.retrieval_docs],
"pipeline_trace": [dict(item) for item in result.pipeline_trace],
"agent_events": [dict(event) for event in result.agent_events],
"metadata": metadata,
"protocol_mode": result.protocol_mode,
"scene_status": result.scene_status,
"image_count": len(vision_request.images),
**({"history_turn_count": len(request.history)} if request.history else {}),
"model": self.config.vision_model,
"scope": request.scope,
"thinking_enabled": thinking_enabled,
"task_mode": request.task_mode,
"warnings": warnings,
"hazard_categories": list(
result.metadata.get("hazard_categories") or []
),
"hazard_taxonomy_version": str(
result.metadata.get("hazard_taxonomy_version")
or HAZARD_TAXONOMY_VERSION
),
"usage": runtime_stats.get("tokens", {}).get("total", {}),
"runtime_stats": runtime_stats,
}
)
def vision_hybrid(
self,
request: CgVisionAgenticRequest,
*,
progress_callback: Callable[[dict[str, Any]], None] | None = None,
cancel_token: Any = None,
) -> dict[str, Any]:
from .cg_rag_hybrid import run_in_process_hybrid
return run_in_process_hybrid(
self,
request,
progress_callback=progress_callback,
cancel_token=cancel_token,
)
# ---- streaming ------------------------------------------------------
def vision_answer_stream_events(
self,
request: CgVisionAnswerRequest,
*,
cancel_token: Any = None,
) -> Iterable[tuple[str, dict[str, Any]]]:
yield from self._stream_events(
lambda emit: self.vision_answer(
request,
progress_callback=emit,
cancel_token=cancel_token,
),
stage_label="正在识别图片隐患",
)
def vision_agentic_stream_events(
self,
request: CgVisionAgenticRequest,
*,
cancel_token: Any = None,
) -> Iterable[tuple[str, dict[str, Any]]]:
yield from self._stream_events(
lambda emit: self.vision_agentic(
request, progress_callback=emit, cancel_token=cancel_token
),
stage_label="正在结合法规识别图片隐患",
)
def vision_hybrid_stream_events(
self,
request: CgVisionAgenticRequest,
*,
cancel_token: Any = None,
) -> Iterable[tuple[str, dict[str, Any]]]:
yield from self._stream_events(
lambda emit: self.vision_hybrid(
request, progress_callback=emit, cancel_token=cancel_token
),
stage_label="正在根据图片描述检索法规",
)
def _stream_events(
self,
run: Callable[[Callable[[dict[str, Any]], None]], dict[str, Any]],
*,
stage_label: str,
) -> Iterable[tuple[str, dict[str, Any]]]:
"""Streams progress live while the (blocking) generation runs.
The runtime and the direct path both push progress through a callback
rather than yielding, so the generation runs on a worker thread and its
callbacks land on a queue this generator drains. Buffering them until
the run finished would technically deliver every event, but a hazard
turn takes 60-90s and the caller would sit in silence for all of it --
which is the whole reason these endpoints stream.
"""
events: queue.Queue[dict[str, Any] | None] = queue.Queue()
outcome: dict[str, Any] = {}
def worker() -> None:
try:
outcome["data"] = run(lambda payload: events.put(dict(payload)))
except BaseException as exc: # re-raised on the consumer side
outcome["error"] = exc
finally:
events.put(None)
yield ("progress", {"stage": "vision", "label": stage_label, "status": "running"})
thread = threading.Thread(target=worker, name="cg-rag-vision", daemon=True)
thread.start()
while True:
payload = events.get()
if payload is None:
break
yield (str(payload.pop("_event", "progress") or "progress"), payload)
thread.join()
error = outcome.get("error")
if isinstance(error, CgVisionError):
yield (
"error",
_with_request_id(
{"code": error.code, "message": error.message, "details": error.details}
),
)
return
if error is not None:
raise error
yield ("final", outcome.get("data") or {})