Building
Effective Agents

from anthropic import Anthropic

client = Anthropic()

def augmented_llm(user_msg: str, tools: list, memory: list):
    # memory: 이전 turns / tools: 함수 스키마 리스트
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system="You are a precise assistant. Use tools when helpful.",
        messages=memory + [{"role": "user", "content": user_msg}],
        tools=tools,
    )
    return response

def llm_call(prompt: str, system: str = "") -> str:
    msg = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=system,
        messages=[{"role": "user", "content": prompt}],
    )
    return msg.content[0].text

def chain(input_text: str, prompts: list[str]) -> str:
    out = input_text
    for i, p in enumerate(prompts):
        out = llm_call(f"{p}\n\n<input>\n{out}\n</input>")
        # 검증 게이트: 형식이 깨졌으면 중단
        if "<error>" in out.lower():
            raise RuntimeError(f"Step {i} failed: {out}")
    return out

# 사용 예: 개요 → 검증 → 본문
result = chain(
    "신년 마케팅 캠페인 주제: 친환경 패션",
    prompts=[
        "3문장 개요를 만든다.",
        "개요가 톤(미니멀, 진정성)에 맞는지 점검하고 다듬어라.",
        "개요를 200자 본문으로 확장하라.",
    ],
)

import json

def classify(query: str) -> str:
    out = llm_call(
        system="분류기. 다음 중 하나로만 답: refund | technical | general",
        prompt=f"고객 문의: {query}\n\n카테고리만 출력.",
    )
    return out.strip().lower()

ROUTES = {
    "refund":    ("claude-haiku-4-5", "환불 정책에 따라 절차를 안내하라."),
    "technical": ("claude-sonnet-4-6", "기술 지원 엔지니어처럼 단계별로 진단하라."),
    "general":   ("claude-haiku-4-5", "친절하게 응답하라."),
}

def route(query: str) -> str:
    category = classify(query)
    model, system = ROUTES.get(category, ROUTES["general"])
    msg = client.messages.create(
        model=model, max_tokens=1024,
        system=system,
        messages=[{"role": "user", "content": query}],
    )
    return msg.content[0].text

import asyncio
from anthropic import AsyncAnthropic

aclient = AsyncAnthropic()

async def acall(prompt: str, system: str) -> str:
    msg = await aclient.messages.create(
        model="claude-sonnet-4-6", max_tokens=512,
        system=system,
        messages=[{"role": "user", "content": prompt}],
    )
    return msg.content[0].text

async def vote_safety_check(code: str) -> bool:
    # Voting: 같은 작업을 다른 프롬프트로 3번 → 하나라도 위험 판정이면 reject
    perspectives = [
        "보안 엔지니어 관점에서 위험 식별",
        "악의적 사용자 관점에서 악용 경로 식별",
        "코드 리뷰어 관점에서 의심스러운 패턴 식별",
    ]
    results = await asyncio.gather(*[
        acall(f"코드:\n{code}\n\n'UNSAFE' 또는 'SAFE'만 답하라.", p)
        for p in perspectives
    ])
    return all("SAFE" in r.upper() for r in results)

def orchestrate(task: str) -> str:
    # 1) 오케스트레이터가 하위 작업 목록을 동적으로 생성
    plan_json = llm_call(
        system="작업을 독립 하위 작업 JSON 배열로 분해. 각 원소: {id, goal}",
        prompt=task,
    )
    subtasks = json.loads(plan_json)

    # 2) 워커들이 하위 작업을 수행
    results = []
    for st in subtasks:
        out = llm_call(
            system="하위 작업을 정확히 수행하고 결과만 반환.",
            prompt=f"{st['goal']}",
        )
        results.append({"id": st["id"], "out": out})

    # 3) 합성
    return llm_call(
        system="하위 결과를 일관된 최종 산출물로 통합.",
        prompt=json.dumps(results, ensure_ascii=False),
    )

def eval_optimize(task: str, max_iters: int = 3) -> str:
    draft = llm_call(system="초안을 작성.", prompt=task)

    for i in range(max_iters):
        critique = llm_call(
            system=("엄격한 평가자. 기준: 명료성, 사실성, 톤. "
                    "마지막 줄에 PASS 또는 REVISE만 출력."),
            prompt=f"<task>{task}</task>\n<draft>{draft}</draft>",
        )
        if critique.strip().splitlines()[-1].strip() == "PASS":
            return draft

        draft = llm_call(
            system="피드백을 반영해 다시 작성.",
            prompt=f"<draft>{draft}</draft>\n<critique>{critique}</critique>",
        )
    return draft  # 한도 도달: 마지막 draft 반환

TOOLS = [
    {
        "name": "read_file",
        "description": "파일 내용을 읽는다. 주니어 개발자에게 설명하듯 명확히.",
        "input_schema": {
            "type": "object",
            "properties": {"path": {"type": "string"}},
            "required": ["path"],
        },
    },
    # ... 다른 도구들
]

def run_tool(name: str, args: dict) -> str:
    if name == "read_file":
        with open(args["path"]) as f:
            return f.read()
    raise ValueError(f"unknown tool: {name}")

def agent(goal: str, max_steps: int = 10):
    messages = [{"role": "user", "content": goal}]

    for step in range(max_steps):
        resp = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            tools=TOOLS,
            messages=messages,
        )

        if resp.stop_reason == "end_turn":
            return resp.content[0].text

        # 도구 호출 처리
        messages.append({"role": "assistant", "content": resp.content})
        tool_results = []
        for block in resp.content:
            if block.type == "tool_use":
                result = run_tool(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result,
                })
        messages.append({"role": "user", "content": tool_results})

    raise RuntimeError("max_steps reached without completion")