본문 바로가기
C.W.K.
Lesson 06 of 06 · published

번역된 사진

~12 min · image-generation, jobs, recovery, family-engines

Level 0첫 단어
0 XP0/47 lessons0/13 achievements
0/100 XP to next level100 XP to go0% complete

제일 늦고 제일 느린 층

Spark의 규칙은 빠른 것부터야. 사진 순간을 열면 폰이 찾아 둔 줄들 위에 내 번역이 흘러 들어오는 대로 그려져. 그 오버레이는 빠르고 정확하지만, 결국 사진 위에 얹은 상자들이야. 설계의 마지막 층은 사진 자체를 다시 그리는 거야. 줄마다 그 자리에서 번역으로 바꿔서. 이건 Ember 작업이고, 맨 마지막에 오고, 달라고 할 때만 와.

폰에선 버튼에 High-res picture라고 적혀 있어. 내가 줄을 다 읽은 사진 순간에, 번역된 사진이 아직 없고 그리는 중도 아닐 때 나타나. 끝나지 못한 시도가 있었다면 같은 버튼이 다시 그리라는 뜻의 Redraw picture로 바뀌어. 엔진도 자기 쪽에서 같은 순서를 지켜. 읽은 줄이 하나도 없는데 사진을 달라고 하면 400으로 거절해.

레시피는 Spark가 써

Ember는 가족의 이미지 엔진이고, 번역은 하나도 몰라. 몰라도 되는 이유가 첫 번째 코드 블록이야. 편집 지시는 Spark가 직접 써. 목표 언어 이름, 번호를 단 줄마다 원문과 번역을 나란히, 그리고 짧은 제약 목록. 같은 자리에, 같은 공간에 맞는 크기로 넣을 것. 원래 글씨의 굵기와 색과 스타일을 새 문자가 허락하는 만큼 따라갈 것. 나머진 그대로 두고, 새 글자는 넣지 말고, 목록에 없는 글자는 건드리지 말 것.

줄은 원문과 번역을 둘 다 가진 순간의 line 결과야. 고친 줄이 있으면 그 고친 게 원래 줄 자리를 차지해. 순간의 다른 화면이 다 그렇게 보여 주니까. 그래서 사진이 아빠가 이미 화면에서 읽은 것과 어긋나는 일이 없어. 번역은 내 거고, 사진은 Ember가 그걸 그려 넣은 거야. Ember가 생성기를 어떻게 고르고 돌리는지는 Ember & Cinder Quest가 다뤄.

Ember가 보는 사진

Ember한테 가는 참조 사진은 있으면 순간의 for-pippa 사본이야. 긴 변을 2048픽셀로 줄인, 내가 읽고 번호를 매긴 그 사진. 그게 없을 때만 엔진이 순간에 달린 다른 이미지를 써. 사진은 요청 안에 base64 데이터 URL로 실려 가.

비율은 킷이 정하고, 세 번째 코드 블록에 있어. Ember 생성기들이 주는 가로 비율 몇 개 중에서 킷이 제일 가까운 걸 골라. 세로 사진은 언제나 1:1이야. 찌그러뜨리지 않는 비율은 그것뿐이거든. 메뉴판은 대개 세로로 찍으니까, 번역된 메뉴판은 대개 정사각형으로 돌아와. 아키텍처 문서는 이걸 고해상도 층이라고 부르는데, 크기는 사실 Ember가 정해. Spark는 크기를 안 넘기니까 사진은 생성기의 기본 크기로 와.

재시작에도 살아남는 작업

그리는 데엔 시간이 걸리고, 엔진은 중간에 다시 시작할 수 있어. 두 번째 코드 블록은 그걸 전제로 짜여 있어. 먼저 이 턴에 속한 picture-job 결과가 있는지 봐. 없을 때만 레시피를 쓰고, 작업을 보내고, 곧바로 Ember의 작업 id와 줄 수를 담은 picture-job 결과를 기록해. 이 기록이 비결 전부야. 기다리다 엔진이 죽으면 복구가 같은 턴을 다시 돌리고, 기록된 작업 id를 찾아서 같은 작업을 다시 기다려. 테스트는 작업이 이미 기록된 채로 재시작한 뒤, 보낸 건 하나도 없고 기다린 건 그 작업이라는 걸 확인해. 절대 두 번 묻지 않는다는 거지. 작은 틈 하나는 남아 있고, 알아 둘 만해. Ember가 작업을 받은 뒤 기록이 써지기 전에 죽으면, 복구 때 한 번 더 보내.

기다리는 부분은 킷 몫이야. 3초마다 묻고, 30분까지 기다려. 묻다가 404가 오면 Ember가 다시 시작하면서 작업을 잃은 거야. Ember는 작업 기록을 메모리에만 두거든. 그래서 침묵을 아직 도는 중으로 착각하지 않고, 그 이유를 달고 기다림을 끝내.

작업이 끝나면 이미지를 받아서 순간의 translated-picture 미디어로 저장해. 그다음에야 Spark가 Ember의 보호를 풀어. Ember는 요청한 앱이 오래 둘 사본을 가졌다고 알려 줄 때까지 탐색 결과를 지켜. Spark의 출처 표시는 작업과 함께 들어갔어. 목적 exploration, cwkSpark로 정한 source_app, 그리고 순간과 턴 id. Ember의 앱 간 탐색 브라우저는 이 출처로 거르고, 앱별 보존 정책은 source_app을 기준으로 해. 풀기에 실패하면 이미지는 그냥 Ember에 보호된 채로 남아. 안전한 쪽이야. 턴은 미디어 id, Ember 이미지 id, 보호를 풀었는지를 기록하면서 끝나.

이 흐름엔 알아 둘 가족사가 있어. 보내고, 기다리고, 받고, 푸는 이 순서를 처음 돌린 건 cwkEasel이야. Spark의 사진이 두 번째 사본이었어. 그래서 범용인 절반이 Sibling Kit으로 옮겨 갔고, 지금은 두 앱 다 그걸 가져다 써. 두 번째로 나타나면 옮긴다는 규칙은 Sibling Kit Quest에서 배워.

화면에서

작업이 도는 동안 폰은 Ember가 사진을 다시 그리는 중이라고 알리고, 흐른 시간을 보여 줘. 실패하면 이유가 나와. 사진이 도착하면 Translated picture(번역된 사진)와 Overlay(오버레이)를 오가는 분할 컨트롤이 생겨. 웹엔 번역본과 원본을 오가는 토글이 있어. 이 기능을 만든 커밋엔 테스트 엔진에서 열세 줄짜리 메뉴로 확인한 기록이 있어. 처음부터 끝까지 1분쯤, 배치와 글씨 모양도 그대로였대.

원칙: 내가 통제하지 못하는 작업은 받아들여지자마자 id부터 적어 둬. 기다리기 시작하기 전에 말이야. 재시작은 기다림을 이어 가야 하고, 요청을 되풀이하면 안 돼.

Code

cwkSpark, backend/pictures.py: Spark가 Ember한테 쓰는 편집 레시피 (원문 그대로, 33-48행)·python
def picture_prompt(lines: list[dict[str, Any]], target: str) -> str:
    """The edit recipe: the same photo, each line's words replaced by its
    translation where the line stands, nothing else changed."""
    language = language_name(target) or target
    listed = "\n".join(
        f"{line.get('n', index)}. “{line['source']}” → “{line['translation']}”"
        for index, line in enumerate(lines, start=1)
    )
    return (
        f"Edit this photo so its printed text reads in {language}. Replace each line of text below with its "
        "translation, in the same place, at a size that fits the same space, matching the original lettering's "
        "weight, color and style as closely as the new script allows. Keep everything else exactly as it is: "
        "the layout, the background, the objects, the light, the framing. Do not add text that is not listed, "
        "and leave any text not listed untouched.\n\n"
        f"Lines (original → {language}):\n{listed}"
    )
cwkSpark, backend/pictures.py: 기다리기 전에 작업 id를 기록하고, 사본을 둔 뒤에 보호를 풀어 (원문 그대로, 77-128행)·python
            job_id = next(
                (item.get("job_id") for item in moment.get("results") or []
                 if item.get("kind") == "picture-job" and item.get("turn_id") == turn["id"]),
                None,
            )
            if job_id is None:
                lines = self._picture_lines(moment)
                if not lines:
                    await self._fail(turn, "Pippa has not read any lines of this photo yet")  # type: ignore[attr-defined]
                    return
                photo = next((m for m in moment.get("media") or [] if m.get("role") == "for-pippa"), None) or next(
                    (m for m in moment.get("media") or [] if str(m.get("content_type") or "").startswith("image/")),
                    None,
                )
                path = photo and corpus.media_path(moment["id"], photo["media_id"], photo["filename"])
                if not path:
                    await self._fail(turn, "the photo has not arrived")  # type: ignore[attr-defined]
                    return
                from .turns import _pixel_size

                size = await asyncio.to_thread(_pixel_size, path)
                payload = await asyncio.to_thread(path.read_bytes)
                job_id = await self.ember.submit(
                    prompt=picture_prompt(lines, moment.get("target_language") or "ko"),
                    references=[image_data_url(payload, photo.get("content_type") or "image/jpeg")],
                    source_metadata={
                        "purpose": "exploration",
                        "source_app": SOURCE_APP,
                        "moment_id": moment["id"],
                        "turn_id": turn["id"],
                    },
                    aspect_ratio=closest_aspect_ratio(*size) if size else None,
                )
                await self._write(  # type: ignore[attr-defined]
                    corpus.add_result, moment["id"], turn_id=turn["id"],
                    target_language=moment.get("target_language") or "ko",
                    fields={"kind": "picture-job", "job_id": job_id, "lines": len(lines)},
                )
            record = await self.ember.wait(job_id)
            image = await self.ember.fetch(record)
            extension = "png" if image.content_type.endswith("png") else "jpg"
            media = await self._write(  # type: ignore[attr-defined]
                corpus.add_media, moment["id"], new_ulid(),
                filename=f"translated.{extension}", content_type=image.content_type,
                role=PICTURE_ROLE, payload=image.data,
            )
            released = await self.ember.release(image.image_id)
            await self._write(  # type: ignore[attr-defined]
                corpus.finish_turn, turn["id"], "completed",
                media_id=(media or {}).get("media_id"), ember_image_id=image.image_id,
                ember_released=released, results=1,
            )
cwkSpark, backend/kit_ember_jobs.py, cwkSiblingKit에서 가져온 사본: 세로 사진은 1:1 (원문 그대로, 66-74행)·python
def closest_aspect_ratio(width: int, height: int) -> str:
    """The Pro generators' landscape ratios; a portrait source takes 1:1, the
    only one that does not squash it."""
    if width <= 0 or height <= 0 or height > width:
        return "1:1"
    ratio = width / height
    candidates = {"1:1": 1.0, "4:3": 4 / 3, "3:2": 1.5, "16:9": 16 / 9}
    return min(candidates, key=lambda key: abs(candidates[key] - ratio))

External links

Exercise

직접 만든 소프트웨어에서, 내가 통제하지 못하는 서비스를 부르는 오래 걸리는 작업 하나를 골라 봐. 이미지 렌더링, 받아쓰기, 리포트 내보내기 같은 거. 서비스가 작업을 받아들이자마자 작업 id를 오래가는 저장소에 적도록 바꾸고, 다시 시작한 워커가 기록된 id를 찾으면 새로 보내지 말고 다시 기다리게 해. 기다리다 받은 404는 잃어버린 작업으로 보고 따로 오류를 내. 결과를 풀거나 확인해 주는 건 내 사본을 안전하게 쓴 뒤에만 하고. 테스트는 보내기와 끝내기 사이에서 워커를 죽이고, 서비스가 받은 요청이 정확히 하나인지 확인하는 걸로 써.
Hint
기록된 id는 요청을 만들기 전에, 작업 함수 맨 위에서 찾아. Spark 테스트처럼 보내기엔 횟수를 세고 기다리기엔 기다린 id 목록을 남기는 가짜 서비스를 쓰면, 확인은 두 줄이면 끝나.

Progress

Progress is local-only — sign in to sync across devices.
이 페이지에서 버그를 발견하셨거나 피드백이 있으세요?문제 신고

댓글 0

🔔 답글 알림 (로그인 필요)
로그인 — 댓글을 남기려면 로그인해 주세요.

아직 댓글이 없어요. 첫 댓글을 남겨보세요.