step5_agent.py

"""Step 5: ReAct 에이전트 루프 — LLM 판단 + 도구 + 파일을 섞은 진짜 에이전트.

시나리오:
1. LLM이 "LangGraph가 언제 출시됐는지"를 자기 지식으로 답함 (판단)
2. 출시일을 release.txt 에 저장 (write_note)
3. 파일을 다시 읽어(read_note) 오늘까지 며칠 지났는지 계산 (days_since)
4. 그 결과를 파일에 덧붙임 (append_note)

핵심: agent <-> tools 순환(cyclic graph) + 종료 조건(should_continue).
- LLM이 더 이상 도구를 안 부르면 종료
- 또는 반복 한도(MAX_STEPS)에 도달하면 강제 종료 (무한루프 방지)

도구(days_since, write_note, read_note, append_note)는 이 단계에서 직접 들고 있는다.
"""

from datetime import date, datetime
from pathlib import Path
from typing import Annotated, TypedDict

from langgraph.graph import START, END, StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool

from llm import get_llm, configure_from_cli, log_messages

NOTES_DIR = Path(__file__).parent / "notes"


@tool
def days_since(date_str: str) -> str:
    """주어진 날짜부터 오늘까지 며칠 지났는지 계산한다. 날짜는 'YYYY-MM-DD' 형식."""
    try:
        d = datetime.strptime(date_str, "%Y-%m-%d").date()
    except ValueError:
        return "오류: 날짜는 'YYYY-MM-DD' 형식이어야 합니다."
    return str((date.today() - d).days)


def _safe_path(filename: str) -> Path | None:
    """notes/ 안의 경로만 허용. 벗어나면 None."""
    NOTES_DIR.mkdir(parents=True, exist_ok=True)
    target = (NOTES_DIR / filename).resolve()
    if NOTES_DIR.resolve() not in target.parents and target != NOTES_DIR.resolve():
        return None
    return target


@tool
def write_note(filename: str, content: str) -> str:
    """메모를 notes/ 폴더의 파일에 저장한다(기존 내용 덮어씀)."""
    target = _safe_path(filename)
    if target is None:
        return "오류: notes/ 폴더 밖 경로는 허용되지 않습니다."
    target.write_text(content, encoding="utf-8")
    return f"{filename} 저장 완료"


@tool
def read_note(filename: str) -> str:
    """notes/ 폴더에 저장된 메모를 읽는다."""
    target = _safe_path(filename)
    if target is None:
        return "오류: notes/ 폴더 밖 경로는 허용되지 않습니다."
    if not target.exists():
        return f"'{filename}' 파일이 없습니다."
    return target.read_text(encoding="utf-8")


@tool
def append_note(filename: str, content: str) -> str:
    """notes/ 폴더의 파일에 내용을 '새 줄'로 덧붙인다(없으면 새로 만든다)."""
    target = _safe_path(filename)
    if target is None:
        return "오류: notes/ 폴더 밖 경로는 허용되지 않습니다."
    # 기존 내용이 줄바꿈으로 끝나지 않으면 줄바꿈을 먼저 넣어 항상 새 줄에 붙인다
    prefix = ""
    if target.exists():
        existing = target.read_text(encoding="utf-8")
        if existing and not existing.endswith("\n"):
            prefix = "\n"
    with target.open("a", encoding="utf-8") as f:
        f.write(prefix + content)
    return f"{filename} 에 추가 완료"


TOOLS = [days_since, write_note, read_note, append_note]
MAX_STEPS = 8  # agent 호출 최대 횟수 (안전 가드)


class State(TypedDict):
    messages: Annotated[list, add_messages]
    steps: int  # agent 호출 횟수 (반복 한도 판단용)


def agent(state: State) -> dict:
    """도구가 바인딩된 LLM 호출 + 호출 횟수 증가."""
    log_messages(state["messages"])  # LLM에 보내는 프롬프트 로그
    llm = get_llm().bind_tools(TOOLS)
    return {
        "messages": [llm.invoke(state["messages"])],
        "steps": state.get("steps", 0) + 1,
    }


def should_continue(state: State) -> str:
    """순환을 이어갈지(tools) 종료할지(END) 결정하는 조건 함수."""
    last = state["messages"][-1]
    if not getattr(last, "tool_calls", None):
        return END  # LLM이 도구를 그만 부름 -> 종료
    if state["steps"] >= MAX_STEPS:
        return END  # 반복 한도 도달 -> 강제 종료
    return "tools"  # 도구 호출 남음 -> 계속 순환


def build_graph():
    graph = StateGraph(State)
    graph.add_node("agent", agent)
    graph.add_node("tools", ToolNode(TOOLS))
    graph.add_edge(START, "agent")
    # 순환 + 컨디셔널 조합: 조건에 따라 tools로 돌거나 END로 종료
    graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
    graph.add_edge("tools", "agent")  # 도구 결과를 다시 LLM으로 -> 루프
    return graph.compile()


if __name__ == "__main__":
    configure_from_cli()  # --profile/--model/--api-key/--base-url (없으면 .env)
    app = build_graph()
    question = (
        "LangGraph가 언제 출시됐는지 알려줘. "
        "release.txt 에 첫 줄로 '랭그래프 출시일: YYYY-MM-DD' 형식으로 저장해줘. "
        "그 다음 파일을 다시 읽어서 오늘까지 며칠 지났는지 계산하고, "
        "줄을 바꿔서 '랭그래프가 출시된지 N일 지남' 형식으로 release.txt 에 덧붙여줘."
    )
    result = app.invoke({"messages": [HumanMessage(content=question)], "steps": 0})
    print(result["messages"][-1].content)  # 최종 답변
    print(f"\n(총 agent 호출 {result['steps']}회)")