Gemini 2.5 Pro Code Execution, 백엔드에서 타입 안전하게 다루는 법

from google import genai
from google.genai.types import Tool, ToolCodeExecution, GenerateContentConfig
 
client = genai.Client()
 
response = client.models.generate_content(
    model="gemini-2.5-pro",
    contents="판다스로 숫자 리스트의 평균과 표준편차를 계산하는 코드를 실행해줘.",
    config=GenerateContentConfig(
        tools=[Tool(code_execution=ToolCodeExecution())]
    ),
)

from google.genai.types import Outcome
 
def parse_code_execution_response(response):
    result = {
        "texts": [],
        "codes": [],
        "outputs": [],
        "final_outcome": None,
    }
 
    for part in response.candidates[0].content.parts:
        if part.text is not None:
            result["texts"].append(part.text)
 
        elif part.executable_code is not None:
            result["codes"].append({
                "language": part.executable_code.language,
                "code": part.executable_code.code,
            })
 
        elif part.code_execution_result is not None:
            outcome = part.code_execution_result.outcome
            output = part.code_execution_result.output
 
            result["outputs"].append({
                "outcome": outcome,
                "output": output,
            })
            result["final_outcome"] = outcome
 
    return result
 
 
parsed = parse_code_execution_response(response)
 
# match 구문은 Python 3.10 이상에서 사용 가능합니다.
match parsed["final_outcome"]:
    case Outcome.OUTCOME_OK:
        print("성공:", parsed["outputs"][-1]["output"])
    case Outcome.OUTCOME_FAILED:
        print("실행 실패. 마지막 출력:", parsed["outputs"][-1]["output"])
    case Outcome.OUTCOME_DEADLINE_EXCEEDED:
        print("실행 시간 초과. 연산량을 줄이거나 입력 데이터를 분할하는 것을 고려해보세요.")
    case Outcome.OUTCOME_UNSPECIFIED | None:
        print("상태가 완결되지 않았거나 알 수 없는 응답입니다.")
    case _:
        print("알 수 없는 상태:", parsed["final_outcome"])

import json
from google import genai
from google.genai.types import (
    Tool, ToolCodeExecution, GenerateContentConfig, Outcome,
)
 
client = genai.Client()
 
 
def _strip_markdown_fence(text: str) -> str:
    # 모델이 JSON 앞뒤에 ```json ... ``` 를 붙이는 경우가 잦습니다.
    # 이 로직은 단일 코드 펜스만 안전하게 처리합니다.
    # 중첩 펜스나 본문 내 백틱 3연발이 있는 경우엔 오작동할 수 있습니다.
    stripped = text.strip()
    if stripped.startswith("```"):
        stripped = stripped.split("\n", 1)[1] if "\n" in stripped else stripped
        stripped = stripped.rsplit("```", 1)[0]
    return stripped.strip()
 
 
def analyze_csv_data(csv_content: str, user_question: str) -> dict:
    prompt = f"""다음 CSV 데이터를 분석해서 질문에 답해줘.
결과는 반드시 JSON 형태로 출력해.
 
CSV 데이터:
{csv_content}
 
질문: {user_question}"""
 
    response = client.models.generate_content(
        model="gemini-2.5-pro",
        contents=prompt,
        config=GenerateContentConfig(
            tools=[Tool(code_execution=ToolCodeExecution())]
        ),
    )
 
    parsed = parse_code_execution_response(response)
    if not parsed["outputs"]:
        return {"status": "error", "outcome": "no_result", "output": ""}
 
    last = parsed["outputs"][-1]
    if last["outcome"] != Outcome.OUTCOME_OK:
        return {
            "status": "error",
            "outcome": str(last["outcome"]),
            "output": last["output"],
        }
 
    raw_output = _strip_markdown_fence(last["output"])
    try:
        data = json.loads(raw_output)
    except json.JSONDecodeError as e:
        # 모델이 유효한 JSON을 생성하지 않은 경우. 호출자가 재시도 또는 폴백을 결정합니다.
        return {
            "status": "error",
            "outcome": "invalid_json",
            "output": raw_output,
            "error": str(e),
        }
 
    return {"status": "ok", "data": data}

import time
import google.api_core.exceptions
 
 
def call_with_backoff(client, model, contents, config, max_retries=4):
    delay = 1.0
    for attempt in range(max_retries):
        try:
            return client.models.generate_content(
                model=model,
                contents=contents,
                config=config,
            )
        except google.api_core.exceptions.ResourceExhausted:
            if attempt == max_retries - 1:
                raise
            time.sleep(delay)
            delay *= 2

import asyncio
 
 
async def call_with_backoff_async(client, model, contents, config, max_retries=4):
    delay = 1.0
    for attempt in range(max_retries):
        try:
            return await client.aio.models.generate_content(
                model=model,
                contents=contents,
                config=config,
            )
        except google.api_core.exceptions.ResourceExhausted:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(delay)
            delay *= 2

def save_inline_images(response, prefix="chart"):
    saved = []
    for idx, part in enumerate(response.candidates[0].content.parts):
        if part.inline_data is not None and part.inline_data.data:
            mime = part.inline_data.mime_type or "application/octet-stream"
            ext = "png" if mime.endswith("png") else "bin"
            path = f"{prefix}_{idx}.{ext}"
            with open(path, "wb") as f:
                f.write(part.inline_data.data)
            saved.append((path, mime))
    return saved