Claude API 도구 사용과 프롬프트 체이닝으로 다단계 계약서 분석 파이프라인 구축하기

개요: 왜 다단계 파이프라인인가?

계약서 분석은 단순한 단일 프롬프트로 해결하기 어렵습니다. 조항 추출, 리스크 평가, 법률 용어 검증, 최종 보고서 생성 등 여러 전문 단계가 필요합니다. Claude API의 Tool Use와 **프롬프트 체이닝(Prompt Chaining)**을 결합하면, 각 단계에서 외부 도구를 호출하고 그 결과를 다음 단계로 전달하는 강력한 자동화 파이프라인을 구축할 수 있습니다. 이 가이드에서는 Python으로 4단계 계약서 리뷰 파이프라인을 처음부터 끝까지 구현합니다.

Step 1: 환경 설정 및 SDK 설치

먼저 필요한 패키지를 설치하고 API 키를 설정합니다. # Python 3.9 이상 필요 pip install anthropic>=0.39.0 pip install python-dotenv

.env 파일을 생성하여 API 키를 안전하게 관리합니다. # .env ANTHROPIC_API_KEY=YOUR_API_KEY

기본 클라이언트를 초기화합니다. import anthropic import json from dotenv import load_dotenv

load_dotenv() client = anthropic.Anthropic() MODEL = “claude-sonnet-4-6”

Step 2: 도구(Tool) 스키마 정의

파이프라인에서 사용할 세 가지 도구를 정의합니다. 각 도구는 Claude가 호출할 수 있는 함수 인터페이스입니다. tools = [ { "name": "extract_clauses", "description": "계약서 텍스트에서 주요 조항을 구조화된 형태로 추출합니다.", "input_schema": { "type": "object", "properties": { "contract_text": { "type": "string", "description": "분석할 계약서 전문" }, "clause_types": { "type": "array", "items": {"type": "string"}, "description": "추출할 조항 유형 목록 (예: 해지, 배상, 기밀유지)" } }, "required": ["contract_text", "clause_types"] } }, { "name": "assess_risk", "description": "추출된 조항의 법적 리스크를 평가합니다.", "input_schema": { "type": "object", "properties": { "clauses": { "type": "array", "items": {"type": "object"}, "description": "평가할 조항 목록" } }, "required": ["clauses"] } }, { "name": "generate_report", "description": "분석 결과를 기반으로 최종 리뷰 보고서를 생성합니다.", "input_schema": { "type": "object", "properties": { "risk_assessment": {"type": "object"}, "format": {"type": "string", "enum": ["summary", "detailed"]} }, "required": ["risk_assessment", "format"] } } ] ## Step 3: 도구 실행 핸들러 구현

Claude가 도구를 호출하면 실제로 실행될 로직을 정의합니다. def handle_tool_call(tool_name, tool_input): if tool_name == "extract_clauses": # 실제로는 NLP 파싱 또는 DB 조회 로직 구현 return { "extracted": [ {"type": "해지 조항", "content": tool_input["contract_text"][:200], "section": "7조"}, {"type": "배상 책임", "content": "손해배상 한도 계약금의 100%", "section": "12조"} ], "total_found": 2 } elif tool_name == "assess_risk": results = [] for clause in tool_input["clauses"]: results.append({ "clause_type": clause.get("type", "unknown"), "risk_level": "high" if "배상" in clause.get("type", "") else "medium", "recommendation": "법률 검토 권장" }) return {"assessments": results} elif tool_name == "generate_report": return {"report": "계약서 리뷰 완료", "status": "generated"} return {"error": "Unknown tool"} ## Step 4: 프롬프트 체이닝 오케스트레이터 구축

핵심 부분입니다. 각 단계의 출력을 다음 단계의 입력으로 자동 연결하며, Tool Use 루프를 처리합니다. def run_pipeline(contract_text): """다단계 계약서 분석 파이프라인 실행"""

# === 1단계: 조항 추출 ===
messages = [{
    "role": "user",
    "content": f"다음 계약서에서 해지, 배상, 기밀유지 조항을 추출해주세요. "
               f"extract_clauses 도구를 사용하세요.\n\n{contract_text}"
}]

step1_result = execute_with_tools(messages, "조항 추출")

# === 2단계: 리스크 평가 (이전 결과를 컨텍스트로 전달) ===
messages = [{
    "role": "user",
    "content": f"추출된 조항을 분석하여 리스크를 평가해주세요. "
               f"assess_risk 도구를 사용하세요.\n\n"
               f"추출 결과: {json.dumps(step1_result, ensure_ascii=False)}"
}]

step2_result = execute_with_tools(messages, "리스크 평가")

# === 3단계: 보고서 생성 ===
messages = [{
    "role": "user",
    "content": f"리스크 평가 결과를 바탕으로 상세 보고서를 생성해주세요. "
               f"generate_report 도구를 사용하세요.\n\n"
               f"평가 결과: {json.dumps(step2_result, ensure_ascii=False)}"
}]

final_result = execute_with_tools(messages, "보고서 생성")
return final_result

response = client.messages.create(
    model=MODEL,
    max_tokens=4096,
    tools=tools,
    messages=messages
)

# Tool Use 루프: Claude가 도구 호출을 멈출 때까지 반복
while response.stop_reason == "tool_use":
    tool_block = next(b for b in response.content if b.type == "tool_use")
    tool_result = handle_tool_call(tool_block.name, tool_block.input)
    
    print(f"  ↳ 도구 호출: {tool_block.name}")
    
    messages.append({"role": "assistant", "content": response.content})
    messages.append({
        "role": "user",
        "content": [{
            "type": "tool_result",
            "tool_use_id": tool_block.id,
            "content": json.dumps(tool_result, ensure_ascii=False)
        }]
    })
    
    response = client.messages.create(
        model=MODEL,
        max_tokens=4096,
        tools=tools,
        messages=messages
    )

final_text = next((b.text for b in response.content if hasattr(b, "text")), "")
print(f"  ✓ {step_name} 완료")
return {"step": step_name, "output": final_text, "raw": tool_result}</code></pre>

# 실행
contract = """소프트웨어 라이선스 계약서
제7조(계약 해지) 일방 당사자가 본 계약을 위반할 경우 30일 서면 통지 후 해지 가능...
제12조(배상 책임) 손해배상 한도는 계약금의 100%로 한다...
"""

result = run_pipeline(contract)
print(json.dumps(result, indent=2, ensure_ascii=False))