jhogyu/Liti-agent-Development
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
Liti-agent-Development/Caching_설계/컨텍스트_캐싱_Gemini.md
T

34 KiB
Raw Permalink Blame History

질문

내가 개발 중인 에이전트는 다중 작업 workflow를 가지고 있다. 특정 stage에서도 여러 작업들이 순차적 혹은 병렬적으로 실행된다.

다음을 가정하라: 특정 작업 stage에서 Task A와 Task B가 순차적으로 실행된다. Task A와 Task B에서 모두 공통된 LLM(gemini-3.1-flash-lite-preview)을 사용한다.

위의 가정 하에서 Task_B는 입력개체로서 Task A 입력의 일부 파일, 그리고 Task A의 결과물 파일을 입력값으로 사용한다. LLM 비용/속도 측면에서 Task A와 Task B 사이에서 google이 제공하는 context caching 작업을 끼워넣으려고 한다. 어떻게 작업을 하면 되는가?

답변

Task A와 Task B가 공통된 입력 파일을 공유하는 워크플로우라면, Gemini의 Context Caching을 도입하여 비용을 절감하고 응답 속도(TTFT, Time To First Token)를 높이기에 아주 이상적인 시나리오입니다.

Context Caching을 설계할 때 가장 중요하게 기억해야 할 점은 **"생성된 캐시는 불변(Immutable) 객체"**라는 것입니다. 즉, 한 번 만든 캐시에 나중에 Task A의 결과물을 '추가'할 수는 없습니다.

따라서 다음과 같은 워크플로우로 아키텍처를 설계하셔야 합니다.

💡 Context Caching 적용 워크플로우

1. 공통 입력 데이터 식별 및 캐시 생성 (Before Task A)

Task A와 Task B가 공통으로 사용하는 '일부 파일'들을 모아 하나의 캐시(Cache)로 생성합니다. 이때 캐시의 수명(TTL, Time To Live)을 해당 Stage가 실행되는 여유 시간(예: 10분~15분) 정도로 설정합니다.

2. Task A 실행

  • 입력: [1번에서 만든 캐시 객체] + [Task A에만 필요한 추가 파일/프롬프트]
  • 출력: Task A의 결과물 파일 (Output A)

3. Task B 실행

  • 입력: [1번에서 만든 동일한 캐시 객체] + [Task A의 결과물 (Output A)] + [Task B 전용 프롬프트]
  • 출력: Task B의 최종 결과물

4. 캐시 삭제 (After Task B)

작업 Stage가 끝나면 생성해 둔 캐시를 명시적으로 삭제합니다. (TTL이 지나면 자동 삭제되지만, 스토리지 유지 비용을 최소화하기 위해 즉시 삭제하는 것이 좋습니다.)

💻 Python 구현 논리 (의사 코드 및 핵심 API)

Google GenAI SDK를 사용할 경우 대략적인 작업 흐름은 다음과 같습니다.

Python

import google.generativeai as genai
from google.generativeai import caching
import datetime

# 1. 공통 파일 업로드
common_file = genai.upload_file("path/to/shared_input.pdf")

# 2. 공통 컨텍스트 캐시 생성 (Task A와 B가 공유할 데이터)
cache = caching.CachedContent.create(
    model='models/gemini-1.5-flash-002', # 혹은 사용하시는 모델
    display_name='shared_stage_cache',
    system_instruction='당신은 다중 작업 워크플로우의 AI 에이전트입니다.',
    contents=[common_file],
    ttl=datetime.timedelta(minutes=15), # 15분 후 자동 삭제
)

# 3. Task A 실행
model_A = genai.GenerativeModel.from_cached_content(cached_content=cache)
response_A = model_A.generate_content("Task A를 수행해줘. 이 파일에서 주요 키워드를 추출해.")
task_a_output = response_A.text

# 4. Task B 실행 (동일한 캐시 재사용 + Task A 결과물 주입)
model_B = genai.GenerativeModel.from_cached_content(cached_content=cache)
response_B = model_B.generate_content([
    f"이전 작업의 결과물은 다음과 같아: {task_a_output}",
    "이 결과물과 캐시된 문서를 바탕으로 Task B(최종 요약 보고서 작성)를 수행해줘."
])
task_b_output = response_B.text

# 5. 작업 완료 후 캐시 즉시 삭제 (비용 절감)
cache.delete()

(참고: gemini-3.1-flash-lite-preview와 같은 최신/프리뷰 모델의 경우, API 버전이나 SDK 업데이트 상태에 따라 캐싱 지원 여부 및 모델명이 다를 수 있으니 사용 전 공식 문서를 확인해 주세요.)

⚠️ 실무 적용 시 주의사항 (Gotchas)

  • 최소 토큰 수 제한: Gemini의 Context Caching은 보통 32,768 토큰 이상의 컨텍스트에 대해서만 작동합니다. 공유하는 입력 파일의 크기가 이보다 작다면 캐시 생성이 거부되거나, 오히려 일반 호출이 더 저렴하고 빠를 수 있습니다.
  • 스토리지 비용: 캐시된 토큰은 호출할 때마다 발생하는 입력(Input) 비용은 크게 줄여주지만, 캐시가 유지되는 동안 분/시간 단위로 '스토리지 보관 비용'이 발생합니다. Task B가 끝나는 즉시 캐시를 폐기(delete)하는 로직을 반드시 try-finally 구문 등으로 안전하게 구현해야 합니다.
  • Task A 출력물의 크기: Task A의 결과물은 캐시되지 않고 매번 Task B 호출 시 동적 입력 토큰으로 계산됩니다. 만약 Task A의 결과물이 너무 방대하고 이후 Task C, Task D에서도 계속 쓰인다면, Task B 단계에서 기존 캐시를 삭제하고 Task A 결과물까지 포함한 새로운 캐시를 굽는(Baking) 것이 나을 수도 있습니다.

현재 개발 중인 에이전트 환경(Python, Node.js 등)에 맞춰 더 구체적인 코드 예시나 에러 핸들링 로직을 작성해 드릴까요?

질문

에러 핸들링 로직을 포함한 Python 코드 예시를 작성해줘.

답변

실무 환경에서 에러 핸들링은 캐시 누수로 인한 불필요한 비용 발생을 막고, 예상치 못한 API 오류(예: 토큰 수 미달, 할당량 초과)에 유연하게 대처하기 위해 필수적입니다.

가장 중요한 부분은 작업 성공 여부와 관계없이 반드시 캐시와 업로드된 파일을 삭제하도록 finally 블록을 구성하는 것입니다.

아래는 Google GenAI SDK를 활용하여 에러 핸들링 로직을 적용한 완전한 형태의 Python 예시 코드입니다.

💻 Context Caching 통합 및 에러 핸들링 구현

Python

import os
import datetime
import google.generativeai as genai
from google.generativeai import caching
from google.api_core import exceptions

# API 키 설정 (환경 변수 사용 권장)
genai.configure(api_key=os.environ.get("GEMINI_API_KEY"))

def run_multi_task_workflow(shared_file_path: str):
    # 사용자가 지정한 모델
    model_name = "models/gemini-3.1-flash-lite-preview"
    
    uploaded_file = None
    cache = None

    try:
        # 1. 공통 파일 업로드
        print("1. 공통 파일 업로드 중...")
        uploaded_file = genai.upload_file(shared_file_path)
        
        # 2. Context Cache 생성
        print("2. Context Cache 생성 중...")
        cache = caching.CachedContent.create(
            model=model_name,
            display_name="workflow_shared_cache",
            system_instruction="당신은 다중 작업 워크플로우를 처리하는 AI 에이전트입니다.",
            contents=[uploaded_file],
            ttl=datetime.timedelta(minutes=15) # 15분 후 자동 삭제되도록 안전장치 설정
        )
        print(f"  -> 캐시 생성 완료 (이름: {cache.name})")

        # 3. Task A 실행
        print("\n3. Task A 실행 중...")
        model_a = genai.GenerativeModel.from_cached_content(cached_content=cache)
        response_a = model_a.generate_content("Task A: 이 문서의 핵심 요약 3가지를 추출해 주세요.")
        task_a_output = response_a.text
        print(f"  -> [Task A 결과]:\n{task_a_output}")

        # 4. Task B 실행 (캐시 재사용 + Task A 결과물 주입)
        print("\n4. Task B 실행 중...")
        model_b = genai.GenerativeModel.from_cached_content(cached_content=cache)
        prompt_b = [
            f"이전 작업(Task A)의 결과물은 다음과 같습니다:\n{task_a_output}\n\n",
            "Task B: 캐시된 원본 문서와 위의 Task A 결과물을 바탕으로, 최종 분석 보고서를 작성해 주세요."
        ]
        response_b = model_b.generate_content(prompt_b)
        task_b_output = response_b.text
        print(f"  -> [Task B 결과]:\n{task_b_output}")

        return task_b_output

    except exceptions.InvalidArgument as e:
        # 캐싱 최소 토큰 수(일반적으로 32k)를 채우지 못했을 때 자주 발생합니다.
        print(f"\n[오류] 잘못된 요청 (토큰 수 미달 등): {e}")
        # 필요하다면 여기서 캐싱을 사용하지 않는 일반 API 호출 로직으로 Fallback 할 수 있습니다.
        
    except exceptions.ResourceExhausted as e:
        print(f"\n[오류] API 할당량 초과: {e}")
        
    except exceptions.GoogleAPIError as e:
        print(f"\n[오류] Google API 통신 오류 발생: {e}")
        
    except Exception as e:
        print(f"\n[오류] 예기치 않은 시스템 오류 발생: {e}")

    finally:
        # 5. 리소스 정리 (성공/실패 여부와 무관하게 실행)
        print("\n5. 리소스 정리 중...")
        
        # 캐시 삭제 (스토리지 비용 발생 방지)
        if cache:
            try:
                cache.delete()
                print(f"  -> 캐시 삭제 완료: {cache.name}")
            except Exception as e:
                print(f"  -> [경고] 캐시 삭제 실패 (수동 확인 필요): {e}")
        
        # 파일 삭제 (File API 스토리지 정리)
        if uploaded_file:
            try:
                genai.delete_file(uploaded_file.name)
                print(f"  -> 업로드된 원본 파일 삭제 완료: {uploaded_file.name}")
            except Exception as e:
                print(f"  -> [경고] 파일 삭제 실패: {e}")

# 실행 예시
# workflow_result = run_multi_task_workflow("sample_document.pdf")

💡 코드 핵심 포인트

  • google.api_core.exceptions 활용: InvalidArgument 예외 처리를 통해 캐시 생성에 필요한 **최소 토큰 조건(보통 32,768 토큰)**을 만족하지 못했을 때 프로그램이 완전히 죽는 것을 방지합니다.
  • Fallback 전략 추가 가능성: InvalidArgument 예외가 발생한 경우, 파일 크기가 작아서 발생한 문제일 확률이 높으므로 이 예외 블록 안에서 캐시를 사용하지 않는 genai.GenerativeModel()을 직접 호출하도록 Fallback 로직을 짜는 것도 좋은 설계입니다.
  • 이중 try-except (Finally 블록 내부): 리소스 정리 중 하나(예: 캐시 삭제)가 실패하더라도 다른 하나(예: 파일 삭제)가 정상적으로 실행될 수 있도록 방어적으로 코딩했습니다.

에이전트가 처리하는 입력 파일의 크기가 작아서(32,000 토큰 미만) 캐시 생성에 실패할 경우를 대비해, 일반 API 호출로 자동 전환(Fallback)되는 로직도 추가로 작성해 드릴까요?

질문

에이전트가 처리하는 입력 파일의 크기가 작아서(32,000 토큰 미만) 캐시 생성에 실패할 경우를 대비해, 일반 API 호출로 자동 전환(Fallback)되는 로직도 추가로 작성하라.

답변

입력 문서의 토큰 수가 적어 캐시 생성이 거부되는 것은 실무 에이전트를 개발할 때 굉장히 자주 마주치는 상황입니다. 이를 예외 처리로 끝내지 않고 일반 API 호출로 자연스럽게 전환(Fallback) 되도록 설계하면, 파일 크기에 구애받지 않는 훨씬 안정적이고 유연한 워크플로우를 만들 수 있습니다.

이 Fallback 로직의 핵심은 캐시 생성에 실패했을 때 분기 플래그(예: use_cache = False)를 설정하고, 이후 Task A와 Task B에서 입력 프롬프트 배열에 원본 파일 객체를 직접 끼워 넣는 것입니다.

다음은 Fallback 메커니즘이 완벽하게 통합된 개선된 파이썬 코드입니다.

💻 Context Caching + 자동 Fallback 통합 코드

Python

import os
import datetime
import google.generativeai as genai
from google.generativeai import caching
from google.api_core import exceptions

# API 키 설정 (환경 변수 사용 권장)
genai.configure(api_key=os.environ.get("GEMINI_API_KEY"))

def run_multi_task_workflow_with_fallback(shared_file_path: str):
    model_name = "models/gemini-3.1-flash-lite-preview"
    system_instruction = "당신은 다중 작업 워크플로우를 처리하는 AI 에이전트입니다."
    
    uploaded_file = None
    cache = None
    use_cache = False # Fallback 분기를 위한 플래그

    try:
        # 1. 공통 파일 업로드 (캐싱 및 일반 호출 모두에 사용됨)
        print("1. 공통 파일 업로드 중...")
        uploaded_file = genai.upload_file(shared_file_path)
        
        # 2. Context Cache 생성 시도
        print("2. Context Cache 생성 시도 중...")
        try:
            cache = caching.CachedContent.create(
                model=model_name,
                display_name="workflow_shared_cache",
                system_instruction=system_instruction,
                contents=[uploaded_file],
                ttl=datetime.timedelta(minutes=15)
            )
            print(f"  -> 캐시 생성 완료 (이름: {cache.name})")
            use_cache = True # 캐시 생성 성공 시 플래그 활성화

        except exceptions.InvalidArgument as e:
            # 32k 토큰 미만 등의 이유로 캐시 생성이 거부된 경우 Fallback 실행
            print(f"  -> [알림] 파일 크기 미달 등으로 캐시 생성 생략. 일반 API 호출로 전환(Fallback)합니다. ({e})")
            use_cache = False

        # 3. Task A 실행
        print("\n3. Task A 실행 중...")
        task_a_prompt = "Task A: 이 문서의 핵심 요약 3가지를 추출해 주세요."
        
        if use_cache:
            # 캐시가 있을 경우: 파일은 이미 캐시에 포함되어 있으므로 텍스트 프롬프트만 전달
            model_a = genai.GenerativeModel.from_cached_content(cached_content=cache)
            response_a = model_a.generate_content(task_a_prompt)
        else:
            # Fallback (캐시 없음): 시스템 프롬프트를 포함하여 모델을 초기화하고, 파일 객체를 함께 전달
            model_a = genai.GenerativeModel(model_name=model_name, system_instruction=system_instruction)
            response_a = model_a.generate_content([uploaded_file, task_a_prompt])
            
        task_a_output = response_a.text
        print(f"  -> [Task A 결과]:\n{task_a_output}")

        # 4. Task B 실행
        print("\n4. Task B 실행 중...")
        task_b_instruction = "Task B: 원본 문서와 위의 Task A 결과물을 바탕으로, 최종 분석 보고서를 작성해 주세요."
        
        if use_cache:
            # 캐시가 있을 경우: 원본 문서는 캐시에 있으므로 Task A 결과와 Task B 지시문만 전달
            model_b = genai.GenerativeModel.from_cached_content(cached_content=cache)
            prompt_b = [
                f"이전 작업(Task A)의 결과물은 다음과 같습니다:\n{task_a_output}\n\n",
                task_b_instruction
            ]
            response_b = model_b.generate_content(prompt_b)
        else:
            # Fallback (캐시 없음): 원본 파일 객체 + Task A 결과 + Task B 지시문을 모두 전달
            model_b = genai.GenerativeModel(model_name=model_name, system_instruction=system_instruction)
            prompt_b = [
                uploaded_file, 
                f"\n\n이전 작업(Task A)의 결과물은 다음과 같습니다:\n{task_a_output}\n\n",
                task_b_instruction
            ]
            response_b = model_b.generate_content(prompt_b)
            
        task_b_output = response_b.text
        print(f"  -> [Task B 결과]:\n{task_b_output}")

        return task_b_output

    except Exception as e:
        print(f"\n[오류] 워크플로우 실행 중 예기치 않은 오류 발생: {e}")
        return None

    finally:
        # 5. 리소스 정리 (성공/실패/Fallback 여부와 무관하게 무조건 실행)
        print("\n5. 리소스 정리 중...")
        
        if cache:
            try:
                cache.delete()
                print(f"  -> 캐시 삭제 완료: {cache.name}")
            except Exception as e:
                print(f"  -> [경고] 캐시 삭제 실패: {e}")
        
        if uploaded_file:
            try:
                genai.delete_file(uploaded_file.name)
                print(f"  -> 업로드된 원본 파일 삭제 완료: {uploaded_file.name}")
            except Exception as e:
                print(f"  -> [경고] 파일 삭제 실패: {e}")

# 실행 예시
# result = run_multi_task_workflow_with_fallback("sample_document.pdf")

💡 로직 설계 포인트

  • 중복 업로드 방지: 파일 크기가 작더라도 File API(genai.upload_file)를 통해 업로드된 파일의 URI 객체는 일반 호출의 generate_content 배열 안에 그대로 사용할 수 있습니다. 따라서 캐싱이 실패하더라도 파일을 다시 읽거나 재업로드할 필요가 없습니다.
  • 시스템 프롬프트 유지: Fallback으로 전환되어 일반 GenerativeModel을 생성할 때, 캐시 생성 시 사용하려 했던 system_instruction을 잊지 않고 주입해주어야 에이전트의 페르소나가 유지됩니다.
  • 유연한 페이로드(Payload) 구성: generate_content는 리스트 형태로 여러 모달리티(파일 객체, 텍스트 등)를 순차적으로 받을 수 있습니다. Fallback 시에는 [uploaded_file, 텍스트_프롬프트] 형태로 구성하여 모델이 문맥을 온전히 이해하도록 구성했습니다.

코드를 에이전트 파이프라인에 이식하실 때, 각 Task의 출력(Output)을 단순히 문자열로 주고받는 것 외에 JSON(Structured Output) 형태로 파싱하는 로직이 추가로 필요하신가요?

질문

각 Task의 출력(Output)을 단순히 문자열로 주고받는 것 외에 JSON(Structured Output) 형태로 파싱하는 로직이 추가로 필요하다. 이를 반영하여 위의 python code 예제를 새로 작성하라.

답변

에이전트 워크플로우에서 각 Task의 결과를 단순 문자열이 아닌 JSON(Structured Output) 형태로 주고받는 것은 선택이 아닌 필수입니다. 결과를 파싱할 때 발생할 수 있는 환각(Hallucination)이나 포맷 오류를 방지하고, 다음 Task로 데이터를 안정적으로 넘기기 위해 아주 훌륭한 접근입니다.

Gemini API에서는 response_mime_type="application/json" 설정과 함께 **Pydantic 모델(스키마)**을 response_schema로 전달하여 출력 형태를 강제할 수 있습니다.

이전의 캐싱 및 Fallback 로직에 Pydantic을 활용한 JSON 구조화 로직을 추가하여 코드를 전면 개편했습니다.

💻 Context Caching + Fallback + JSON Structured Output 통합 코드

Python

import os
import json
import datetime
import google.generativeai as genai
from google.generativeai import caching
from google.api_core import exceptions
from pydantic import BaseModel, Field

# API 키 설정 (환경 변수 사용 권장)
genai.configure(api_key=os.environ.get("GEMINI_API_KEY"))

# ==========================================
# 📊 1. JSON 출력을 위한 Pydantic 스키마 정의
# ==========================================
class TaskAOutput(BaseModel):
    key_summaries: list[str] = Field(
        description="문서에서 추출한 핵심 요약 3가지",
        min_items=1,
        max_items=3
    )
    extracted_keywords: list[str] = Field(
        description="문서의 주요 키워드 목록"
    )

class TaskBOutput(BaseModel):
    report_title: str = Field(description="최종 분석 보고서의 제목")
    executive_summary: str = Field(description="Task A의 결과를 바탕으로 한 전체 요약")
    detailed_analysis: str = Field(description="원본 문서와 Task A 결과를 종합한 심층 분석 내용")
    conclusion: str = Field(description="최종 결론 및 인사이트")

# ==========================================
# ⚙️ 2. 메인 워크플로우 함수
# ==========================================
def run_structured_multi_task_workflow(shared_file_path: str):
    model_name = "models/gemini-3.1-flash-lite-preview"
    system_instruction = "당신은 다중 작업 워크플로우를 처리하고 반드시 지정된 JSON 포맷으로만 응답하는 AI 에이전트입니다."
    
    uploaded_file = None
    cache = None
    use_cache = False

    try:
        # [공통] 파일 업로드
        print("1. 공통 파일 업로드 중...")
        uploaded_file = genai.upload_file(shared_file_path)
        
        # [캐시] Context Cache 생성 시도
        print("2. Context Cache 생성 시도 중...")
        try:
            cache = caching.CachedContent.create(
                model=model_name,
                display_name="workflow_json_cache",
                system_instruction=system_instruction,
                contents=[uploaded_file],
                ttl=datetime.timedelta(minutes=15)
            )
            print(f"  -> 캐시 생성 완료 (이름: {cache.name})")
            use_cache = True
        except exceptions.InvalidArgument as e:
            print(f"  -> [알림] 파일 크기 미달로 일반 API 호출로 전환(Fallback)합니다.")
            use_cache = False

        # ==========================================
        # 🚀 3. Task A 실행 (JSON 출력 강제)
        # ==========================================
        print("\n3. Task A 실행 중 (JSON 파싱)...")
        task_a_prompt = "Task A: 이 문서의 핵심 요약 3가지와 주요 키워드를 추출해 주세요."
        
        # Task A를 위한 JSON Generation Config 설정
        config_a = genai.GenerationConfig(
            response_mime_type="application/json",
            response_schema=TaskAOutput
        )
        
        model_a = (genai.GenerativeModel.from_cached_content(cached_content=cache) if use_cache 
                   else genai.GenerativeModel(model_name=model_name, system_instruction=system_instruction))
        
        prompt_payload_a = [task_a_prompt] if use_cache else [uploaded_file, task_a_prompt]
        response_a = model_a.generate_content(prompt_payload_a, generation_config=config_a)
        
        # 응답 텍스트는 완벽한 JSON 문자열입니다. 파이썬 딕셔너리로 변환합니다.
        task_a_dict = json.loads(response_a.text)
        print(f"  -> [Task A JSON 결과]:\n{json.dumps(task_a_dict, indent=2, ensure_ascii=False)}")

        # ==========================================
        # 🚀 4. Task B 실행 (Task A JSON 주입 -> JSON 출력)
        # ==========================================
        print("\n4. Task B 실행 중 (JSON 파싱)...")
        
        # Task A의 파싱된 딕셔너리를 다시 문자열 형태로 깔끔하게 프롬프트에 주입
        task_b_instruction = f"""
        Task B: 원본 문서와 이전 작업(Task A)의 구조화된 결과물을 바탕으로 최종 분석 보고서를 작성해 주세요.
        
        [Task A 결과 데이터]:
        {json.dumps(task_a_dict, ensure_ascii=False)}
        """
        
        # Task B를 위한 JSON Generation Config 설정
        config_b = genai.GenerationConfig(
            response_mime_type="application/json",
            response_schema=TaskBOutput
        )
        
        model_b = (genai.GenerativeModel.from_cached_content(cached_content=cache) if use_cache 
                   else genai.GenerativeModel(model_name=model_name, system_instruction=system_instruction))
        
        prompt_payload_b = [task_b_instruction] if use_cache else [uploaded_file, task_b_instruction]
        response_b = model_b.generate_content(prompt_payload_b, generation_config=config_b)
        
        task_b_dict = json.loads(response_b.text)
        print(f"  -> [Task B JSON 결과]:\n{json.dumps(task_b_dict, indent=2, ensure_ascii=False)}")

        return task_b_dict

    except json.JSONDecodeError as e:
        print(f"\n[오류] JSON 파싱 실패 (모델이 JSON 형식을 어겼습니다): {e}")
        return None
    except Exception as e:
        print(f"\n[오류] 워크플로우 실행 중 예기치 않은 오류 발생: {e}")
        return None

    finally:
        # 5. 리소스 정리
        print("\n5. 리소스 정리 중...")
        if cache:
            try: cache.delete(); print(f"  -> 캐시 삭제 완료")
            except: pass
        if uploaded_file:
            try: genai.delete_file(uploaded_file.name); print(f"  -> 원본 파일 삭제 완료")
            except: pass

# 실행 예시
# final_json_report = run_structured_multi_task_workflow("sample_document.pdf")

💡 구조화된 출력(Structured Output) 적용 포인트

  1. response_schema의 강력함: Pydantic 클래스(TaskAOutput, TaskBOutput)를 genai.GenerationConfig에 넘겨주면, Gemini 모델이 백엔드 단에서 문법적으로 완벽한 JSON을 생성하도록 강제됩니다. 문자열 파싱 로직이나 정규표현식을 쓸 필요가 없어집니다.
  2. 에이전트 간 명확한 데이터 핸드오프(Handoff): Task A의 출력인 response_a.text는 순수한 JSON 문자열입니다. 이를 json.loads()로 파이썬 딕셔너리로 바꾼 뒤, Task B의 프롬프트에 json.dumps()를 통해 명확한 키-값(Key-Value) 형태로 주입했습니다. 이렇게 하면 Task B의 LLM이 데이터를 훨씬 정확하게 인식합니다.
  3. Field(description="...")의 활용: Pydantic의 description 속성은 단순한 주석이 아닙니다. 이 설명 자체가 내부적으로 프롬프트의 일부로 모델에게 전달되어, 각 필드에 어떤 내용을 채워야 하는지 모델을 직접적으로 가이드합니다.

이 코드를 실행하다 보면 가끔 모델이 스키마를 미세하게 어기거나(극히 드묾), 네트워크 지연이 발생할 수 있습니다. 안정성을 더 높이기 위해 재시도(Retry) 로직이나 Timeout 설정을 추가로 구성해 드릴까요?

질문

안정성을 더 높이기 위해 재시도(Retry) 로직이나 Timeout 설정을 추가로 구성하여 python 예제 코드를 재작성하라.

답변

실무 환경에서 LLM API를 호출할 때는 네트워크 지연(Timeout), 일시적인 서버 오류(500, 503), 혹은 API 호출 제한(Rate Limit) 등의 예기치 않은 상황이 빈번하게 발생합니다.

이를 방어하기 위해 **재시도(Retry) 알고리즘(Exponential Backoff 적용)**과 Timeout(요청 시간 초과) 설정을 통합한 최종 완성형 코드를 작성했습니다.

외부 라이브러리(예: tenacity) 없이도 바로 복사해서 사용할 수 있도록 파이썬 기본 모듈만 사용하여 구성했습니다.

💻 Context Caching + Fallback + JSON + Retry/Timeout 통합 코드

Python

import os
import json
import time
import datetime
import google.generativeai as genai
from google.generativeai import caching
from google.api_core import exceptions as google_exceptions
from pydantic import BaseModel, Field

# API 키 설정
genai.configure(api_key=os.environ.get("GEMINI_API_KEY"))

# ==========================================
# 📊 1. JSON 출력을 위한 Pydantic 스키마
# ==========================================
class TaskAOutput(BaseModel):
    key_summaries: list[str] = Field(description="문서에서 추출한 핵심 요약 3가지", min_items=1, max_items=3)
    extracted_keywords: list[str] = Field(description="문서의 주요 키워드 목록")

class TaskBOutput(BaseModel):
    report_title: str = Field(description="최종 분석 보고서의 제목")
    detailed_analysis: str = Field(description="원본 문서와 Task A 결과를 종합한 심층 분석 내용")

# ==========================================
# 🛡️ 2. 안정성을 위한 Retry & Timeout 래퍼 함수
# ==========================================
def generate_with_retry(model, prompt_payload, generation_config, max_retries=3, timeout_sec=60):
    """
    Timeout을 적용하고, 일시적인 네트워크/서버 오류 시 지수 백오프(Exponential Backoff)로 재시도합니다.
    """
    for attempt in range(max_retries):
        try:
            # request_options를 통해 타임아웃(초)을 설정합니다.
            response = model.generate_content(
                prompt_payload,
                generation_config=generation_config,
                request_options={"timeout": timeout_sec}
            )
            return response

        except (google_exceptions.ServiceUnavailable, 
                google_exceptions.DeadlineExceeded, 
                google_exceptions.InternalServerError,
                google_exceptions.TooManyRequests) as e:
            
            print(f"  -> [경고] API 통신 지연 또는 일시적 오류 (시도 {attempt + 1}/{max_retries}): {e}")
            if attempt < max_retries - 1:
                sleep_time = 2 ** attempt  # 1초, 2초, 4초 대기 (Exponential Backoff)
                print(f"  -> {sleep_time}초 후 재시도합니다...")
                time.sleep(sleep_time)
            else:
                print("  -> [실패] 최대 재시도 횟수를 초과했습니다.")
                raise e # 최종 실패 시 예외를 던져 워크플로우 중단

# ==========================================
# ⚙️ 3. 메인 워크플로우 함수
# ==========================================
def run_robust_workflow(shared_file_path: str):
    model_name = "models/gemini-3.1-flash-lite-preview"
    system_instruction = "당신은 다중 작업 워크플로우를 처리하고 반드시 지정된 JSON 포맷으로 응답하는 AI입니다."
    
    uploaded_file = None
    cache = None
    use_cache = False

    try:
        # [공통] 파일 업로드
        print("1. 파일 업로드 중...")
        uploaded_file = genai.upload_file(shared_file_path)
        
        # [캐시] Context Cache 생성 (15분 유지)
        print("2. 캐시 생성 시도 중...")
        try:
            cache = caching.CachedContent.create(
                model=model_name,
                display_name="robust_workflow_cache",
                system_instruction=system_instruction,
                contents=[uploaded_file],
                ttl=datetime.timedelta(minutes=15)
            )
            use_cache = True
            print(f"  -> 캐시 성공: {cache.name}")
        except google_exceptions.InvalidArgument:
            use_cache = False
            print("  -> [Fallback] 토큰 수 미달로 일반 호출로 전환합니다.")

        # ==========================================
        # 🚀 4. Task A 실행 (Retry + Timeout + JSON)
        # ==========================================
        print("\n3. Task A 실행 중...")
        task_a_prompt = "Task A: 이 문서의 핵심 요약 3가지와 주요 키워드를 추출해 주세요."
        config_a = genai.GenerationConfig(response_mime_type="application/json", response_schema=TaskAOutput)
        
        model_a = (genai.GenerativeModel.from_cached_content(cached_content=cache) if use_cache 
                   else genai.GenerativeModel(model_name=model_name, system_instruction=system_instruction))
        
        payload_a = [task_a_prompt] if use_cache else [uploaded_file, task_a_prompt]
        
        # 래퍼 함수를 통해 안전하게 호출 (Timeout 60초, 재시도 3회)
        response_a = generate_with_retry(model_a, payload_a, config_a, max_retries=3, timeout_sec=60)
        task_a_dict = json.loads(response_a.text)
        print("  -> [Task A 완료]")

        # ==========================================
        # 🚀 5. Task B 실행 (Retry + Timeout + JSON)
        # ==========================================
        print("\n4. Task B 실행 중...")
        task_b_prompt = f"""
        Task B: 원본 문서와 이전 작업(Task A)의 결과물을 바탕으로 최종 분석 보고서를 작성해 주세요.
        [Task A 결과]: {json.dumps(task_a_dict, ensure_ascii=False)}
        """
        config_b = genai.GenerationConfig(response_mime_type="application/json", response_schema=TaskBOutput)
        
        model_b = (genai.GenerativeModel.from_cached_content(cached_content=cache) if use_cache 
                   else genai.GenerativeModel(model_name=model_name, system_instruction=system_instruction))
        
        payload_b = [task_b_prompt] if use_cache else [uploaded_file, task_b_prompt]
        
        # 래퍼 함수를 통해 안전하게 호출 (Timeout 90초 설정 - Task B가 더 오래 걸릴 것을 가정)
        response_b = generate_with_retry(model_b, payload_b, config_b, max_retries=3, timeout_sec=90)
        task_b_dict = json.loads(response_b.text)
        print("  -> [Task B 완료]")

        return task_b_dict

    except Exception as e:
        print(f"\n[치명적 오류] 워크플로우 중단: {e}")
        return None

    finally:
        # [정리] 리소스 해제
        print("\n5. 리소스 정리 중...")
        if cache:
            try: cache.delete(); print("  -> 캐시 삭제 완료")
            except: pass
        if uploaded_file:
            try: genai.delete_file(uploaded_file.name); print("  -> 파일 삭제 완료")
            except: pass

# 실행 예시
# result = run_robust_workflow("sample_document.pdf")

💡 주요 개선 사항 요약

  1. request_options={"timeout": timeout_sec} 적용:
    • Gemini API 호출 시 응답이 무한정 길어지거나 연결이 끊겨 에이전트가 멈추는(Hang) 현상을 방지합니다.
    • 복잡한 추론이 들어가는 Task B는 시간을 더 넉넉하게(90초) 주는 등 유연한 설정이 가능해졌습니다.
  2. generate_with_retry 헬퍼 함수 구현:
    • DeadlineExceeded(타임아웃), ServiceUnavailable(서버 점검/일시 오류), TooManyRequests(초당 호출 제한 초과) 등 복구 가능한 오류만 필터링하여 재시도합니다.
    • 지수 백오프(Exponential Backoff): 첫 실패 시 1초 대기, 두 번째 2초 대기, 세 번째 4초 대기하는 방식으로 서버 부하를 줄이면서 성공 확률을 높입니다.
  3. 오류 전파 최적화:
    • 최대 재시도 횟수를 넘기면 강제로 예외(raise e)를 발생시켜, 비정상적인 상태로 워크플로우가 계속 진행되어 불필요한 비용이 발생하는 것을 차단합니다.
Reference in New Issue View Git Blame Copy Permalink