ReAct 심화

from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from typing import TypedDict, Annotated
from operator import add
import json

MAX_ITERATIONS = 10  # 무한 루프 방지 상한

class AgentState(TypedDict):
    messages: Annotated[list, add]
    iteration: int
    tool_calls_made: list[str]  # 중복 도구 호출 추적

llm = ChatAnthropic(model="claude-opus-4-7")

# ─── THOUGHT 노드 ────────────────────────────────────
async def reasoning_node(state: AgentState) -> dict:
    """LLM이 다음 행동을 추론"""
    iteration = state["iteration"] + 1

    # 시스템 프롬프트에 현재 반복 횟수 주입
    system = f"""당신은 도구를 사용하여 질문에 답하는 AI 에이전트입니다.
현재 반복: {iteration}/{MAX_ITERATIONS}

규칙:
1. 확실한 정보가 있으면 바로 Final Answer 작성
2. 동일한 도구를 동일한 인수로 두 번 호출하지 마세요
3. 불필요한 도구 호출은 피하세요"""

    llm_with_tools = llm.bind_tools(tools)
    response = await llm_with_tools.ainvoke(
        state["messages"],
        system=system
    )
    return {"messages": [response], "iteration": iteration}

# ─── ACTION 노드 ────────────────────────────────────
async def action_node(state: AgentState) -> dict:
    """도구 실행 + 중복 호출 감지"""
    last_message = state["messages"][-1]
    tool_messages = []
    calls_made = state["tool_calls_made"][:]

    for call in last_message.tool_calls:
        call_signature = f"{call['name']}:{json.dumps(call['args'], sort_keys=True)}"

        if call_signature in calls_made:
            # 중복 호출 — 캐시된 결과 반환
            result = "[캐시] 이전과 동일한 결과"
        else:
            calls_made.append(call_signature)
            result = await execute_tool(call["name"], call["args"])

        tool_messages.append(ToolMessage(
            tool_call_id=call["id"],
            content=str(result)
        ))

    return {"messages": tool_messages, "tool_calls_made": calls_made}

# ─── 라우팅 ────────────────────────────────────────
def should_continue(state: AgentState) -> str:
    last = state["messages"][-1]

    # 강제 종료: 반복 초과
    if state["iteration"] >= MAX_ITERATIONS:
        return "end"

    # 도구 호출이 있으면 action으로
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "action"

    return "end"

# ─── 그래프 구성 ─────────────────────────────────────
builder = StateGraph(AgentState)
builder.add_node("reasoning", reasoning_node)
builder.add_node("action",    action_node)
builder.set_entry_point("reasoning")
builder.add_conditional_edges("reasoning", should_continue, {
    "action": "action",
    "end":    END
})
builder.add_edge("action", "reasoning")

agent = builder.compile()

REACT_SYSTEM_PROMPT = """당신은 정확하고 효율적인 AI 에이전트입니다.

## 사고 과정 (Thought)
각 단계에서 다음을 명확히 합니다:
- 현재까지 파악한 정보
- 아직 확인이 필요한 정보
- 다음 행동의 이유

## 도구 사용 원칙
1. **최소 호출**: 한 번의 도구 호출로 필요한 모든 정보를 얻으세요
2. **병렬 가능 도구**: 독립적인 정보는 Parallel Tool Calling 활용
3. **오류 처리**: 도구 실패 시 대체 방법을 시도하거나 현재 정보로 답변

## 최종 답변 기준
다음 조건 중 하나라도 충족하면 즉시 최종 답변을 작성하세요:
- 질문에 답하기에 충분한 정보를 수집한 경우
- 더 이상 새로운 정보를 얻을 수 없는 경우
- 5회 이상 도구를 사용한 경우

## 금지 사항
- 동일한 도구를 같은 인수로 두 번 호출
- 이미 알고 있는 정보를 재확인하기 위한 도구 호출
- 과도한 자기 의심으로 인한 루프"""

# 도구별 예시 포함 프롬프트 생성
def build_system_prompt(tools: list, domain: str = "") -> str:
    tool_descriptions = "\n".join(
        f"- {t.name}: {t.description}" for t in tools
    )
    domain_context = f"\n## 도메인: {domain}" if domain else ""

    return f"""{REACT_SYSTEM_PROMPT}

## 사용 가능한 도구
{tool_descriptions}{domain_context}"""