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} 字节。",
)
parts.append(
VisionImagePart(
attachment_id=f"inline-{ordinal}",
ordinal=ordinal,
mime_type=mime_type,
data=data,
)
)
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),
)
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,
cancel_token: Any = None,
) -> dict[str, Any]:
_raise_if_cancelled(cancel_token)
with self.owner.capacity_limiters.generation.acquire(cancel_token=cancel_token):
try:
if streaming:
return self._post_streaming(payload, on_token=on_token, 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:
raise CgVisionError(
VISION_UPSTREAM_ERROR_CODE, "视觉模型调用失败。"
) from exc
if not isinstance(body, dict):
raise CgVisionError(VISION_RESPONSE_INVALID_CODE, "视觉模型返回的响应体不是对象。")
return body
def _post_streaming(
self,
payload: dict[str, Any],
*,
on_token: Callable[[str, str], 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)
if on_token is not None:
on_token("answer", 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()
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,
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
)
body = self._post(
payload,
streaming=bool(self.config.vision_streaming),
on_token=on_token,
cancel_token=cancel_token,
)
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, "视觉模型没有返回可用的回答内容。"
)
elapsed_ms = (time.perf_counter() - started_at) * 1000.0
return _with_request_id(
{
"ok": True,
"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),
"model": self.config.vision_model,
"thinking_enabled": thinking_enabled,
"citations": [],
"retrieval_docs": [],
"usage": self.owner._extract_usage(body),
"runtime_stats": {"total_time_ms": elapsed_ms},
}
)
# ---- 观察 -----------------------------------------------------------
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.
"""
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),
"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 _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,
) -> 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,
"max_tokens": int(self.config.vision_max_tokens),
}
if include_tools:
payload["tools"] = [build_vision_retrieval_tool_schema()]
payload["tool_choice"] = "auto"
# The final phase's content must be a strict JSON envelope. A
# guided json_schema catches most drift at decode time instead of
# burning protocol-correction retries. Only safe with thinking
# off: with it on this deployment routes the whole JSON into
# reasoning_content and leaves content null.
enforce_final_schema = phase == "final" and not thinking_enabled
adapter.apply_payload_options(
payload,
response_format_json=enforce_final_schema,
enable_thinking=thinking_enabled,
response_json_schema=(
build_public_envelope_json_schema() if enforce_final_schema else None
),
)
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,
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 == "final"
)
if phase == "final":
content = strip_answer_tags(content).strip()
return ModelPhaseResponse(
content=content,
reasoning=(
self.owner._extract_reasoning_content(body) if 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
)
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,
)
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,
)
elapsed_ms = (time.perf_counter() - started_at) * 1000.0
runtime_stats = dict(result.runtime_stats)
runtime_stats.setdefault("total_time_ms", elapsed_ms)
return _with_request_id(
{
"ok": True,
"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": dict(result.metadata),
"protocol_mode": result.protocol_mode,
"scene_status": result.scene_status,
"image_count": len(vision_request.images),
"model": self.config.vision_model,
"scope": request.scope,
"thinking_enabled": thinking_enabled,
"warnings": list(result.warnings),
"usage": runtime_stats.get("tokens", {}).get("total", {}),
"runtime_stats": runtime_stats,
}
)
# ---- 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,
on_token=lambda channel, text: emit(
{"_event": "token", "channel": channel, "text": text}
),
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 _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 {})