Autonomous Graph - Difficulty-Based Task Execution Graph
Building on Claude Control's autonomous execution mode, I implemented an Autonomous Graph that selects different execution strategies based on task difficulty. Using LangGraph's StateGraph, easy tasks are handled in a single session while hard tasks follow a manager-worker pattern.
Difficulty Classification
Tasks are classified into three tiers.
from enum import IntEnum
from langgraph.graph import StateGraph, END
from typing import TypedDict, Literal
class Difficulty(IntEnum):
EASY = 1 # 단일 파일 수정, 간단한 버그 픽스
MEDIUM = 2 # 여러 파일 수정, 기능 추가
HARD = 3 # 대규모 리팩토링, 아키텍처 변경
class TaskState(TypedDict):
task: str
difficulty: Difficulty
plan: list
results: list
iteration: int
max_iterations: int
is_complete: bool
Graph Structure
[classify] → EASY → [single execute] → [verify] → END
→ MEDIUM → [plan] → [sequential execute] → [verify] → END
→ HARD → [decompose] → [parallel execute] → [integrate] → [verify] → END
async def classify_difficulty(state: TaskState) -> TaskState:
llm = get_claude_model()
response = await llm.ainvoke([HumanMessage(content=f"""
다음 태스크의 난이도를 1(쉬움), 2(보통), 3(어려움)으로 분류하세요.
숫자만 출력하세요.
태스크: {state['task']}""")])
state["difficulty"] = Difficulty(int(response.content.strip()))
return state
def route_by_difficulty(state: TaskState) -> Literal["easy", "medium", "hard"]:
if state["difficulty"] == Difficulty.EASY:
return "easy"
elif state["difficulty"] == Difficulty.MEDIUM:
return "medium"
return "hard"
Execution Strategy by Difficulty
async def execute_easy(state: TaskState) -> TaskState:
"""단일 세션으로 직접 실행"""
result = await session_manager.send_prompt(
"worker-0", state["task"]
)
state["results"].append(result)
state["is_complete"] = True
return state
async def plan_medium(state: TaskState) -> TaskState:
"""중간 난이도: 단계별 계획 수립"""
llm = get_claude_model()
response = await llm.ainvoke([HumanMessage(content=f"""
다음 태스크를 순차적 단계로 나누세요. JSON 배열로 출력하세요.
태스크: {state['task']}""")])
state["plan"] = json.loads(response.content)
return state
async def decompose_hard(state: TaskState) -> TaskState:
"""높은 난이도: 병렬 가능한 서브태스크로 분해"""
llm = get_claude_model()
response = await llm.ainvoke([HumanMessage(content=f"""
다음 태스크를 독립적으로 병렬 실행 가능한 서브태스크로 분해하세요.
각 서브태스크는 다른 작업에 의존하지 않아야 합니다.
JSON 배열로 출력하세요.
태스크: {state['task']}""")])
state["plan"] = json.loads(response.content)
return state
Assembling the Graph
def build_autonomous_graph() -> StateGraph:
graph = StateGraph(TaskState)
graph.add_node("classify", classify_difficulty)
graph.add_node("easy_execute", execute_easy)
graph.add_node("medium_plan", plan_medium)
graph.add_node("medium_execute", execute_sequential)
graph.add_node("hard_decompose", decompose_hard)
graph.add_node("hard_execute", execute_parallel)
graph.add_node("hard_integrate", integrate_results)
graph.add_node("verify", verify_results)
graph.set_entry_point("classify")
graph.add_conditional_edges("classify", route_by_difficulty, {
"easy": "easy_execute",
"medium": "medium_plan",
"hard": "hard_decompose",
})
graph.add_edge("easy_execute", "verify")
graph.add_edge("medium_plan", "medium_execute")
graph.add_edge("medium_execute", "verify")
graph.add_edge("hard_decompose", "hard_execute")
graph.add_edge("hard_execute", "hard_integrate")
graph.add_edge("hard_integrate", "verify")
graph.add_edge("verify", END)
return graph.compile()
Visualization
I also wrote a script that visualizes the Autonomous Graph's execution flow as a Mermaid diagram, making it possible to trace which path was taken after the fact.
Results
Difficulty-based routing eliminated unnecessary overhead for simple tasks and cut turnaround time on complex tasks through parallel execution. In practice, average processing time for HARD tasks dropped by roughly 40%.