File size: 3,696 Bytes
48d895c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """XAI image-edit protocol — payload builder and SSE field extractors."""
from typing import Any
from app.platform.config.snapshot import get_config
IMAGE_EDIT_MODEL_NAME = "imagine-image-edit"
IMAGE_EDIT_MODEL_KIND = "imagine"
IMAGE_POST_MEDIA_TYPE = "MEDIA_POST_TYPE_IMAGE"
IMAGE_EDIT_GENERATION_COUNT = 2
def build_image_edit_payload(
*,
prompt: str,
image_references: list[str],
parent_post_id: str,
) -> dict[str, Any]:
"""Build the JSON payload for imagine image-edit chat requests."""
cfg = get_config()
return {
"temporary": cfg.get_bool("features.temporary", True),
"modelName": IMAGE_EDIT_MODEL_NAME,
"message": prompt,
"enableImageGeneration": True,
"returnImageBytes": False,
"returnRawGrokInXaiRequest": False,
"enableImageStreaming": True,
"imageGenerationCount": IMAGE_EDIT_GENERATION_COUNT,
"forceConcise": False,
"enableSideBySide": True,
"sendFinalMetadata": True,
"isReasoning": False,
"disableTextFollowUps": True,
"responseMetadata": {
"modelConfigOverride": {
"modelMap": {
"imageEditModel": IMAGE_EDIT_MODEL_KIND,
"imageEditModelConfig": {
"imageReferences": image_references,
"parentPostId": parent_post_id,
},
}
}
},
"disableMemory": not cfg.get_bool("features.memory", False),
"forceSideBySide": False,
}
def extract_streaming_response(data: dict[str, Any]) -> dict[str, Any] | None:
"""Return ``response.streamingImageGenerationResponse`` when present."""
result = data.get("result")
if not isinstance(result, dict):
return None
response = result.get("response")
if not isinstance(response, dict):
return None
stream = response.get("streamingImageGenerationResponse")
return stream if isinstance(stream, dict) else None
def extract_model_response_urls(data: dict[str, Any]) -> list[str]:
"""Return fallback ``modelResponse.generatedImageUrls`` values."""
result = data.get("result")
if not isinstance(result, dict):
return []
response = result.get("response")
if not isinstance(response, dict):
return []
model_response = response.get("modelResponse")
if not isinstance(model_response, dict):
return []
urls = model_response.get("generatedImageUrls")
if not isinstance(urls, list):
return []
return [url for url in urls if isinstance(url, str) and url]
def extract_model_response_file_attachments(data: dict[str, Any]) -> list[str]:
"""Return fallback ``modelResponse.fileAttachments`` asset IDs."""
result = data.get("result")
if not isinstance(result, dict):
return []
response = result.get("response")
if not isinstance(response, dict):
return []
model_response = response.get("modelResponse")
if not isinstance(model_response, dict):
return []
attachments = model_response.get("fileAttachments")
if not isinstance(attachments, list):
return []
return [attachment for attachment in attachments if isinstance(attachment, str) and attachment]
__all__ = [
"IMAGE_EDIT_MODEL_NAME",
"IMAGE_EDIT_MODEL_KIND",
"IMAGE_POST_MEDIA_TYPE",
"IMAGE_EDIT_GENERATION_COUNT",
"build_image_edit_payload",
"extract_streaming_response",
"extract_model_response_urls",
"extract_model_response_file_attachments",
]
|