leefer/zhuangxiu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: ship end-to-end interior design MVP

Browse Source
This commit is contained in:
Codex
2026-08-02 08:57:10 +08:00
parent 4d97a30616
commit a96b1bf9f9
12 changed files with 2497 additions and 485 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/api/app/api/routes.py
+337 -1
View File
@@ -1,18 +1,23 @@
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from uuid import uuid4
import base64
from fastapi import APIRouter, Depends, File, Form, HTTPException, Response, UploadFile, status
from pydantic import BaseModel
from pydantic import BaseModel, Field
from app.config import Settings, get_settings
from app.api.settings import require_workflow_ready
from app.domain.models import (
CommandRequest,
DesignBrief,
PlanIssue,
PlanLayer,
PlanState,
ProjectSnapshot,
ProjectEvent,
RenderAsset,
WorkflowDefinition,
WorkflowStage,
)
@@ -24,11 +29,23 @@ from app.domain.workflow import (
workflow_definition,
)
from app.integrations.model_router import ModelRouter
from app.integrations.openai_compatible import OpenAICompatibleGateway
from app.integrations.ocr import BaiduOcrAdapter
from app.integrations.storage import S3Storage
from app.repositories.postgres import postgres_repository
from app.runtime_settings import EncryptedSettingsStore, get_runtime_store
from app.services.plan_ingestion import inspect_pdf
from app.services.design_pipeline import (
analyze_space,
create_style_directions,
crop_plan_preview,
design_chat,
fallback_structure,
generated_image_bytes,
model_display_name,
render_id,
resolve_model_instance,
)
router = APIRouter(prefix="/v1")
MAX_UPLOAD_BYTES = 25 * 1024 * 1024
@@ -39,6 +56,38 @@ class UploadRequest(BaseModel):
content_type: str
class SpatialAnalysisRequest(BaseModel):
model_instance_id: str = ""
gross_area_sqm: float | None = Field(default=None, gt=0, le=2000)
class StyleDirectionRequest(BaseModel):
brief: DesignBrief
class RenderRequest(BaseModel):
direction_id: str
room: str = "客餐厅"
view: str = "入口看向客厅"
model_instance_id: str = ""
aspect_ratio: str = "16:9"
size: str = "1K"
class RenderEditRequest(BaseModel):
instruction: str = Field(min_length=2, max_length=1000)
model_instance_id: str = ""
class DesignChatRequest(BaseModel):
message: str = Field(min_length=1, max_length=2000)
class DesignChatResponse(BaseModel):
project: ProjectSnapshot
reply: str
def get_project_repository(settings: Settings = Depends(get_settings)):
return postgres_repository(settings.database_url)
@@ -116,6 +165,293 @@ def execute_command(
return repository.save(updated)
def _event(kind: str, message: str) -> ProjectEvent:
return ProjectEvent(id=f"event-{uuid4().hex[:10]}", kind=kind, message=message)
def _selected_region(project: ProjectSnapshot):
return next(
(region for region in project.plan.regions if region.id == project.plan.selected_region_id),
None,
)
@router.post(
"/projects/{project_id}/chat",
response_model=DesignChatResponse,
dependencies=[Depends(require_workflow_ready)],
)
async def chat_with_designer(
project_id: str,
request: DesignChatRequest,
runtime_store: EncryptedSettingsStore = Depends(get_runtime_store),
repository: Any = Depends(get_project_repository),
) -> DesignChatResponse:
project = repository.get(project_id)
if project is None:
raise HTTPException(status_code=404, detail="Project not found.")
selected = next(
(item for item in project.directions if item.id == project.style.selected_direction_id),
None,
)
reply, updates = await design_chat(
runtime_store.merged_values(),
stage=project.stage.value,
message=request.message,
brief=project.brief,
structure=project.plan.structure,
selected_direction=selected,
)
allowed = {
"residents",
"lifestyle",
"focus_rooms",
"preferred_styles",
"preferred_colors",
"disliked_elements",
"must_keep",
"budget_level",
"additional_notes",
}
merged = project.brief.model_dump()
merged.update({key: value for key, value in updates.items() if key in allowed})
project.brief = DesignBrief.model_validate(merged)
project.events.extend(
[
_event("user", request.message),
_event("assistant", reply),
]
)
project.revision += 1
project.updated_at = datetime.now(UTC)
return DesignChatResponse(project=repository.save(project), reply=reply)
@router.post(
"/projects/{project_id}/spatial-analysis",
response_model=ProjectSnapshot,
dependencies=[Depends(require_workflow_ready)],
)
async def run_spatial_analysis(
project_id: str,
request: SpatialAnalysisRequest,
runtime_store: EncryptedSettingsStore = Depends(get_runtime_store),
repository: Any = Depends(get_project_repository),
) -> ProjectSnapshot:
project = repository.get(project_id)
if project is None:
raise HTTPException(status_code=404, detail="Project not found.")
if project.stage != WorkflowStage.PLAN_REVIEW or not project.plan.selected_region_id:
raise HTTPException(status_code=422, detail="请先选择目标户型区域。")
values = runtime_store.merged_values()
storage = S3Storage(values)
try:
preview, _ = storage.get_output(str(project.plan.preview_object_key))
cropped = crop_plan_preview(preview, _selected_region(project))
crop_key = f"projects/{project_id}/derived/selected-plan.png"
storage.put_output(crop_key, cropped, "image/png")
analysis = await analyze_space(
values,
cropped,
model_instance_id=request.model_instance_id,
gross_area_sqm=request.gross_area_sqm,
)
message = f"{analysis.model_name} 已完成空间理解，识别到 {len(analysis.rooms)} 个房间。"
except Exception as exc:
analysis = fallback_structure(str(exc))
message = "空间模型调用失败，已建立可编辑房间草案，不阻塞后续设计。"
project.plan.structure = analysis
if request.gross_area_sqm is not None:
project.plan.gross_area_sqm = request.gross_area_sqm
project.events.append(_event("spatial_analysis", message))
project.revision += 1
project.updated_at = datetime.now(UTC)
return repository.save(project)
@router.post(
"/projects/{project_id}/style-directions",
response_model=ProjectSnapshot,
dependencies=[Depends(require_workflow_ready)],
)
async def generate_style_directions(
project_id: str,
request: StyleDirectionRequest,
runtime_store: EncryptedSettingsStore = Depends(get_runtime_store),
repository: Any = Depends(get_project_repository),
) -> ProjectSnapshot:
project = repository.get(project_id)
if project is None:
raise HTTPException(status_code=404, detail="Project not found.")
if project.stage != WorkflowStage.STYLE_BRIEF:
raise HTTPException(status_code=422, detail="请先确认户型结构并生成空间骨架。")
brief = DesignBrief.model_validate({**request.brief.model_dump(), "completed": True})
directions, model_name = await create_style_directions(
runtime_store.merged_values(), brief, project.plan.structure
)
project.brief = brief
project.directions = directions
project.stage = WorkflowStage.DIRECTION_SELECTION
project.available_commands = available_commands(project.stage)
project.revision += 1
project.updated_at = datetime.now(UTC)
project.events.append(_event("style_directions", f"{model_name} 已生成三套风格方向。"))
return repository.save(project)
def _image_options(values: dict[str, Any], instance_id: str, request: RenderRequest) -> dict[str, Any]:
item = next(
(candidate for candidate in values.get("model_pool", []) if candidate.get("id") == instance_id),
{},
)
profile = item.get("image_parameter_profile", "generic")
if profile == "gpt_image_2":
return {"size": "1536x1024", "quality": "medium"}
return {"aspect_ratio": request.aspect_ratio, "size": request.size}
@router.post(
"/projects/{project_id}/renders",
response_model=ProjectSnapshot,
dependencies=[Depends(require_workflow_ready)],
)
async def generate_project_render(
project_id: str,
request: RenderRequest,
runtime_store: EncryptedSettingsStore = Depends(get_runtime_store),
repository: Any = Depends(get_project_repository),
) -> ProjectSnapshot:
project = repository.get(project_id)
if project is None:
raise HTTPException(status_code=404, detail="Project not found.")
direction = next((item for item in project.directions if item.id == request.direction_id), None)
if direction is None:
raise HTTPException(status_code=422, detail="请先选择有效的风格方向。")
if project.stage not in {WorkflowStage.RENDER_REVIEW, WorkflowStage.EDITING}:
raise HTTPException(status_code=422, detail="当前阶段不能生成效果图。")
values = runtime_store.merged_values()
instance_id = await resolve_model_instance(values, "image", request.model_instance_id)
storage = S3Storage(values)
preview, _ = storage.get_output(str(project.plan.preview_object_key))
cropped = crop_plan_preview(preview, _selected_region(project))
reference = f"data:image/png;base64,{base64.b64encode(cropped).decode('ascii')}"
rooms = "、".join(room.name for room in project.plan.structure.rooms)
prompt = (
f"住宅室内设计效果图，空间：{request.room}，视角：{request.view}。"
f"设计方向：{direction.name}。{direction.thesis}。"
f"关键词：{'、'.join(direction.keywords)}。材质：{'、'.join(direction.materials)}。"
f"照明：{direction.lighting}。户型包含：{rooms or '以参考平面图为准'}。"
"严格尊重参考平面图的空间关系、门窗位置和主要动线，不新增不存在的门窗或房间。"
"真实住宅尺度，专业室内摄影，材质自然，色彩协调，不出现文字、水印和施工尺寸标注。"
)
options = _image_options(values, instance_id, request)
options["images"] = [reference]
try:
payload = await OpenAICompatibleGateway(values).generate_image(
prompt, model_instance_id=instance_id, **options
)
content, content_type = await generated_image_bytes(payload)
asset_id = render_id()
extension = "png" if "png" in content_type else "jpg"
object_key = f"projects/{project_id}/renders/{asset_id}.{extension}"
storage.put_render(object_key, content, content_type)
except Exception as exc:
raise HTTPException(status_code=502, detail=f"真实效果图生成失败：{exc}") from exc
asset = RenderAsset(
id=asset_id,
direction_id=direction.id,
room=request.room,
view=request.view,
object_key=object_key,
prompt=prompt,
model_name=model_display_name(values, instance_id),
)
project.renders.append(asset)
project.stage = WorkflowStage.EDITING
project.available_commands = available_commands(project.stage)
project.revision += 1
project.updated_at = datetime.now(UTC)
project.events.append(_event("render", f"已生成 {request.room} 的 {request.view} 效果图。"))
return repository.save(project)
@router.post(
"/projects/{project_id}/renders/{asset_id}/edit",
response_model=ProjectSnapshot,
dependencies=[Depends(require_workflow_ready)],
)
async def edit_project_render(
project_id: str,
asset_id: str,
request: RenderEditRequest,
runtime_store: EncryptedSettingsStore = Depends(get_runtime_store),
repository: Any = Depends(get_project_repository),
) -> ProjectSnapshot:
project = repository.get(project_id)
if project is None:
raise HTTPException(status_code=404, detail="Project not found.")
parent = next((item for item in project.renders if item.id == asset_id), None)
if parent is None:
raise HTTPException(status_code=404, detail="没有找到要修改的效果图版本。")
values = runtime_store.merged_values()
instance_id = await resolve_model_instance(values, "image", request.model_instance_id)
storage = S3Storage(values)
original, _ = storage.get_render(parent.object_key)
prompt = (
f"只修改以下内容：{request.instruction}。"
"保持原图的空间结构、相机位置、门窗、家具尺度和未提及区域完全一致。"
"保持专业住宅摄影质感，不添加文字或水印。"
)
try:
payload = await OpenAICompatibleGateway(values).edit_image(
prompt, original, model_instance_id=instance_id
)
content, content_type = await generated_image_bytes(payload)
new_id = render_id()
extension = "png" if "png" in content_type else "jpg"
object_key = f"projects/{project_id}/renders/{new_id}.{extension}"
storage.put_render(object_key, content, content_type)
except Exception as exc:
raise HTTPException(status_code=502, detail=f"局部修改失败：{exc}") from exc
project.renders.append(
RenderAsset(
id=new_id,
direction_id=parent.direction_id,
room=parent.room,
view=parent.view,
object_key=object_key,
prompt=prompt,
model_name=model_display_name(values, instance_id),
parent_id=parent.id,
edit_instruction=request.instruction,
)
)
project.revision += 1
project.updated_at = datetime.now(UTC)
project.events.append(_event("edit", f"已完成修改：{request.instruction}"))
return repository.save(project)
@router.get("/projects/{project_id}/renders/{asset_id}")
def get_project_render(
project_id: str,
asset_id: str,
runtime_store: EncryptedSettingsStore = Depends(get_runtime_store),
repository: Any = Depends(get_project_repository),
) -> Response:
project = repository.get(project_id)
if project is None:
raise HTTPException(status_code=404, detail="Project not found.")
asset = next((item for item in project.renders if item.id == asset_id), None)
if asset is None:
raise HTTPException(status_code=404, detail="Render not found.")
try:
content, content_type = S3Storage(runtime_store.merged_values()).get_render(asset.object_key)
except Exception as exc:
raise HTTPException(status_code=503, detail=f"读取效果图失败：{exc}") from exc
return Response(content=content, media_type=content_type, headers={"Cache-Control": "private, max-age=300"})
@router.post(
"/projects/ingest",
response_model=ProjectSnapshot,
services/api/app/domain/models.py
+73
View File
@@ -61,6 +61,27 @@ class PlanIssue(BaseModel):
resolved: bool = False
class RoomProfile(BaseModel):
id: str
name: str
kind: str
area_sqm: float | None = None
confidence: float = Field(default=0.5, ge=0, le=1)
notes: str = ""
class StructureAnalysis(BaseModel):
status: str = "pending"
summary: str = ""
rooms: list[RoomProfile] = Field(default_factory=list)
circulation: str = ""
daylight: str = ""
layout_opportunities: list[str] = Field(default_factory=list)
risks: list[str] = Field(default_factory=list)
model_name: str = ""
degraded: bool = False
class PlanState(BaseModel):
source_name: str
source_kind: str
@@ -81,6 +102,8 @@ class PlanState(BaseModel):
issues: list[PlanIssue] = Field(default_factory=list)
scale_mm_per_unit: float | None = None
ceiling_height_mm: int = 2800
gross_area_sqm: float | None = None
structure: StructureAnalysis = Field(default_factory=StructureAnalysis)
class CameraView(BaseModel):
@@ -113,6 +136,52 @@ class StyleState(BaseModel):
density: str = "适度留白"
avoid: list[str] = Field(default_factory=list)
locked_decisions: list[str] = Field(default_factory=list)
selected_direction_id: str | None = None
class DesignBrief(BaseModel):
residents: str = ""
lifestyle: list[str] = Field(default_factory=list)
focus_rooms: list[str] = Field(default_factory=list)
preferred_styles: list[str] = Field(default_factory=list)
preferred_colors: list[str] = Field(default_factory=list)
disliked_elements: list[str] = Field(default_factory=list)
must_keep: list[str] = Field(default_factory=list)
budget_level: str = "适中"
additional_notes: str = ""
completed: bool = False
class StyleDirection(BaseModel):
id: str
name: str
thesis: str
keywords: list[str] = Field(default_factory=list)
palette: list[ColorToken] = Field(default_factory=list)
materials: list[str] = Field(default_factory=list)
lighting: str = ""
prompt: str = ""
model_name: str = ""
class RenderAsset(BaseModel):
id: str
direction_id: str
room: str
view: str
object_key: str
prompt: str
model_name: str
parent_id: str | None = None
edit_instruction: str = ""
created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
class ProjectEvent(BaseModel):
id: str
kind: str
message: str
created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
class ProjectSnapshot(BaseModel):
@@ -123,6 +192,10 @@ class ProjectSnapshot(BaseModel):
plan: PlanState
scene: SceneState = Field(default_factory=SceneState)
style: StyleState = Field(default_factory=StyleState)
brief: DesignBrief = Field(default_factory=DesignBrief)
directions: list[StyleDirection] = Field(default_factory=list)
renders: list[RenderAsset] = Field(default_factory=list)
events: list[ProjectEvent] = Field(default_factory=list)
available_commands: list[WorkflowCommand] = Field(default_factory=list)
last_stable_stage: WorkflowStage | None = None
failure_reason: str | None = None
services/api/app/domain/workflow.py
+41
View File
@@ -2,8 +2,11 @@ from copy import deepcopy
from datetime import UTC, datetime
from app.domain.models import (
CameraView,
CommandRequest,
DesignBrief,
ProjectSnapshot,
RoomProfile,
WorkflowCommand,
WorkflowDefinition,
WorkflowStage,
@@ -91,13 +94,51 @@ def apply_command(project: ProjectSnapshot, request: CommandRequest) -> ProjectS
updated.plan.ceiling_height_mm = int(
request.payload.get("ceiling_height_mm", updated.plan.ceiling_height_mm)
)
if request.payload.get("gross_area_sqm") is not None:
updated.plan.gross_area_sqm = float(request.payload["gross_area_sqm"])
if request.payload.get("room_names"):
existing = {room.name: room for room in updated.plan.structure.rooms}
updated.plan.structure.rooms = [
existing.get(name)
or RoomProfile(
id=f"room-{index + 1}",
name=name,
kind="other",
confidence=1,
notes="用户确认",
)
for index, raw_name in enumerate(request.payload["room_names"])
if (name := str(raw_name).strip())
]
for issue in updated.plan.issues:
if issue.kind in {"scale", "ocr"}:
issue.resolved = True
elif request.command == WorkflowCommand.BUILD_BLOCKOUT:
room_names = [room.name for room in updated.plan.structure.rooms]
focus = room_names[:3] or ["客餐厅", "主卧", "入口"]
updated.scene.cameras = [
CameraView(id=f"camera-{index + 1}", label=f"{room}主视角", room=room)
for index, room in enumerate(focus)
]
elif request.command == WorkflowCommand.SUBMIT_STYLE_BRIEF:
updated.brief = DesignBrief.model_validate({**updated.brief.model_dump(), **request.payload, "completed": True})
for field in ("concept", "lighting", "forms", "density"):
if field in request.payload:
setattr(updated.style, field, request.payload[field])
for field in ("keywords", "materials", "avoid", "locked_decisions"):
if field in request.payload:
setattr(updated.style, field, list(request.payload[field]))
elif request.command == WorkflowCommand.SELECT_DIRECTION:
direction_id = str(request.payload.get("direction_id", ""))
direction = next((item for item in updated.directions if item.id == direction_id), None)
if direction is None:
raise InvalidTransitionError(f"Direction '{direction_id}' does not exist in this project.")
updated.style.selected_direction_id = direction_id
updated.style.concept = direction.thesis
updated.style.keywords = direction.keywords
updated.style.palette = direction.palette
updated.style.materials = direction.materials
updated.style.lighting = direction.lighting
updated.available_commands = available_commands(updated.stage)
return updated
services/api/app/integrations/storage.py
+12
View File
@@ -87,6 +87,18 @@ class S3Storage:
)
return response["Body"].read(), response.get("ContentType", "application/octet-stream")
def put_render(self, object_key: str, content: bytes, content_type: str) -> None:
self._put(self._value("s3_bucket_renders"), object_key, content, content_type)
def get_render(self, object_key: str) -> tuple[bytes, str]:
if not self.configured:
raise RuntimeError("MinIO/S3 is not configured.")
response = self._client().get_object(
Bucket=self._value("s3_bucket_renders"),
Key=object_key,
)
return response["Body"].read(), response.get("ContentType", "application/octet-stream")
def _put(self, bucket: str, object_key: str, content: bytes, content_type: str) -> None:
if not self.configured:
raise RuntimeError("MinIO/S3 is not configured.")
services/api/app/services/design_pipeline.py
+352
View File
@@ -0,0 +1,352 @@
import base64
import json
import re
from io import BytesIO
from typing import Any
from uuid import uuid4
import httpx
from PIL import Image
from app.domain.models import (
ColorToken,
DesignBrief,
PlanRegion,
RoomProfile,
StructureAnalysis,
StyleDirection,
)
from app.integrations.openai_compatible import OpenAICompatibleGateway
def crop_plan_preview(preview: bytes, region: PlanRegion | None) -> bytes:
image = Image.open(BytesIO(preview)).convert("RGB")
if region is not None:
x0, y0, x1, y1 = region.bounds
padding = 0.015
box = (
max(0, int((x0 - padding) * image.width)),
max(0, int((y0 - padding) * image.height)),
min(image.width, int((x1 + padding) * image.width)),
min(image.height, int((y1 + padding) * image.height)),
)
image = image.crop(box)
image.thumbnail((1536, 1536))
output = BytesIO()
image.save(output, format="PNG", optimize=True)
return output.getvalue()
def _json_from_model(payload: dict[str, Any]) -> dict[str, Any]:
try:
content = payload["choices"][0]["message"]["content"]
except (KeyError, IndexError, TypeError) as exc:
raise ValueError("模型没有返回可读取的消息内容。") from exc
if isinstance(content, list):
content = "".join(
str(item.get("text", "")) for item in content if isinstance(item, dict)
)
text = str(content).strip()
fenced = re.search(r"```(?:json)?\s*(\{.*\})\s*```", text, re.DOTALL)
if fenced:
text = fenced.group(1)
else:
start, end = text.find("{"), text.rfind("}")
if start >= 0 and end > start:
text = text[start : end + 1]
parsed = json.loads(text)
if not isinstance(parsed, dict):
raise ValueError("模型返回内容不是 JSON 对象。")
return parsed
async def resolve_model_instance(values: dict[str, Any], route: str, explicit_id: str = "") -> str:
if explicit_id:
return explicit_id
capability = "spatial_understanding" if route == "spatial" else "image_generation"
mode_key = "spatial_routing_mode" if route == "spatial" else "image_routing_mode"
selected_key = "spatial_model_id" if route == "spatial" else "image_model_id"
if values.get(mode_key) == "manual" and values.get(selected_key):
return str(values[selected_key])
candidates = [
item
for item in values.get("model_pool", [])
if isinstance(item, dict)
and item.get("enabled", True)
and capability in item.get("capabilities", [])
]
if not candidates:
raise RuntimeError(f"没有支持 {capability} 的已启用模型。")
if len(candidates) == 1:
return str(candidates[0]["id"])
choices = [
{
"id": item.get("id"),
"name": item.get("name"),
"model_id": item.get("model_id"),
"capabilities": item.get("capabilities", []),
}
for item in candidates
]
task = "理解住宅平面图的空间关系" if route == "spatial" else "生成高审美住宅室内效果图"
prompt = (
f"为任务“{task}”从候选模型中选择一个。只返回 JSON："
f"{{\"model_instance_id\":\"候选 id\"}}。候选：{json.dumps(choices, ensure_ascii=False)}"
)
try:
response = await OpenAICompatibleGateway(values).chat([{"role": "user", "content": prompt}])
selected = str(_json_from_model(response).get("model_instance_id") or "")
if any(item.get("id") == selected for item in candidates):
return selected
except Exception:
pass
return str(candidates[0]["id"])
def model_display_name(values: dict[str, Any], instance_id: str) -> str:
for item in values.get("model_pool", []):
if isinstance(item, dict) and item.get("id") == instance_id:
return str(item.get("name") or item.get("model_id") or instance_id)
return instance_id
async def analyze_space(
values: dict[str, Any],
preview: bytes,
*,
model_instance_id: str = "",
gross_area_sqm: float | None = None,
) -> StructureAnalysis:
instance_id = await resolve_model_instance(values, "spatial", model_instance_id)
encoded = base64.b64encode(preview).decode("ascii")
area_hint = f"已知建筑面积约 {gross_area_sqm} 平方米。" if gross_area_sqm else "建筑面积未知。"
prompt = f"""你是住宅空间设计师。请分析这张户型平面图，重点服务于布局和风格效果图，不做施工承诺。{area_hint}
只返回 JSON，不要解释，格式如下：
{{
"summary": "一句话户型判断",
"rooms": [{{"name":"客厅","kind":"living","area_sqm":20.0,"confidence":0.8,"notes":"采光或连接关系"}}],
"circulation": "主要动线判断",
"daylight": "采光判断",
"layout_opportunities": ["最多4条可利用的布局机会"],
"risks": ["最多4条需要用户确认的问题"]
}}
无法确定的面积请填 null，不能凭空假定墙体可拆。"""
response = await OpenAICompatibleGateway(values).chat(
[
{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{
"type": "image_url",
"image_url": {"url": f"data:image/png;base64,{encoded}"},
},
],
}
],
vision=True,
model_instance_id=instance_id,
)
data = _json_from_model(response)
rooms = []
for index, item in enumerate(data.get("rooms", [])):
if not isinstance(item, dict) or not item.get("name"):
continue
rooms.append(
RoomProfile(
id=str(item.get("id") or f"room-{index + 1}"),
name=str(item["name"]),
kind=str(item.get("kind") or "other"),
area_sqm=item.get("area_sqm"),
confidence=float(item.get("confidence", 0.55)),
notes=str(item.get("notes") or ""),
)
)
return StructureAnalysis(
status="ready",
summary=str(data.get("summary") or "空间模型已完成初步识别，请人工确认。"),
rooms=rooms,
circulation=str(data.get("circulation") or ""),
daylight=str(data.get("daylight") or ""),
layout_opportunities=[str(item) for item in data.get("layout_opportunities", [])][:6],
risks=[str(item) for item in data.get("risks", [])][:6],
model_name=model_display_name(values, instance_id),
)
def fallback_structure(reason: str = "") -> StructureAnalysis:
defaults = [
("客厅", "living"),
("餐厅", "dining"),
("主卧", "bedroom"),
("次卧", "bedroom"),
("厨房", "kitchen"),
("卫生间", "bathroom"),
]
return StructureAnalysis(
status="needs_review",
summary="空间模型暂未给出可靠结果，已建立可编辑房间草案。",
rooms=[
RoomProfile(id=f"room-{index + 1}", name=name, kind=kind, confidence=0.3)
for index, (name, kind) in enumerate(defaults)
],
layout_opportunities=["先确认房间数量和主要公共区，再进入风格设计。"],
risks=[reason or "房间名称、面积与墙体属性需要人工确认。"],
degraded=True,
)
def _fallback_directions(brief: DesignBrief) -> list[StyleDirection]:
preferred = "、".join(brief.preferred_styles) or "现代简约"
avoid = "、".join(brief.disliked_elements) or "避免过度装饰"
return [
StyleDirection(
id="clear-modern",
name="清透现代",
thesis=f"以{preferred}为基础，用低饱和中性色和清晰体块获得明亮、耐看的日常空间。",
keywords=["通透", "低饱和", "整洁体块"],
palette=[
ColorToken(name="雾白", hex="#E8E9E6", role="墙面"),
ColorToken(name="石墨灰", hex="#555B58", role="家具"),
ColorToken(name="苔绿", hex="#65796A", role="点缀"),
],
materials=["哑光乳胶漆", "浅灰石材", "烟熏木饰面"],
lighting="自然光优先，线性洗墙与低位落地灯补充层次",
prompt=f"清透现代住宅，{preferred}，低饱和，克制体块，{avoid}",
),
StyleDirection(
id="soft-natural",
name="柔和自然",
thesis="弱化硬边界，用温和木色、织物和漫反射光让公共区更松弛，适合长期居住。",
keywords=["松弛", "木质", "柔光"],
palette=[
ColorToken(name="浅岩灰", hex="#D9D5CD", role="墙面"),
ColorToken(name="橡木", hex="#A58D70", role="木作"),
ColorToken(name="森林绿", hex="#40584A", role="点缀"),
],
materials=["自然橡木", "亚麻织物", "细纹微水泥"],
lighting="窗边自然光与隐藏式间接光结合，色温保持统一",
prompt=f"柔和自然住宅，{preferred}，自然木材，亚麻，安静柔光，{avoid}",
),
StyleDirection(
id="graphic-contrast",
name="克制对比",
thesis="保持空间背景安静，用少量深色构件和艺术家具建立记忆点，画面更有设计感。",
keywords=["对比", "艺术家具", "干净线条"],
palette=[
ColorToken(name="冷白", hex="#ECEDEA", role="背景"),
ColorToken(name="炭黑", hex="#292D2B", role="构件"),
ColorToken(name="砖红", hex="#9A5547", role="点缀"),
],
materials=["冷灰涂料", "深色金属", "胡桃木"],
lighting="重点照明突出家具与材质，整体控制眩光",
prompt=f"克制对比住宅，{preferred}，冷白背景，深色构件，艺术家具，{avoid}",
),
]
async def create_style_directions(
values: dict[str, Any], brief: DesignBrief, structure: StructureAnalysis
) -> tuple[list[StyleDirection], str]:
fallback = _fallback_directions(brief)
room_names = "、".join(room.name for room in structure.rooms) or "户型房间待确认"
prompt = f"""你是资深住宅室内设计总监。根据以下信息生成三套差异明确但可落地的风格方向。
居住者：{brief.residents or '未填写'}
生活方式：{'、'.join(brief.lifestyle) or '未填写'}
重点空间：{'、'.join(brief.focus_rooms) or room_names}
偏好风格：{'、'.join(brief.preferred_styles) or '现代、自然'}
偏好颜色：{'、'.join(brief.preferred_colors) or '低饱和中性色'}
不喜欢：{'、'.join(brief.disliked_elements) or '过度装饰'}
保留项：{'、'.join(brief.must_keep) or '无'}
预算：{brief.budget_level}
补充：{brief.additional_notes or '无'}
只返回 JSON：{{"directions":[{{"id":"英文短标识","name":"中文名","thesis":"一句设计主张","keywords":["3项"],"palette":[{{"name":"颜色名","hex":"#RRGGBB","role":"用途"}}],"materials":["3项"],"lighting":"照明策略","prompt":"适合图像模型的中文提示词"}}]}}。必须恰好三套。"""
try:
response = await OpenAICompatibleGateway(values).chat([{"role": "user", "content": prompt}])
data = _json_from_model(response)
model_name = model_display_name(values, str(values.get("orchestrator_model_id", "")))
directions = []
for index, item in enumerate(data.get("directions", [])):
if not isinstance(item, dict):
continue
directions.append(
StyleDirection(
id=str(item.get("id") or f"direction-{index + 1}"),
name=str(item.get("name") or fallback[index].name),
thesis=str(item.get("thesis") or fallback[index].thesis),
keywords=[str(value) for value in item.get("keywords", [])][:5],
palette=[ColorToken.model_validate(value) for value in item.get("palette", [])][:5],
materials=[str(value) for value in item.get("materials", [])][:6],
lighting=str(item.get("lighting") or ""),
prompt=str(item.get("prompt") or fallback[index].prompt),
model_name=model_name,
)
)
if len(directions) == 3:
return directions, model_name
except Exception:
pass
return fallback, "内置设计策略（模型降级）"
async def generated_image_bytes(payload: dict[str, Any]) -> tuple[bytes, str]:
url = str(payload.get("result_url") or "")
data = payload.get("data")
if not url and isinstance(data, list) and data and isinstance(data[0], dict):
url = str(data[0].get("url") or "")
encoded = data[0].get("b64_json")
if encoded:
return base64.b64decode(str(encoded)), "image/png"
if url.startswith("data:image/"):
header, encoded = url.split(",", 1)
mime = header.split(";", 1)[0].replace("data:", "")
return base64.b64decode(encoded), mime
if not url:
raise ValueError("生图模型返回了成功响应，但没有图片地址或图片数据。")
async with httpx.AsyncClient(timeout=90, follow_redirects=True) as client:
response = await client.get(url)
response.raise_for_status()
content_type = response.headers.get("content-type", "image/jpeg").split(";", 1)[0]
if not content_type.startswith("image/") or len(response.content) < 1024:
raise ValueError("生图结果不是可读取的真实图片。")
return response.content, content_type
def render_id() -> str:
return f"render-{uuid4().hex[:12]}"
async def design_chat(
values: dict[str, Any],
*,
stage: str,
message: str,
brief: DesignBrief,
structure: StructureAnalysis,
selected_direction: StyleDirection | None,
) -> tuple[str, dict[str, Any]]:
context = {
"stage": stage,
"brief": brief.model_dump(),
"rooms": [room.model_dump() for room in structure.rooms],
"structure_summary": structure.summary,
"selected_direction": selected_direction.model_dump() if selected_direction else None,
}
prompt = f"""你是一个住宅风格设计 Agent，当前项目上下文：
{json.dumps(context, ensure_ascii=False)}
用户说：{message}
请判断信息是否足够。需要追问时只追问一个最关键问题；能够执行时说明你记录了什么，以及下一步该点击什么。
只返回 JSON：{{"reply":"自然、简短的中文回复","brief_updates":{{"residents":"可选","lifestyle":["可选"],"focus_rooms":["可选"],"preferred_styles":["可选"],"preferred_colors":["可选"],"disliked_elements":["可选"],"must_keep":["可选"],"budget_level":"可选","additional_notes":"可选"}}}}。
不要承诺施工准确性，不要虚构用户没有表达的偏好。"""
try:
response = await OpenAICompatibleGateway(values).chat([{"role": "user", "content": prompt}])
data = _json_from_model(response)
reply = str(data.get("reply") or "已记录，我会把它作为后续设计约束。")
updates = data.get("brief_updates") if isinstance(data.get("brief_updates"), dict) else {}
return reply, updates
except Exception:
note = brief.additional_notes.strip()
combined = f"{note}\n{message}".strip() if note else message
return "已把这条要求记入项目。你可以继续补充，或按当前页面的主按钮进入下一步。", {
"additional_notes": combined
}
services/api/tests/test_design_pipeline.py
+66
View File
@@ -0,0 +1,66 @@
import base64
import pytest
from app.domain.models import DesignBrief, StructureAnalysis
from app.services.design_pipeline import (
create_style_directions,
fallback_structure,
generated_image_bytes,
resolve_model_instance,
)
@pytest.mark.asyncio
async def test_manual_model_routing_uses_selected_pool_item() -> None:
values = {
"image_routing_mode": "manual",
"image_model_id": "image-2",
"model_pool": [
{
"id": "image-2",
"enabled": True,
"capabilities": ["image_generation"],
}
],
}
assert await resolve_model_instance(values, "image") == "image-2"
@pytest.mark.asyncio
async def test_generated_image_bytes_accepts_verified_inline_image() -> None:
content = b"\x89PNG\r\n\x1a\n" + b"x" * 2048
payload = {
"data": [
{
"url": "data:image/png;base64,"
+ base64.b64encode(content).decode("ascii")
}
]
}
image, content_type = await generated_image_bytes(payload)
assert image == content
assert content_type == "image/png"
@pytest.mark.asyncio
async def test_style_direction_generation_has_non_blocking_fallback() -> None:
brief = DesignBrief(preferred_styles=["现代简约"], disliked_elements=["复杂吊顶"])
directions, model_name = await create_style_directions({}, brief, StructureAnalysis())
assert len(directions) == 3
assert all(direction.prompt for direction in directions)
assert "降级" in model_name
def test_spatial_fallback_remains_editable() -> None:
analysis = fallback_structure("provider unavailable")
assert analysis.degraded is True
assert analysis.status == "needs_review"
assert len(analysis.rooms) >= 6
assert analysis.risks == ["provider unavailable"]