단일 루프로 묶는 Claude Tool Use — stop_reason 분기, 병렬 호출, 오류 데이터화까지

import anthropic
 
client = anthropic.Anthropic()
MAX_ITERATIONS = 25  # 예시 기준. 실제 상한은 파이프라인 특성에 맞춰 조정하세요.
 
def run_agentic_loop(initial_prompt: str, tools: list) -> str:
    messages = [{"role": "user", "content": initial_prompt}]
 
    for iteration in range(MAX_ITERATIONS):
        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        )
 
        messages.append({"role": "assistant", "content": response.content})
 
        if response.stop_reason == "end_turn":
            for block in response.content:
                if block.type == "text":
                    return block.text
            return ""
 
        elif response.stop_reason == "tool_use":
            tool_use_blocks = [b for b in response.content if b.type == "tool_use"]
            # execute_tools_parallel은 다음 섹션에서 정의합니다
            tool_results = asyncio.run(execute_tools_parallel(tool_use_blocks))
            messages.append({"role": "user", "content": tool_results})
 
        else:
            # max_tokens, stop_sequence, pause_turn 등
            raise RuntimeError(f"Unexpected stop_reason: {response.stop_reason}")
 
    raise RuntimeError(f"Max iterations ({MAX_ITERATIONS}) reached")

import asyncio
 
async def execute_single_tool(block) -> dict:
    try:
        result_content = await dispatch_tool(block.name, block.input)
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": result_content,
        }
    except Exception as e:
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "is_error": True,
            "content": f"Tool execution failed: {type(e).__name__}: {str(e)}",
        }
 
async def execute_tools_parallel(tool_use_blocks: list) -> list:
    tasks = [execute_single_tool(block) for block in tool_use_blocks]
    return await asyncio.gather(*tasks)

import anthropic
import asyncio
import json
 
client = anthropic.Anthropic()
 
TOOLS = [
    {
        "name": "fetch_price",
        "description": "경쟁사의 특정 제품 가격을 가져옵니다",
        "input_schema": {
            "type": "object",
            "properties": {
                "competitor": {
                    "type": "string",
                    "description": "경쟁사 식별자 (예: alpha, beta, gamma)",
                },
                "product_id": {
                    "type": "string",
                    "description": "조회할 제품 ID",
                },
            },
            "required": ["competitor", "product_id"],
        },
    },
    {
        "name": "search_news",
        "description": "특정 키워드로 최신 뉴스 기사를 검색합니다",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "검색 키워드"},
                "max_results": {
                    "type": "integer",
                    "description": "반환할 최대 결과 수",
                    "default": 5,
                },
            },
            "required": ["query"],
        },
    },
    {
        "name": "get_market_data",
        "description": "카테고리별 시장 통계와 트렌드 데이터를 가져옵니다",
        "input_schema": {
            "type": "object",
            "properties": {
                "category": {"type": "string", "description": "제품 카테고리"},
            },
            "required": ["category"],
        },
    },
]
 
 
# 개념적 예시. 실제 환경에서는 httpx.AsyncClient 등으로 외부 API를 호출합니다.
async def dispatch_tool(name: str, inputs: dict) -> str:
    if name == "fetch_price":
        await asyncio.sleep(0.3)
        return json.dumps({"competitor": inputs["competitor"], "price": 29900, "currency": "KRW"})
 
    elif name == "search_news":
        await asyncio.sleep(0.5)
        return json.dumps({"articles": [{"title": f"{inputs['query']} 관련 최신 뉴스", "sentiment": "neutral"}]})
 
    elif name == "get_market_data":
        await asyncio.sleep(0.4)
        return json.dumps({"category": inputs["category"], "market_size": "1.2조 원", "growth_rate": "8.3%"})
 
    raise ValueError(f"Unknown tool: {name}")
 
 
async def execute_single_tool(block) -> dict:
    try:
        content = await dispatch_tool(block.name, block.input)
        return {"type": "tool_result", "tool_use_id": block.id, "content": content}
    except Exception as e:
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "is_error": True,
            "content": f"{type(e).__name__}: {str(e)}",
        }
 
 
async def execute_tools_parallel(tool_use_blocks: list) -> list:
    return await asyncio.gather(*(execute_single_tool(b) for b in tool_use_blocks))
 
 
SYSTEM_PROMPT = (
    "당신은 경쟁 환경 분석 에이전트입니다. 여러 도구를 병렬로 호출해 데이터를 수집하세요.\n"
    "도구가 is_error로 실패하면: (1) 파라미터 오류로 보이면 수정 후 최대 1회 재시도, "
    "(2) 외부 API 장애로 보이면 대체 소스가 있는지 판단, (3) 복구가 어려우면 해당 데이터 "
    "누락을 명시하고 나머지로 분석을 완성하세요. 무한 재시도는 금지합니다."
)
 
 
async def run_pipeline(product_id: str, category: str) -> str:
    prompt = (
        f"제품 ID '{product_id}'에 대해 경쟁사 3곳(alpha, beta, gamma)의 가격을 조회하고, "
        f"'{category}' 카테고리 관련 최신 뉴스와 시장 데이터를 수집한 뒤 "
        "종합적인 경쟁 환경 분석을 제공해주세요."
    )
 
    messages = [{"role": "user", "content": prompt}]
    MAX_ITERATIONS = 25
 
    for iteration in range(MAX_ITERATIONS):
        response = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=4096,
            system=SYSTEM_PROMPT,
            tools=TOOLS,
            messages=messages,
        )
 
        messages.append({"role": "assistant", "content": response.content})
 
        if response.stop_reason == "end_turn":
            for block in response.content:
                if block.type == "text":
                    return block.text
            return ""
 
        elif response.stop_reason == "tool_use":
            tool_use_blocks = [b for b in response.content if b.type == "tool_use"]
            print(f"[{iteration+1}] Claude가 {len(tool_use_blocks)}개 도구 병렬 실행 요청")
 
            tool_results = await execute_tools_parallel(tool_use_blocks)
            messages.append({"role": "user", "content": tool_results})
 
        else:
            raise RuntimeError(f"Unexpected stop_reason: {response.stop_reason}")
 
    raise RuntimeError(f"Max iterations ({MAX_ITERATIONS}) exceeded")
 
 
if __name__ == "__main__":
    result = asyncio.run(run_pipeline(product_id="PROD-001", category="소비자가전"))
    print(result)

import httpx
 
# 잘못된 방식 — 루프 전체를 죽입니다
async def bad_tool_handler(block):
    result = await call_external_api(block.input)  # 실패 시 예외 발생
    return result
 
# 올바른 방식 — 오류를 데이터로 Claude에게 전달합니다
async def good_tool_handler(block) -> dict:
    try:
        content = await call_external_api(block.input)
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": content,
        }
    except httpx.TimeoutException:
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "is_error": True,
            "content": (
                f"API timeout for {block.name} with input {block.input}. "
                "Last successful call was over 2 hours ago. "
                "Consider retrying once or noting the missing data."
            ),
        }
    except Exception as e:
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "is_error": True,
            "content": f"Unexpected error: {type(e).__name__}: {str(e)}",
        }

async def execute_single_tool_with_timeout(block, timeout_seconds: float = 10.0) -> dict:
    try:
        content = await asyncio.wait_for(
            dispatch_tool(block.name, block.input),
            timeout=timeout_seconds,
        )
        return {"type": "tool_result", "tool_use_id": block.id, "content": content}
    except asyncio.TimeoutError:
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "is_error": True,
            "content": f"Tool '{block.name}' timed out after {timeout_seconds}s",
        }