5장: 계획 수립 패턴 - 복잡한 작업의 분해와 실행

import anthropic
import json
 
client = anthropic.Anthropic()
 
def create_plan(task: str) -> list[dict]:
    """작업을 분석하고 실행 계획을 생성합니다."""
    plan_prompt = f"""다음 작업을 수행하기 위한 단계별 계획을 수립하십시오.
 
작업: {task}
 
각 단계는 다음 정보를 포함해야 합니다:
- step: 단계 번호
- description: 수행할 작업 설명
- dependencies: 이 단계가 의존하는 이전 단계 번호 목록
- tools_needed: 필요한 도구 목록
 
JSON 배열로 응답하십시오.
예시:
[
  {{"step": 1, "description": "...", "dependencies": [], "tools_needed": ["search_web"]}},
  {{"step": 2, "description": "...", "dependencies": [1], "tools_needed": ["calculator"]}}
]"""
 
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2048,
        messages=[{"role": "user", "content": plan_prompt}],
    )
 
    try:
        return json.loads(response.content[0].text)
    except json.JSONDecodeError:
        return [{"step": 1, "description": task, "dependencies": [], "tools_needed": []}]
 
def execute_step(
    step: dict,
    previous_results: dict[int, str],
    tools: list[dict]
) -> str:
    """계획의 한 단계를 실행합니다."""
    # 의존하는 단계의 결과를 컨텍스트로 구성
    context = ""
    for dep in step["dependencies"]:
        if dep in previous_results:
            context += f"\n[단계 {dep}의 결과]: {previous_results[dep]}"
 
    exec_prompt = f"""다음 작업을 수행하십시오.
 
작업: {step['description']}
 
{f'이전 단계의 결과:{context}' if context else ''}
 
제공된 도구를 활용하여 작업을 완료하십시오."""
 
    messages = [{"role": "user", "content": exec_prompt}]
 
    # ReAct 루프로 단계 실행
    for _ in range(5):
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        )
 
        if response.stop_reason == "end_turn":
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text
            return "단계 완료"
 
        # 도구 호출 처리
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                result = execute_tool(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result,
                })
 
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})
 
    return "최대 반복 횟수 도달"
 
def plan_and_execute(task: str, tools: list[dict]) -> str:
    """Plan-and-Execute를 실행합니다."""
    # 1. 계획 수립
    plan = create_plan(task)
    print(f"[계획] {len(plan)}개 단계로 구성")
    for step in plan:
        print(f"  단계 {step['step']}: {step['description']}")
 
    # 2. 단계별 실행
    results = {}
    for step in plan:
        print(f"\n[실행] 단계 {step['step']}: {step['description']}")
        result = execute_step(step, results, tools)
        results[step["step"]] = result
        print(f"[결과] {result[:200]}...")
 
    # 3. 최종 종합
    synthesis_prompt = f"""다음 작업의 각 단계별 결과를 종합하여 최종 답변을 작성하십시오.
 
원래 작업: {task}
 
단계별 결과:
{json.dumps(results, ensure_ascii=False, indent=2)}
 
모든 결과를 통합하여 포괄적인 최종 답변을 제공하십시오."""
 
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=[{"role": "user", "content": synthesis_prompt}],
    )
    return response.content[0].text

class AdaptivePlanner:
    def __init__(self):
        self.client = anthropic.Anthropic()
        self.plan = []
        self.results = {}
        self.current_step = 0
 
    def should_replan(self, step_result: str, step: dict) -> bool:
        """재계획이 필요한지 판단합니다."""
        check_prompt = f"""다음 단계의 실행 결과를 분석하십시오.
 
계획된 작업: {step['description']}
실행 결과: {step_result}
 
다음 중 해당하는 상황이 있으면 "REPLAN"을, 정상이면 "CONTINUE"를 응답하십시오:
1. 단계가 실패했거나 예상과 다른 결과를 반환했다
2. 새로운 정보가 발견되어 이후 계획을 수정해야 한다
3. 원래 가정이 잘못되었음이 드러났다
 
한 단어로만 응답하십시오: REPLAN 또는 CONTINUE"""
 
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=10,
            messages=[{"role": "user", "content": check_prompt}],
        )
        return "REPLAN" in response.content[0].text.upper()
 
    def replan(self, task: str, completed_results: dict) -> list[dict]:
        """지금까지의 결과를 바탕으로 남은 계획을 재수립합니다."""
        replan_prompt = f"""원래 작업과 지금까지의 결과를 바탕으로 남은 단계를 재계획하십시오.
 
원래 작업: {task}
 
완료된 단계와 결과:
{json.dumps(completed_results, ensure_ascii=False, indent=2)}
 
남은 작업을 완료하기 위한 새로운 단계들을 JSON 배열로 제시하십시오.
이미 완료된 결과를 최대한 활용하십시오."""
 
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2048,
            messages=[{"role": "user", "content": replan_prompt}],
        )
        try:
            return json.loads(response.content[0].text)
        except json.JSONDecodeError:
            return []
 
    def run(self, task: str, tools: list[dict]) -> str:
        """적응적 재계획이 포함된 실행 루프입니다."""
        self.plan = create_plan(task)
        replan_count = 0
        max_replans = 3
 
        i = 0
        while i < len(self.plan):
            step = self.plan[i]
            print(f"\n[실행] 단계 {step['step']}: {step['description']}")
 
            result = execute_step(step, self.results, tools)
            self.results[step["step"]] = result
 
            # 재계획 필요 여부 확인
            if self.should_replan(result, step) and replan_count < max_replans:
                print(f"[재계획] 실행 결과에 따라 계획을 수정합니다.")
                remaining = self.replan(task, self.results)
                if remaining:
                    self.plan = list(self.plan[:i + 1]) + remaining
                    replan_count += 1
 
            i += 1
 
        return self.synthesize(task)
 
    def synthesize(self, task: str) -> str:
        """모든 결과를 종합합니다."""
        prompt = f"""작업: {task}\n결과: {json.dumps(self.results, ensure_ascii=False)}
모든 결과를 종합하여 최종 답변을 작성하십시오."""
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            messages=[{"role": "user", "content": prompt}],
        )
        return response.content[0].text

[
  {"step": 1, "description": "요구 사항 분석", "dependencies": []},
  {"step": 2, "description": "데이터 수집", "dependencies": [1]},
  {"step": 3, "description": "분석 실행", "dependencies": [2]},
  {"step": 4, "description": "보고서 작성", "dependencies": [3]}
]

from collections import defaultdict
import asyncio
 
class DAGPlanExecutor:
    def __init__(self, plan: list[dict], tools: list[dict]):
        self.plan = {step["step"]: step for step in plan}
        self.tools = tools
        self.results = {}
        self.completed = set()
 
    def get_ready_steps(self) -> list[dict]:
        """의존성이 모두 충족된 실행 가능한 단계를 반환합니다."""
        ready = []
        for step_id, step in self.plan.items():
            if step_id in self.completed:
                continue
            deps = set(step.get("dependencies", []))
            if deps.issubset(self.completed):
                ready.append(step)
        return ready
 
    async def execute_parallel(self) -> dict:
        """DAG 구조에 따라 병렬 실행합니다."""
        while len(self.completed) < len(self.plan):
            ready = self.get_ready_steps()
            if not ready:
                break
 
            print(f"[병렬 실행] {len(ready)}개 단계 동시 실행")
            tasks = [
                asyncio.to_thread(execute_step, step, self.results, self.tools)
                for step in ready
            ]
            results = await asyncio.gather(*tasks)
 
            for step, result in zip(ready, results):
                self.results[step["step"]] = result
                self.completed.add(step["step"])
                print(f"  단계 {step['step']} 완료")
 
        return self.results

def create_hierarchical_plan(task: str, depth: int = 2) -> dict:
    """계층적 계획을 생성합니다."""
    prompt = f"""다음 작업을 계층적으로 분해하십시오.
 
작업: {task}
 
1단계: 3-5개의 고수준 단계로 나누십시오.
2단계: 각 고수준 단계를 2-4개의 세부 단계로 나누십시오.
 
JSON으로 응답:
{{
  "task": "원래 작업",
  "phases": [
    {{
      "name": "고수준 단계 이름",
      "description": "설명",
      "substeps": [
        {{"description": "세부 단계 1"}},
        {{"description": "세부 단계 2"}}
      ]
    }}
  ]
}}"""
 
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4096,
        messages=[{"role": "user", "content": prompt}],
    )
    return json.loads(response.content[0].text)

def validate_plan(plan: list[dict], available_tools: list[str]) -> list[str]:
    """계획의 유효성을 검증하고 문제점을 반환합니다."""
    issues = []
 
    # 1. 의존성 순환 검사
    visited = set()
    def has_cycle(step_id, path):
        if step_id in path:
            return True
        path.add(step_id)
        step = next((s for s in plan if s["step"] == step_id), None)
        if step:
            for dep in step.get("dependencies", []):
                if has_cycle(dep, path.copy()):
                    return True
        return False
 
    for step in plan:
        if has_cycle(step["step"], set()):
            issues.append(f"단계 {step['step']}에 순환 의존성이 있습니다.")
 
    # 2. 존재하지 않는 의존성 검사
    step_ids = {s["step"] for s in plan}
    for step in plan:
        for dep in step.get("dependencies", []):
            if dep not in step_ids:
                issues.append(
                    f"단계 {step['step']}이 존재하지 않는 "
                    f"단계 {dep}에 의존합니다."
                )
 
    # 3. 필요한 도구 가용성 검사
    for step in plan:
        for tool in step.get("tools_needed", []):
            if tool not in available_tools:
                issues.append(
                    f"단계 {step['step']}에 필요한 도구 "
                    f"'{tool}'을 사용할 수 없습니다."
                )
 
    return issues