04. 구조화된 추론 & LangGraph

from pydantic import BaseModel, Field
from typing import List
 
class SubQueries(BaseModel):
    queries: List[str] = Field(description="원래 쿼리를 해결하기 위한 하위 질문들")
 
def decompose_query(query: str) -> List[str]:
    """복잡한 쿼리를 하위 쿼리로 분해합니다."""
    completion = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "복잡한 문제를 3-4개의 순차적 하위 질문으로 분해하세요."},
            {"role": "user", "content": query}
        ],
        response_format=SubQueries
    )
    return completion.choices[0].message.parsed.queries

def solve_decomposed_queries(original_query: str):
    sub_qs = decompose_query(original_query)
    context = ""
 
    for i, q in enumerate(sub_qs):
        print(f"[Step {i+1}] Q: {q}")
        answer = answer_question(q, context)
        print(f"  ✅ A: {answer}")
        context += f"Q: {q}\nA: {answer}\n\n"
 
    # 최종 답변 종합
    final_answer = answer_question(f"답변: '{original_query}'", context)
    return final_answer

# 1. Proposer: 후보 연속 생성
def propose_next_steps(current_state: str, n: int = 3) -> List[str]:
    prompt = f"""현재 상태: {current_state}
 
{n}개의 가능한 다음 단계를 제안하세요. 1, 2, 3으로 번호를 매기세요.
"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}]
    )
    # 번호가 매겨진 응답 파싱
    lines = response.choices[0].message.content.strip().split("\n")
    return [line.split(". ", 1)[1] for line in lines if ". " in line][:n]
 
# 2. Evaluator: 후보 점수 부여
def evaluate_candidates(candidates: List[str], goal: str) -> List[int]:
    scores = []
    for cand in candidates:
        prompt = f"""목표: {goal}
후보: {cand}
 
관련성과 품질을 1-10점으로 평가하세요. 숫자만 반환하세요.
"""
        response = client.chat.completions.create(...)
        scores.append(int(response.choices[0].message.content.strip()))
    return scores
 
# 3. ToT 검색 루프
def run_tot(goal: str, steps: int = 3):
    current_state = ""
 
    for i in range(steps):
        candidates = propose_next_steps(current_state)
        scores = evaluate_candidates(candidates, goal)
 
        # 최선의 후보 선택
        best_idx = scores.index(max(scores))
        current_state += candidates[best_idx] + " "
 
    return current_state

from langgraph.graph import StateGraph
from typing import TypedDict, List
 
# 1. 상태 스키마 정의
class AgentState(TypedDict):
    query: str
    plan: List[str]
    results: List[str]
    done: bool
 
# 2. 그래프 생성
graph = StateGraph(AgentState)
 
# 3. 노드 추가
graph.add_node("planner", planner_node)
graph.add_node("executor", executor_node)
graph.add_node("checker", checker_node)
 
# 4. 엣지 추가
graph.add_edge("planner", "executor")
graph.add_edge("executor", "checker")
 
# 5. 조건부 엣지 추가
graph.add_conditional_edges(
    "checker",
    should_replan,  # 다음 노드 이름을 반환하는 함수
    {
        "replan": "planner",
        "done": END
    }
)
 
# 6. 컴파일 및 실행
app = graph.compile()
result = app.invoke({"query": "...", "plan": [], "results": [], "done": False})

def planner_node(state: AgentState) -> AgentState:
    """실행 계획 생성"""
    plan = create_plan(state["query"])
    return {"plan": plan}
 
def executor_node(state: AgentState) -> AgentState:
    """계획의 다음 단계 실행"""
    results = execute_step(state["plan"][0])
    remaining_plan = state["plan"][1:]
    return {"plan": remaining_plan, "results": state["results"] + [results]}
 
def checker_node(state: AgentState) -> AgentState:
    """완료 여부 확인"""
    done = len(state["plan"]) == 0
    return {"done": done}
 
def should_replan(state: AgentState) -> str:
    if state["done"]:
        return "done"
    return "continue"