Claude Files API로 PDF·이미지를 한 번 올리고 멀티턴 대화에서 재사용하기

import anthropic
 
client = anthropic.Anthropic()  # ANTHROPIC_API_KEY 환경변수
 
# 1단계: 파일 업로드 (최초 1회)
with open("contract.pdf", "rb") as f:
    file_obj = client.beta.files.upload(
        file=("contract.pdf", f, "application/pdf"),
    )
file_id = file_obj.id
print(f"업로드 완료: {file_id}")
 
 
def extract_text(response) -> str:
    # content 블록 순서는 상황에 따라 달라질 수 있어서 타입으로 골라냄
    return next(b.text for b in response.content if b.type == "text")
 
 
# 2단계: 멀티턴 대화 — 첫 턴에만 문서 블록을 넣고, 이후엔 텍스트만 추가
conversation_history = []
 
 
def ask(question: str) -> str:
    if not conversation_history:
        user_content = [
            {
                "type": "document",
                "source": {"type": "file", "file_id": file_id},
                "title": "계약서",
                "citations": {"enabled": True},
            },
            {"type": "text", "text": question},
        ]
    else:
        user_content = [{"type": "text", "text": question}]
 
    conversation_history.append({"role": "user", "content": user_content})
 
    response = client.beta.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        messages=conversation_history,
        betas=["files-api-2025-04-14"],
    )
 
    answer = extract_text(response)
    conversation_history.append({"role": "assistant", "content": answer})
    return answer
 
 
print(ask("이 계약서의 계약 기간은 얼마인가요?"))
print(ask("해지 조건은 어떻게 되나요?"))
print(ask("위약금 조항을 요약해줘."))

with open("product_screenshot.png", "rb") as f:
    img_obj = client.beta.files.upload(
        file=("product_screenshot.png", f, "image/png"),
    )
img_id = img_obj.id
 
response = client.beta.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "source": {"type": "file", "file_id": img_id},
                },
                {"type": "text", "text": "이 UI에서 개선할 수 있는 부분을 알려줘."},
            ],
        }
    ],
    betas=["files-api-2025-04-14"],
)
print(next(b.text for b in response.content if b.type == "text"))

# 목록 조회
files = client.beta.files.list()
for f in files.data:
    print(f.id, f.filename, f.created_at)
 
# 삭제
client.beta.files.delete(file_id)

response = client.beta.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=2048,
    system=[
        {
            "type": "text",
            "text": "당신은 법무 문서 분석 전문가입니다.",
            "cache_control": {"type": "ephemeral"},
        }
    ],
    messages=[
        {
            "role": "user",
            "content": [
                {
                    "type": "document",
                    "source": {"type": "file", "file_id": file_id},
                    "cache_control": {"type": "ephemeral"},
                },
                {"type": "text", "text": "핵심 조항을 세 줄로 요약해줘."},
            ],
        }
    ],
    betas=["files-api-2025-04-14"],
)

import time
from pathlib import Path
import anthropic
 
client = anthropic.Anthropic()
 
# 1단계: PDF 일괄 업로드
pdf_dir = Path("./reports")
file_ids = {}
for pdf_path in pdf_dir.glob("*.pdf"):
    with open(pdf_path, "rb") as f:
        file_obj = client.beta.files.upload(
            file=(pdf_path.name, f, "application/pdf"),
        )
    file_ids[pdf_path.stem] = file_obj.id
 
# 2단계: 각 file_id로 배치 요청 구성
requests = []
for doc_name, fid in file_ids.items():
    requests.append({
        "custom_id": doc_name,
        "params": {
            "model": "claude-sonnet-4-6",
            "max_tokens": 1024,
            "messages": [
                {
                    "role": "user",
                    "content": [
                        {"type": "document", "source": {"type": "file", "file_id": fid}},
                        {"type": "text", "text": "핵심 수치와 결론을 200자 이내로 요약해줘."},
                    ],
                }
            ],
        },
    })
 
# 3단계: 배치 제출
batch = client.beta.messages.batches.create(
    requests=requests,
    betas=["files-api-2025-04-14"],
)
 
# 4단계: 완료 대기 및 결과 스트리밍 회수
while True:
    status = client.beta.messages.batches.retrieve(batch.id)
    if status.processing_status == "ended":
        break
    time.sleep(30)
 
for result in client.beta.messages.batches.results(batch.id):
    if result.result.type == "succeeded":
        message = result.result.message
        text = next(b.text for b in message.content if b.type == "text")
        print(result.custom_id, "->", text)

KNOWLEDGE_BASE = {
    "user_manual": "file_abc123",
    "faq": "file_def456",
    "policy": "file_ghi789",
}
 
 
def answer_support_query(user_query: str, doc_key: str) -> str:
    fid = KNOWLEDGE_BASE[doc_key]
    response = client.beta.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "document",
                        "source": {"type": "file", "file_id": fid},
                        "cache_control": {"type": "ephemeral"},
                    },
                    {"type": "text", "text": user_query},
                ],
            }
        ],
        betas=["files-api-2025-04-14"],
    )
    return next(b.text for b in response.content if b.type == "text")