truloop Wiki

다크 모드

아키텍처 ​

전체 아키텍처 ​

처리 흐름 상세 ​

이미지 분석 파이프라인 ​

FallbackModel ​

이미지 분석에 pydantic-ai의 FallbackModel을 사용하여 모델 장애에 대비합니다.

python
# 파이프(|) 구분 모델 목록에서 FallbackModel 생성
model_names = [name.strip() for name in settings.image_analysis_models.split("|")]
# 기본값: ["google/gemini-3-flash-preview", "google/gemini-2.5-flash"]

models = [OpenAIChatModel(name, provider=provider) for name in model_names]
self.model = FallbackModel(*models) if len(models) > 1 else models[0]

첫 번째 모델(gemini-3-flash-preview)이 실패하면 자동으로 두 번째 모델(gemini-2.5-flash)로 전환됩니다. 환경변수 IMAGE_ANALYSIS_MODELS로 런타임에 모델을 변경할 수 있습니다.

재시도 전략 ​

tenacity 라이브러리로 네트워크 오류에 대한 재시도를 수행합니다.

설정값
최대 시도 횟수3회
백오프지수 (min 2초, max 10초)
재시도 대상httpx.ConnectError, httpx.TimeoutException, httpx.RemoteProtocolError, APIConnectionError, APITimeoutError
HTTP 타임아웃120초
커넥션 제한max_keepalive=5, max_connections=10

병렬 처리 ​

asyncio.gather로 모든 이미지를 동시에 분석합니다. 개별 이미지 실패는 ImageAnalysisResult.error에 기록되고, 성공한 이미지만으로 스토리 생성을 진행합니다.

python
async def analyze_images(self, media_items, loop_metadata, locale=None):
    locale_config = get_locale_config(locale, timezone=loop_metadata.timezone)
    tasks = [self.analyze_image(item, loop_metadata, locale_config) for item in media_items]
    results = await asyncio.gather(*tasks)
    # 개별 실패는 error 필드에 기록, 예외를 전파하지 않음
    return results

Agent 구조 ​

pydantic-ai Agent는 두 레벨의 프롬프트로 구성됩니다:

프롬프트위치역할
system_promptAgent 생성 시 고정구조적 분석 지침 (언어 중립)
instructionsagent.run() 시 동적 전달로케일별 문화/톤 지침

user prompt에는 Loop 메타데이터(제목, 참가자명), 태깅된 사용자, 댓글 등의 컨텍스트가 포함됩니다.

스토리 생성 파이프라인 ​

모델 구성 ​

스토리 생성은 단일 모델(anthropic/claude-3.5-sonnet)을 사용합니다. FallbackModel을 사용하지 않습니다.

프롬프트 구조 ​

프롬프트내용
story_system_prompt출력 포맷 규칙 (text/media 블록 교차 패턴, 언어 중립)
story_instructions로케일별 톤/스타일 (예: 한국어 "20대 인스타 일상글" 톤)
story_user_prompt_template모임 정보 + 사진 설명 → 스토리 작성 요청

instructions는 story_system_prompt + story_instructions를 결합하여 전달됩니다.

출력 구조 ​

pydantic-ai Agent가 structured output으로 StoryGenerationOutput을 반환합니다:

python
class StoryGenerationOutput(BaseModel):
    title: str          # 최대 50자, 캐치한 제목
    preview_text: str   # 2-3문장, 최대 150자
    blocks: list[StoryBlockOutput]  # text/media 블록 배열

블록 재배치 로직 ​

Claude가 연속 media 블록을 생성하는 경우를 감지하여 자동으로 text-media 교차 패턴으로 재배치합니다.

python
def _convert_to_story_content(self, output, media_items):
    # 연속 media 블록 감지
    if self._has_consecutive_media_blocks(output.blocks):
        logger.warning("Consecutive media blocks detected. Reordering...")
        return self._reorder_blocks(output.blocks, media_items)
    return self._process_blocks_in_order(output.blocks, media_items)

재배치 전략:

  1. text 블록과 media 블록을 분리
  2. media 블록을 image_index 순으로 정렬
  3. text 블록을 media 블록 사이에 균등 분배 (text → media → text → media 패턴)

이미지 그룹핑 알고리즘 ​

group_and_reorder 함수가 타임스탬프 기반으로 이미지를 시간순으로 재배치합니다.

python
def group_and_reorder(
    media_items: list[MediaItem],
    analysis_results: list[ImageAnalysisResult],
    gap_minutes: int = 30,
) -> tuple[list[MediaItem], list[ImageAnalysisResult]]:
    # ...

타임스탬프 분리 ​

이미지를 타임스탬프가 있는 것과 없는 것으로 분리합니다. 타임스탬프가 전혀 없으면 원래 순서를 유지합니다.

시간순 정렬 ​

타임스탬프가 있는 이미지를 촬영 시간 순으로 정렬합니다.

갭 기반 그룹핑 ​

인접 이미지 간 시간 차이가 gap_minutes(기본 30분) 이상이면 새 그룹을 시작합니다.

타임스탬프 없는 이미지 추가 ​

타임스탬프가 없는 이미지는 원래 순서를 유지하여 마지막 그룹으로 추가합니다.

평탄화 ​

모든 그룹을 순서대로 평탄화하여 최종 정렬된 리스트를 반환합니다.

다국어 지원 아키텍처 ​

LocaleConfig ​

LocaleConfig 데이터클래스가 각 언어별 프롬프트, 레이블, 날짜 포맷을 완전 분리하여 관리합니다.

필드 카테고리필드 수용도
스토리 생성 프롬프트3system_prompt, instructions, user_prompt_template
이미지 분석 프롬프트2instructions, user_prompt_prefix
로컬라이즈 레이블10"모임 제목", "참가자" 등 프롬프트 내 레이블
날짜 포맷1strftime 포맷 문자열

resolve_locale 로직 ​

우선순위방법예시
1명시적 locale 값"ko", "ja", "en"
2locale 태그의 언어 prefix"en-US" → en, "ko_KR" → ko
3timezone fallback (레거시)Asia/Seoul → ko, Asia/Tokyo → ja, 기타 → en

에러 처리 흐름 ​

에러 처리 계층 ​

부분 이미지 실패 처리 ​

개별 이미지 분석 실패는 전체 처리를 중단하지 않습니다. ImageAnalysisResult.error에 에러를 기록하고, 성공한 이미지만으로 스토리 생성을 진행합니다. 스토리 생성 프롬프트에서 실패한 이미지는 [이미지 분석 실패] 레이블로 표시됩니다.

DLQ 핸들러 ​

dlqWorker는 3회 실패한 메시지를 수신하여 truloop-core API에 해당 Story를 FAILED로 마킹합니다.

python
def dlq_handler(event, context):
    for record in event.get("Records", []):
        body = json.loads(record["body"])
        story_eid = body.get("story_eid")
        # Story를 FAILED로 마킹
        asyncio.run(api_client.update_story_failure(
            request_id=body.get("request_id", "unknown"),
            story_eid=story_eid,
            error_message="Story generation failed after maximum retries",
        ))

truloop-core 연동 ​

Internal API 클라이언트 ​

TruloopCoreClient가 truloop-core의 Internal API와 통신합니다.

설정값
인증Authorization: Bearer {INTERNAL_API_KEY}
타임아웃전체 30초, 연결 10초
엔드포인트POST /core/internal/stories/{story_eid}/requests/{request_id}

성공과 실패 모두 동일한 엔드포인트를 사용하며, status 필드(completed / failed)로 구분합니다.

주의

update_story_failure 메서드의 API 호출 실패 시 예외를 전파하지 않습니다 (except 블록에서 로그만 기록). 이는 API 보고 실패가 Lambda 실패로 이어지지 않도록 의도한 설계이지만, handler.py에서는 이 메서드의 실패를 별도로 감지하여 success=False를 반환합니다.

변경 이력 ​

날짜내용
2026-03-11최초 작성 — truloop-story 아키텍처 상세 문서