354 lines
16 KiB
Python
354 lines
16 KiB
Python
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, safe_response_diagnostic
|
||
|
||
|
||
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:
|
||
diagnostic = json.dumps(safe_response_diagnostic(payload), ensure_ascii=False)
|
||
raise ValueError(f"平台返回体中没有可识别的图片结果。脱敏诊断：{diagnostic}")
|
||
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
|
||
}
|