feat: 生图支持 reference_subject，画真实产品时找参考图

backend/app/ai/prompts/runtime_prompts.toml

@@ -422,7 +422,8 @@ prompt = """Detailed visual description of the image: subject, setting, composit
 aspect_ratio = """Image aspect ratio. Use 1:1 unless the user asks for another shape: landscape such as 16:9, 3:2, 4:3, or 21:9 for an ultra-wide banner; portrait such as 9:16, 2:3, 3:4, or 4:5 for a poster or phone wallpaper."""
 style = """Optional style hint, such as watercolor, photorealistic, flat illustration, oil painting, or pixel art."""
 regenerate = """Set to true only when the user asks for another version of an image already generated in this conversation and the description stays the same. The service then draws a new image instead of returning the earlier one for an identical request. Omit it otherwise."""
-success_context = """The image was generated and is shown to the user as an image card in this reply; the card already labels which image model drew it, so do not mention the model. Do not output an image link, a markdown image, or the file id. Reply briefly in the user's language: say what was drawn in one or two sentences and offer to adjust it."""
+reference_subject = """Set only when the image must show a specific real-world object whose exact appearance matters, such as a named product model, vehicle, building, or landmark. Use its exact name as the user wrote it, for example a phone model name. The service then looks up current reference photos of it so the drawing matches the real object. Omit it for generic, imaginary, or stylized subjects."""
+success_context = """The image was generated and is shown to the user as an image card in this reply; the card already labels which image model drew it, so do not mention the model. Do not output an image link, a markdown image, or the file id.{% if reference_search_status in ("none_accepted", "search_failed") %} No usable reference photo of {{ reference_subject }} was found, so it was drawn without one; tell the user plainly that its appearance may differ from the real one.{% endif %} Reply briefly in the user's language: say what was drawn in one or two sentences and offer to adjust it."""
 failed_context = """{% if error_code == "content_filtered" %}{% if fallback_from %}The primary image model refused this request under its content safety filter, and the fallback model also refused it.{% else %}The image model refused this request under its content safety filter.{% endif %} Rewording the same request and calling again will not help, so do not retry on your own. Tell the user plainly that this image could not be generated because the image provider's content filter blocked it; do not guess a more specific reason. Do not quietly generate a different image that drops what the user asked for; you may offer an alternative, such as an original design inspired by the request, and generate it only if the user agrees.{% elif fallback_from == "content_filtered" %}The primary image model refused this request under its content safety filter, and the fallback model then failed (error code: {{ error_code }}). Tell the user plainly that the image could not be generated this time.{% if retryable %} They can try again later.{% endif %}{% elif retryable %}Image generation failed temporarily (error code: {{ error_code }}). Tell the user plainly that the image could not be generated this time and that they can try again.{% else %}Image generation failed (error code: {{ error_code }}). Tell the user plainly that the image could not be generated this time.{% endif %} Do not describe the image as if it exists and do not output any image link."""
 
 [qweather]

backend/app/services/mcp/image_service_tools.py

@@ -45,6 +45,9 @@
 })
 MAX_PROMPT_CHARS = 1_000
 MAX_STYLE_CHARS = 60
+MAX_REFERENCE_SUBJECT_CHARS = 80
+# image-service 找参考图的结果状态；none_accepted 与 search_failed 时照常出图，但没有参考图，外观可能和真品不同。
+REFERENCE_SEARCH_STATUSES = frozenset({"found", "cached", "none_accepted", "search_failed", "skipped"})
 # 生图有真实费用：同一 run 最多生成的张数。
 MAX_IMAGES_PER_RUN = 3
 # 实测 Gemini Flash 约 15 秒，Pro 与 Seedream 更慢；image-service 自身上游超时为 300 秒。
@@ -107,6 +110,8 @@ class GeneratedImageOutcome:
     requested_model: str | None = None
     fallback_used: bool = False
     grounding_used: bool = False
+    reference_search_status: str | None = None
+    reference_image_count: int = 0
 
 
 def is_image_service_row(row: Any) -> bool:
@@ -142,6 +147,11 @@ def build_generate_image_definition() -> dict[str, Any]:
                         "type": "boolean",
                         "description": render_runtime_prompt("image_service.regenerate"),
                     },
+                    "reference_subject": {
+                        "type": "string",
+                        "maxLength": MAX_REFERENCE_SUBJECT_CHARS,
+                        "description": render_runtime_prompt("image_service.reference_subject"),
+                    },
                 },
                 "required": ["prompt"],
                 "additionalProperties": False,
@@ -202,9 +212,20 @@ def parse_generate_image_payload(payload: Any) -> GeneratedImageOutcome:
         requested_model=_model_id(structured.get("requested_model")),
         fallback_used=structured.get("fallback_used") is True,
         grounding_used=structured.get("grounding_used") is True,
+        reference_search_status=_reference_search_status(structured.get("reference_search")),
+        reference_image_count=_reference_image_count(structured.get("reference_images_used")),
     )
 
 
+def _reference_search_status(value: Any) -> str | None:
+    status = value.get("status") if isinstance(value, dict) else None
+    return status if status in REFERENCE_SEARCH_STATUSES else None
+
+
+def _reference_image_count(value: Any) -> int:
+    return len(value) if isinstance(value, list) else 0
+
+
 def _image_path(url: Any) -> str:
     if not isinstance(url, str):
         raise ImageGenerationError("invalid_response")
@@ -338,6 +359,9 @@ async def execute(self, args: dict) -> ToolResult:
                 "requested_model": outcome.requested_model,
                 "fallback_used": outcome.fallback_used,
                 "grounding_used": outcome.grounding_used,
+                "reference_subject": arguments.get("reference_subject"),
+                "reference_search_status": outcome.reference_search_status,
+                "reference_image_count": outcome.reference_image_count,
             },
         )
 
@@ -421,7 +445,11 @@ def format_llm_context(self, result: ToolResult, *, citation_numbers: list[int]
                 retryable=data.get("retryable") is True,
                 fallback_from=data.get("fallback_from"),
             )
-        return render_runtime_prompt("image_service.success_context")
+        return render_runtime_prompt(
+            "image_service.success_context",
+            reference_subject=data.get("reference_subject"),
+            reference_search_status=data.get("reference_search_status"),
+        )
 
     def _build_result_summary(self, result: ToolResult) -> dict:
         data = result.data or {}
@@ -434,7 +462,7 @@ def _build_result_summary(self, result: ToolResult) -> dict:
 
 
 def _normalized_arguments(args: Any) -> dict[str, Any]:
-    if not isinstance(args, dict) or set(args) - {"prompt", "aspect_ratio", "style", "regenerate"}:
+    if not isinstance(args, dict) or set(args) - {"prompt", "aspect_ratio", "style", "regenerate", "reference_subject"}:
         raise ImageGenerationError("invalid_arguments")
     prompt = args.get("prompt")
     if not isinstance(prompt, str) or not prompt.strip() or len(prompt) > MAX_PROMPT_CHARS:
@@ -459,6 +487,12 @@ def _normalized_arguments(args: Any) -> dict[str, Any]:
             raise ImageGenerationError("invalid_arguments")
         if style.strip():
             arguments["style"] = style.strip()
+    reference_subject = args.get("reference_subject")
+    if reference_subject is not None:
+        if not isinstance(reference_subject, str) or len(reference_subject) > MAX_REFERENCE_SUBJECT_CHARS:
+            raise ImageGenerationError("invalid_arguments")
+        if reference_subject.strip():
+            arguments["reference_subject"] = reference_subject.strip()
     return arguments
 
 

backend/test/test_image_service_tools.py

@@ -194,6 +194,8 @@ async def test_invalid_arguments_fail_before_remote_call(self):
             {"prompt": "x", "aspect_ratio": "2:1"},
             {"prompt": "x", "size": "4k"},
             {"prompt": "x", "regenerate": "yes"},
+            {"prompt": "x", "reference_subject": 18},
+            {"prompt": "x", "reference_subject": "x" * 81},
         ):
             with self.subTest(args=args):
                 self.assertTrue(handler.validate_arguments(args))
@@ -265,6 +267,42 @@ async def test_grounding_is_always_requested_and_reported(self):
         self.assertIs(client.calls[0][2]["grounding"], True)
         self.assertTrue(result.data["grounding_used"])
 
+    async def test_reference_subject_is_passed_only_when_given(self):
+        handler, client, *_ = self.build_handler()
+
+        await handler.execute({"prompt": "a fox"})
+        await handler.execute({"prompt": "a fox", "reference_subject": " "})
+
+        self.assertNotIn("reference_subject", client.calls[0][2])
+        self.assertNotIn("reference_subject", client.calls[1][2])
+
+    async def test_reference_search_status_reaches_result_and_context(self):
+        for status, warns in (("found", False), ("cached", False), ("none_accepted", True), ("search_failed", True)):
+            with self.subTest(status=status):
+                client = self.structured_client(
+                    {
+                        "ok": True,
+                        "image_url": f"http://image-service.dev.seanfield.org/static/images/{IMAGE_NAME}",
+                        "model": "gemini-3.1-flash-image-preview",
+                        "reference_images_used": ["https://www.apple.com/a.jpg"] if not warns else [],
+                        "reference_search": {"status": status, "candidates": 8, "accepted": 0 if warns else 1},
+                        "id": "gen-3",
+                    },
+                    text=SUCCESS_TEXT,
+                )
+                handler, _, _, _ = self.build_handler(client=client)
+
+                result = await handler.execute({"prompt": "iPhone 18 Pro 产品宣传图", "reference_subject": " iPhone 18 Pro "})
+
+                self.assertEqual(client.calls[0][2]["reference_subject"], "iPhone 18 Pro")
+                self.assertEqual(result.data["reference_subject"], "iPhone 18 Pro")
+                self.assertEqual(result.data["reference_search_status"], status)
+                self.assertEqual(result.data["reference_image_count"], 0 if warns else 1)
+                context = handler.format_llm_context(result)
+                self.assertEqual("appearance may differ" in context, warns)
+                if warns:
+                    self.assertIn("iPhone 18 Pro", context)
+
     async def test_structured_failures_map_error_codes_and_guidance(self):
         cases = {
             "content_filtered": (