LangGraph로 Agentic AI 시스템 구축하기 — 테디노트 밋업 발표 완전 정리

# 나쁜 예: grade_documents가 query_rewrite에 강하게 의존
def grade_documents(state, query_rewrite_result):  # 직접 참조
    ...

# 좋은 예: 모든 노드는 State만 읽는다
def grade_documents(state: AgentState) -> dict:
    docs = state["documents"]  # State에서 읽기만
    ...
    return {"grade": "relevant"}  # State에 쓰기만

# LLM 노드에서 모델만 바꾸면 끝
def generate_response(state: AgentState) -> dict:
    # llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash")
    # llm = ChatOpenAI(model="gpt-5")
    llm = ChatAnthropic(model="claude-opus-4-7")  # 한 줄만 바꾸면 됨
    response = llm.invoke(state["messages"])
    return {"response": response.content}

tools = [web_search_tool, document_search_tool]
llm_with_tools = llm.bind_tools(tools)

def agent_node(state):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

from pydantic import BaseModel
from typing import Literal

class RouteQuery(BaseModel):
    destination: Literal["web_search", "document_search"]

structured_llm = llm.with_structured_output(RouteQuery)

def route_query(state):
    result = structured_llm.invoke(state["messages"])
    return result.destination  # "web_search" 또는 "document_search"

# 문서 검색과 웹 검색을 동시에
graph.add_edge("start", "document_search")
graph.add_edge("start", "web_search")

# 두 결과를 합치는 노드
graph.add_edge("document_search", "merge_results")
graph.add_edge("web_search", "merge_results")

def selective_fanout(state) -> list[str]:
    if state["needs_web"]:
        return ["document_search", "web_search"]
    return ["document_search"]  # 웹 검색 건너뜀

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# thread_id로 대화 세션 구분
config = {"configurable": {"thread_id": "user_123_session_A"}}

app.invoke({"messages": [HumanMessage("내 이름은 테디야")]}, config)
app.invoke({"messages": [HumanMessage("내 이름이 뭐라고?")]}, config)
# → "당신의 이름은 테디입니다."

# thread_id 변경하면 이전 대화 기억 없음
config2 = {"configurable": {"thread_id": "user_123_session_B"}}
app.invoke({"messages": [HumanMessage("내 이름이 뭐라고?")]}, config2)
# → "이름을 알 수 없습니다."

# 4번 노드까지 실행됐는데 2번 결과가 잘못됨
history = list(app.get_state_history(config))

# 2번 노드 시점으로 롤백
past_state = history[2]
# State를 직접 수정 (잘못된 쿼리 → 올바른 쿼리)
app.update_state(past_state.config, {"query": "개선된 검색어"})
# 2번 시점부터 재실행 (1번은 다시 안 함)
app.invoke(None, past_state.config)

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
app = graph.compile(checkpointer=memory, store=store)

# user_id로 장기 메모리 구분 (thread가 달라도 공유됨)
config = {
    "configurable": {
        "thread_id": "session_new",
        "user_id": "teddy"
    }
}

# LLM이 대화에서 개인 정보를 자동 추출해 저장
class PersonalEntity(BaseModel):
    name: Optional[str] = None
    job: Optional[str] = None
    location: Optional[str] = None
    hobby: Optional[str] = None

def extract_and_save_memory(state, store, config):
    user_id = config["configurable"]["user_id"]
    
    # 대화에서 개인 정보 추출
    extractor = llm.with_structured_output(PersonalEntity)
    entity = extractor.invoke(state["messages"])
    
    # 장기 메모리에 저장
    store.put(("user_profile", user_id), "profile", entity.dict())

from langgraph.types import interrupt

def review_content(state):
    # 여기서 실행이 멈추고 사람의 입력을 기다림
    user_feedback = interrupt({
        "content": state["draft"],
        "question": "이 내용으로 진행할까요? (승인/수정 내용)"
    })
    return {"feedback": user_feedback}

# interrupt_before로 특정 노드 실행 전 중단
app = graph.compile(
    checkpointer=memory,
    interrupt_before=["send_email"]
)

# 실행 → send_email 직전에 멈춤
app.invoke(state, config)

# 사람이 State 확인 후 승인
print(app.get_state(config).values)
app.invoke(None, config)  # 멈춘 지점부터 재개

def supervisor_node(state):
    system_prompt = """당신은 팀 슈퍼바이저입니다.
    팀원: researcher(정보 수집), writer(글 작성), analyst(데이터 분석)
    작업을 완료할 팀원을 선택하거나, 완료시 FINISH를 반환하세요."""
    
    response = llm.with_structured_output(NextAction).invoke(
        [SystemMessage(system_prompt)] + state["messages"]
    )
    return {"next": response.next_agent}

# 복잡한 "페이퍼 작성 팀" 그래프
paper_writing_team = StateGraph(...)
# ... (리서처, 라이터, 에디터 노드 복잡하게 구성)
paper_writing_app = paper_writing_team.compile()

# 상위 그래프에서 하나의 노드로 사용
outer_graph = StateGraph(OuterState)
outer_graph.add_node("paper_writing_team", paper_writing_app)