🕸️
LangChain
LangGraph
패스트트랙
LangChain 빌딩 블록부터 LangGraph 상태 그래프, Agent까지 한 번에
#LangChain
#LangGraph
#State
#Node
#Graph
#Interrupt
#Checkpoint
LangChain vs LangGraph
두 라이브러리의 역할과 차이점
🔗 LangChain
- LLM 애플리케이션의 기본 빌딩 블록 제공
- Prompt, Model, Parser, Tool, Retriever 등 컴포넌트 모음
- 체인(Chain) — 단방향 파이프라인
- 간단한 RAG, 요약, 번역 등에 적합
- 분기/반복/상태 관리는 제한적
# 전형적인 단방향 체인
chain = prompt | model | parser
chain.invoke(input={"messages": [("user", "hi")]})
🕸️ LangGraph
- LangChain 위에 얹힌 상태 기반 그래프 프레임워크
- 순환(loop), 분기(branch), 병렬 실행 가능
- Agent 간 협업·툴 호출 루프를 자연스럽게 표현
- 상태(State) + 체크포인트로 지속성·재개 지원
- Human-in-the-loop (Interrupt) 내장
# 상태 기반 그래프
graph = StateGraph(State)
graph.add_node("agent", agent_fn)
graph.add_edge("agent", "tool")
🔗
LangChain 기초
Prompt · Model · Output Parser — LLM 앱의 3대 빌딩 블록
📝 Prompt
변수 치환·역할별 메시지로 재사용 가능한 프롬프트 구성
🤖 Model
OpenAI·Gemini 등 Chat Model을 동일한 인터페이스로 호출
📦 Output Parser
문자열 응답을 구조화된 데이터(dict, Pydantic)로 변환
이어지는 슬라이드에서 영화 리뷰 → JSON 예제를 세 단계로 완성해봅니다.
Prompt Template 예제
재사용 가능한 프롬프트 — 변수 치환 + 메시지 구조화
📝 ChatPromptTemplate
- System / User / Assistant 역할별 메시지를 구성
{변수} 플레이스홀더를 invoke 시 채움MessagesPlaceholder로 대화 이력 주입 가능- 모델과 파이프(
|)로 바로 체인 구성
🧭 언제 쓰나?
- 같은 지시문을 여러 입력에 반복 적용할 때
- 시스템 프롬프트를 고정하고 사용자 입력만 바꿀 때
- few-shot 예시·히스토리를 구조적으로 끼워넣을 때
🎬 예제 시나리오 — 영화 리뷰를 분석해 {sentiment, summary, keywords} JSON으로 추출하는
체인을 3단계(프롬프트 → 모델 → 파서)에 걸쳐 완성해봅니다. 이번 슬라이드는 1단계: 프롬프트.
1 / 3
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.prompts import ChatPromptTemplate
# 1) 리뷰 분석용 프롬프트 — 출력 형식을 시스템 프롬프트에 명시
prompt = ChatPromptTemplate.from_messages([
("system",
"너는 영화 리뷰 분석가야. 아래 JSON 스키마로만 답해.\n"
'{"sentiment": "positive|negative|neutral", '
'"summary": str, "keywords": [str]}'),
("user", "다음 리뷰를 분석해줘:\n{review}"),
])
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system",
"너는 영화 리뷰 분석가야. 아래 JSON 스키마로만 답해.\n"
'{"sentiment": "positive|negative|neutral", '
'"summary": str, "keywords": [str]}'),
("user", "다음 리뷰를 분석해줘:\n{review}"),
])
# 프롬프트만 먼저 확인 (모델/파서는 다음 슬라이드에서 연결)
prompt.invoke({"review": "연출은 좋았지만 각본이 아쉬웠다."})
LangChain 모델 선언 예제
Provider별 Chat Model 초기화 — OpenAI · Gemini · init_chat_model
# OpenAI — pip install -U langchain-openai
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="gpt-4o-mini",
)
llm.invoke([("user", "안녕!")])
# Gemini (Google) — pip install -U langchain-google-genai
from langchain_google_genai import ChatGoogleGenerativeAI
llm = ChatGoogleGenerativeAI(
model="gemini-2.0-flash",
)
llm.invoke([("user", "안녕!")])
# Gemini (Google) — pip install -U langchain-google-genai
from langchain_google_genai import ChatGoogleGenerativeAI
llm = ChatGoogleGenerativeAI(
model="gemini-2.0-flash",
vertexai=True,
project="...",
location="..."
)
llm.invoke([("user", "안녕!")])
# 공통 인터페이스 — Provider 문자열만 바꿔 전환
from langchain.chat_models import init_chat_model
llm = init_chat_model("openai:gpt-4o-mini")
# 또는
llm = init_chat_model("google_genai:gemini-2.0-flash")
# 🎬 앞 슬라이드의 prompt 에 llm 을 파이프로 연결
from langchain.chat_models import init_chat_model
llm = init_chat_model("openai:gpt-4o-mini")
chain = prompt | llm
chain.invoke({"review": "연출은 좋았지만 각본이 아쉬웠다."})
# → AIMessage(content='{"sentiment": "neutral", ...}') ← 아직 문자열
🎬 2단계: 모델 — 프롬프트 + LLM 체인이 완성됩니다. 다음 슬라이드에서 JSON 파서까지 연결합니다.
JsonOutputParser 예제
LLM의 문자열 응답을 dict로 파싱 — 체인 최종 단계
🧪 무엇을 해주나?
- 모델 출력에서 JSON 블록을 추출 후
dict로 파싱
- 마크다운 코드펜스(```json ... ```)도 허용
- Pydantic 모델을 주면 스키마 검증까지 수행
- 스트리밍 시 부분 JSON도 점진적으로 파싱
🔗 전체 체인 조립
prompt | model | parser — 3단계 파이프라인- 입력(
{review}) → 메시지 → 응답 → dict
- 파서의
get_format_instructions()를 프롬프트에 주입하면 더 안정적
1 / 4
from langchain_core.output_parsers import JsonOutputParser
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
# 1) 결과 스키마 정의 (선택)
class ReviewAnalysis(BaseModel):
sentiment: str = Field(description="positive | negative | neutral")
summary: str
keywords: list[str]
from langchain_core.output_parsers import JsonOutputParser
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
class ReviewAnalysis(BaseModel):
sentiment: str = Field(description="positive | negative | neutral")
summary: str
keywords: list[str]
parser = JsonOutputParser(pydantic_object=ReviewAnalysis)
from langchain_core.output_parsers import JsonOutputParser
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
class ReviewAnalysis(BaseModel):
sentiment: str = Field(description="positive | negative | neutral")
summary: str
keywords: list[str]
parser = JsonOutputParser(pydantic_object=ReviewAnalysis)
# 2) 앞 슬라이드의 prompt 와 llm 을 그대로 사용 → 파서까지 파이프 연결
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | parser
from langchain_core.output_parsers import JsonOutputParser
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
class ReviewAnalysis(BaseModel):
sentiment: str = Field(description="positive | negative | neutral")
summary: str
keywords: list[str]
parser = JsonOutputParser(pydantic_object=ReviewAnalysis)
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | parser
result = chain.invoke({"review": "연출은 좋았지만 각본이 아쉬웠다."})
# → {"sentiment": "neutral", "summary": "...", "keywords": [...]}
print(result["sentiment"])
with_structured_output 으로 바꿔보기
파서 없이 모델이 직접 Pydantic 객체를 반환하도록
✨ 무엇이 달라지나?
- 모델이 내부적으로 tool/function calling 또는 JSON 모드를 사용
- 출력이 Pydantic 인스턴스(또는 dict)로 바로 반환
- 프롬프트에 포맷 지침을 적지 않아도 스키마가 강제됨
JsonOutputParser 불필요 → 체인이 더 단순
⚖️ Parser vs Structured Output
- Parser 어떤 모델에서도 동작 (문자열 기반)
- Structured 스키마 준수율↑, 코드 간결
- 구조화 스펙이 단단하면 Structured 권장
1 / 4
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
# 1) 앞 슬라이드와 동일한 스키마
class ReviewAnalysis(BaseModel):
sentiment: str = Field(description="positive | negative | neutral")
summary: str
keywords: list[str]
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
class ReviewAnalysis(BaseModel):
sentiment: str = Field(description="positive | negative | neutral")
summary: str
keywords: list[str]
# 2) 모델 자체에 스키마를 바인딩 — 파서 불필요
llm = ChatOpenAI(model="gpt-4o-mini")
structured_llm = llm.with_structured_output(ReviewAnalysis)
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
class ReviewAnalysis(BaseModel):
sentiment: str = Field(description="positive | negative | neutral")
summary: str
keywords: list[str]
llm = ChatOpenAI(model="gpt-4o-mini")
structured_llm = llm.with_structured_output(ReviewAnalysis)
# 3) prompt 는 앞 슬라이드 것을 그대로 재사용
chain = prompt | structured_llm
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
class ReviewAnalysis(BaseModel):
sentiment: str = Field(description="positive | negative | neutral")
summary: str
keywords: list[str]
llm = ChatOpenAI(model="gpt-4o-mini")
structured_llm = llm.with_structured_output(ReviewAnalysis)
chain = prompt | structured_llm
result = chain.invoke({"review": "연출은 좋았지만 각본이 아쉬웠다."})
# → ReviewAnalysis(sentiment='neutral', summary='...', keywords=[...])
print(result.sentiment, result.keywords)
Runnable 인터페이스
한 번 Runnable이 되면 LangChain · LangGraph 전체로 동일한 동작이 전파
🧩 핵심 아이디어
- Prompt · Model · Parser · Tool 은 모두 Runnable
- 공통 메서드:
invoke / stream / batch (+ async 버전)
- 파이프(
|)로 합성해도 결과 역시 Runnable
- LangGraph는 노드 함수를 내부적으로
RunnableLambda로 감쌈
→ 그래서 그래프 자체에도 invoke/stream/batch가 그대로 열림
🌊 무엇이 전파되나?
- 스트리밍 체인·그래프 어디서든 토큰 단위
.stream()
- 배치 여러 입력을
.batch()로 병렬 처리
- 비동기
.ainvoke() / .astream() 자동 지원
- 트레이싱 LangSmith 추적·콜백·재시도가 일관 적용
1 / 3
from langchain_core.runnables import RunnableLambda
# 1) 평범한 파이썬 함수를 Runnable 로 감싸기
def shout(x: str) -> str:
return x.upper() + "!"
shout_r = RunnableLambda(shout)
shout_r.invoke("hi") # → "HI!"
shout_r.batch(["hi", "bye"]) # → ["HI!", "BYE!"] ← 병렬 자동
from langchain_core.runnables import RunnableLambda
def shout(x: str) -> str:
return x.upper() + "!"
shout_r = RunnableLambda(shout)
# 2) 다른 Runnable 과 파이프로 합성 — 동일 규약이 그대로 전파
chain = prompt | llm | parser | shout_r
chain.invoke({"review": "..."}) # invoke / stream / batch 모두 동작
from langchain_core.runnables import RunnableLambda
def shout(x: str) -> str:
return x.upper() + "!"
shout_r = RunnableLambda(shout)
chain = prompt | llm | parser | shout_r
chain.invoke({"review": "..."})
# 3) LangGraph 노드로 그대로 꽂아도 동일하게 동작
builder.add_node("shout", shout_r) # 함수든 Runnable이든 OK
graph = builder.compile()
graph.stream({"review": "..."}) # ← 그래프에도 stream 이 전파됨
🕸️
LangGraph 기초
상태(State) 기반 그래프로 순환·분기·사람 개입이 있는 워크플로우 설계
🗂️ State · Node · Graph
공유 상태 위에서 노드들이 실행되고, 엣지가 다음 단계를 결정
🔀 Conditional Edge
노드 반환값·State 값에 따라 다음 노드를 동적으로 분기
⏸ Interrupt
민감 동작 전 사람 승인·편집을 받아 재개하는 Human-in-the-loop
💾 Checkpoint
스텝마다 상태 스냅샷을 저장해 지속성·재개·시간여행
LangChain 컴포넌트를 그래프로 엮어 상태 있는 워크플로우를 만듭니다.
핵심 구성: State · Node · Graph
LangGraph를 이루는 세 가지 뼈대
🗂️ State
그래프 실행 전체에서 공유되는 데이터 컨테이너.
TypedDict / Pydantic으로 스키마 정의.
- 각 노드는 State를 입력으로 받아 일부를 업데이트해 반환
- 필드마다 병합 방식을 지정 — 덮어쓰기(기본) / 이어붙이기 등
※ 이 "병합 규칙"을 LangGraph에서는 reducer라 부름 (예: 메시지 이력을 누적하는 add_messages)
🧩 Node
State를 받아 업데이트된 State를 반환하는 함수 단위.
- LLM 호출, 툴 실행, 후처리 등 임의 로직
- 파이썬 함수 / Runnable 무엇이든 가능
add_node("name", fn) 로 등록
🔀 Graph (Edge)
노드 간 흐름을 정의. 정적 엣지와 조건부 엣지를 지원.
add_edge(a, b) — 무조건 이동add_conditional_edges — 함수 반환값으로 분기START / END 상수로 시작·종료 지정
1 / 5
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
# 앞 슬라이드의 chain = prompt | structured_llm 을 그대로 재사용
class State(TypedDict):
review: str
analysis: ReviewAnalysis | None
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
# 스테이트 정의 - 그래프 전체에서 공유
class State(TypedDict):
review: str
analysis: ReviewAnalysis | None
# 노드 정의
def agent(state: State):
result = chain.invoke({"review": state["review"]})
return {"analysis": result}
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
review: str
analysis: ReviewAnalysis | None
def agent(state: State):
result = chain.invoke({"review": state["review"]})
return {"analysis": result}
builder = StateGraph(State)
# 노드 추가
builder.add_node("agent", agent)
# 엣지 추가
builder.add_edge(START, "agent")
builder.add_edge("agent", END)
# 그래프 생성
graph = builder.compile()
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
review: str
analysis: ReviewAnalysis | None
def agent(state: State):
result = chain.invoke({"review": state["review"]})
return {"analysis": result}
builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_edge(START, "agent")
builder.add_edge("agent", END)
graph = builder.compile()
graph.invoke(input={"review": "연출은 좋았지만 각본이 아쉬웠다."})
# → {"review": "...", "analysis": ReviewAnalysis(sentiment='neutral', ...)}
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
review: str
analysis: ReviewAnalysis | None
def agent(state: State):
result = chain.invoke({"review": state["review"]})
return {"analysis": result}
builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_edge(START, "agent")
builder.add_edge("agent", END)
graph = builder.compile()
# 초기 State 를 명시적으로 선언해서 실행
initial_state: State = {
"review": "연출은 좋았지만 각본이 아쉬웠다.",
"analysis": None,
}
graph.invoke(initial_state)
# → {"review": "...", "analysis": ReviewAnalysis(sentiment='neutral', ...)}
Conditional Edge (조건부 분기)
노드 반환값에 따라 다음 노드를 동적으로 결정
START
→
agent
→
route?
→
reply_positive
reply_negative
→
END
1 / 3
from langgraph.graph import StateGraph, START, END
# 앞 페이지의 State / agent 를 그대로 사용
def reply_positive(state: State): return {}
def reply_negative(state: State): return {}
from langgraph.graph import StateGraph, START, END
def reply_positive(state: State): return {}
def reply_negative(state: State): return {}
# 라우팅 함수 — 다음 노드 이름(문자열)을 반환
def route(state: State) -> str:
if state["analysis"].sentiment == "positive":
return "reply_positive"
return "reply_negative"
from langgraph.graph import StateGraph, START, END
def reply_positive(state: State): return {}
def reply_negative(state: State): return {}
def route(state: State) -> str:
if state["analysis"].sentiment == "positive":
return "reply_positive"
return "reply_negative"
builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("reply_positive", reply_positive)
builder.add_node("reply_negative", reply_negative)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", route) # ← 분기
builder.add_edge("reply_positive", END)
builder.add_edge("reply_negative", END)
graph = builder.compile()
Interrupt (사람이 끼어드는 지점)
그래프 실행을 일시 정지하고, 외부 입력을 기다렸다가 재개
START
→
agent
→
⏸ human_review
→
END
⏸ 언제 쓰나?
- Human-in-the-loop — 툴 호출 전 사람 승인
- 민감한 동작(결제·삭제) 확인
- 중간 결과 편집·검토
- 외부 이벤트(알림·폼 제출) 대기
🧪 두 가지 방식
- 정적
compile(interrupt_before=["tool"]) — 특정 노드 앞에서 무조건 멈춤
- 동적
interrupt(payload) — 노드 내부에서 조건적으로 중단
- 재개는
Command(resume=...) 로 값을 주입
1 / 4
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
# 앞 그래프 + human_review 노드 추가
def human_review(state: State):
edited = interrupt({
"question": "분석 결과를 확인해 주세요. 수정할 부분이 있나요?",
"analysis": state["analysis"].model_dump(),
})
return {"analysis": ReviewAnalysis(**edited)}
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
def human_review(state: State):
edited = interrupt({
"question": "분석 결과를 확인해 주세요. 수정할 부분이 있나요?",
"analysis": state["analysis"].model_dump(),
})
return {"analysis": ReviewAnalysis(**edited)}
builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("human_review", human_review)
builder.add_edge(START, "agent")
builder.add_edge("agent", "human_review")
builder.add_edge("human_review", END)
graph = builder.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "review-1"}}
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
def human_review(state: State):
edited = interrupt({
"question": "분석 결과를 확인해 주세요. 수정할 부분이 있나요?",
"analysis": state["analysis"].model_dump(),
})
return {"analysis": ReviewAnalysis(**edited)}
builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("human_review", human_review)
builder.add_edge(START, "agent")
builder.add_edge("agent", "human_review")
builder.add_edge("human_review", END)
graph = builder.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "review-1"}}
# 1) 실행 → human_review 의 interrupt 에서 멈춤
graph.invoke(input={"review": "연출은 좋았지만 각본이 아쉬웠다."}, config=cfg)
builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("human_review", human_review)
builder.add_edge(START, "agent")
builder.add_edge("agent", "human_review")
builder.add_edge("human_review", END)
graph = builder.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "review-1"}}
# 1) 실행 → human_review 의 interrupt 에서 멈춤
graph.invoke(input={"review": "연출은 좋았지만 각본이 아쉬웠다."}, config=cfg)
# 2) 사람이 확인·수정한 값을 주입해 재개
graph.invoke(input=Command(resume={
"sentiment": "negative",
"summary": "각본의 약점이 두드러진 리뷰.",
"keywords": ["각본", "연출"],
}), config=cfg)
Checkpoint (상태 저장과 재개)
매 단계마다 State 스냅샷을 저장 → 중단·복구·시간여행이 가능
💾 무엇을 저장하나?
- 각 super-step 직후의 State 스냅샷
- 다음에 실행할 노드(=다음 작업)
thread_id 단위로 이력을 묶음
🧰 주요 Checkpointer
InMemorySaver — 개발·테스트SqliteSaver — 단일 프로세스 영속화PostgresSaver — 프로덕션 다중 워커- 커스텀 백엔드 구현 가능
🎁 얻는 것
- 지속성 세션 간 대화 유지
- 재개 Interrupt 후 이어서 실행
- 시간여행 과거 체크포인트로 분기
- 관측 각 스텝의 상태를 감사
1 / 3
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
cfg = {"configurable": {"thread_id": "user-42"}}
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
cfg = {"configurable": {"thread_id": "user-42"}}
graph.invoke(input={"messages": [{"role": "user", "content": "안녕, 내 이름은 지호야"}]}, config=cfg)
graph.invoke(input={"messages": [{"role": "user", "content": "내 이름 기억해?"}]}, config=cfg) # 이어짐
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
cfg = {"configurable": {"thread_id": "user-42"}}
graph.invoke(input={"messages": [{"role": "user", "content": "안녕, 내 이름은 지호야"}]}, config=cfg)
graph.invoke(input={"messages": [{"role": "user", "content": "내 이름 기억해?"}]}, config=cfg)
# 과거 상태 확인 / 시간여행
for snap in graph.get_state_history(cfg):
print(snap.values, snap.next)
🤖
Agent
LLM이 스스로 툴을 선택·호출하며 목표를 달성하는 루프
🔁 ReAct 루프
Reason → Act → Observe 를 반복하며 툴 결과를 보고 다음 행동을 결정
🧰 Tool Calling
함수 시그니처를 모델에 제공 → 모델이 호출할 툴·인자를 JSON으로 지정
🎁 create_agent
모델 + 툴만 주면 위 루프 그래프를 자동으로 조립해주는 프리셋
앞서 본 State · Graph · Checkpoint · Interrupt 가 모두 하나의 에이전트에 녹아듭니다.
create_agent — 프리셋 ReAct 에이전트
모델 + 툴 + 시스템 프롬프트만 주면 그래프를 자동으로 조립
🎁 무엇을 해주나?
- 내부적으로 agent ↔ tools 를 오가는 ReAct 루프 그래프를 자동 생성
- 메시지 상태(
messages)와 add_messages reducer를 기본 제공
checkpointer 전달 시 대화 지속성 바로 사용HumanInTheLoopMiddleware 로 툴별 approve·edit·reject 승인 설정- 반환값은 일반
CompiledGraph — 커스터마이즈·확장 가능
🧭 언제 쓰나?
- 툴을 쓰는 일반적인 챗 에이전트를 빠르게 만들고 싶을 때
- 직접
StateGraph로 조립하기 전, 프로토타입용
- 표준 패턴으로 충분하고, 특이한 분기가 필요 없을 때
※ 분기/병렬/도메인 특화 흐름이 필요하면 그대로 StateGraph로 내려가면 됩니다.
1 / 4
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
# 1) 툴 정의
@tool
def get_weather(city: str) -> str:
"""도시의 날씨를 반환."""
return f"{city}: 맑음 22°C"
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
@tool
def get_weather(city: str) -> str:
"""도시의 날씨를 반환."""
return f"{city}: 맑음 22°C"
# 2) 에이전트 생성 (모델 + 툴 + 프롬프트)
agent = create_agent(
model="anthropic:claude-sonnet-4-6",
tools=[get_weather],
system_prompt="너는 친절한 한국어 비서야.",
)
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
@tool
def get_weather(city: str) -> str:
"""도시의 날씨를 반환."""
return f"{city}: 맑음 22°C"
# 2) 에이전트 생성 (+ HITL 미들웨어 + 체크포인트)
agent = create_agent(
model="anthropic:claude-sonnet-4-6",
tools=[get_weather],
system_prompt="너는 친절한 한국어 비서야.",
middleware=[
HumanInTheLoopMiddleware(
interrupt_on={"get_weather": True}, # 툴 실행 전 사람 승인
),
],
checkpointer=InMemorySaver(),
)
agent = create_agent(
model="anthropic:claude-sonnet-4-6",
tools=[get_weather],
system_prompt="너는 친절한 한국어 비서야.",
middleware=[
HumanInTheLoopMiddleware(
interrupt_on={"get_weather": True},
),
],
checkpointer=InMemorySaver(),
)
# 3) 채팅 메시지 인터페이스로 호출
cfg = {"configurable": {"thread_id": "user-42"}}
result = agent.invoke(
input={"messages": [{"role": "user", "content": "서울 날씨 알려줘"}]},
config=cfg,
)
print(result.interrupts) # 승인 대기 중인 action_requests
그래프 안에서 Agent 호출하기
create_agent 결과는 일반 Runnable — 노드 하나로 꽂을 수 있음
START
→
plan
→
🤖 weather_agent
→
summarize
→
END
1 / 4
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.agents import create_agent
# 1) 앞 슬라이드의 에이전트를 그대로 사용
weather_agent = create_agent(
model="anthropic:claude-sonnet-4-6",
tools=[get_weather],
system_prompt="너는 친절한 한국어 비서야.",
)
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.agents import create_agent
weather_agent = create_agent(
model="anthropic:claude-sonnet-4-6",
tools=[get_weather],
system_prompt="너는 친절한 한국어 비서야.",
)
class State(TypedDict):
question: str
agent_output: dict
final: str
weather_agent = create_agent(
model="anthropic:claude-sonnet-4-6",
tools=[get_weather],
system_prompt="너는 친절한 한국어 비서야.",
)
class State(TypedDict):
question: str
agent_output: dict
final: str
# 2) 에이전트를 노드 함수 안에서 호출
def call_agent(state: State):
result = weather_agent.invoke(
{"messages": [{"role": "user", "content": state["question"]}]}
)
return {"agent_output": result}
def summarize(state: State):
last = state["agent_output"]["messages"][-1].content
return {"final": f"[요약] {last}"}
def call_agent(state: State):
result = weather_agent.invoke(
{"messages": [{"role": "user", "content": state["question"]}]}
)
return {"agent_output": result}
def summarize(state: State):
last = state["agent_output"]["messages"][-1].content
return {"final": f"[요약] {last}"}
builder = StateGraph(State)
builder.add_node("weather_agent", call_agent) # ← 에이전트를 한 노드로
builder.add_node("summarize", summarize)
builder.add_edge(START, "weather_agent")
builder.add_edge("weather_agent", "summarize")
builder.add_edge("summarize", END)
graph = builder.compile()
graph.invoke({"question": "서울 날씨 알려줘"})
한 장 요약
LangChain(빌딩 블록) → LangGraph(상태 그래프) → Agent(ReAct 루프)
🔗 LangChain
- Prompt — 변수 치환·역할별 메시지
- Model — OpenAI·Gemini 등 통합 인터페이스
- Output Parser — JsonOutputParser
- Structured Output — 모델에 Pydantic 스키마 바인딩
🕸️ LangGraph
- State · Node · Graph — 상태 머신으로 설계
- Conditional Edge — 반환값으로 동적 분기
- Interrupt — 사람 승인·편집 후 재개
- Checkpoint — 지속성·재개·시간여행
🤖 Agent
- ReAct 루프 — Reason → Act → Observe
- Tool Calling — 모델이 툴·인자 선택
- create_agent (langchain.agents) — 모델+툴만 주면 그래프 자동 조립
- 필요 시
StateGraph로 커스터마이즈
🧠 사고 모델
"여러 단계의 LLM/툴 호출을 상태 머신처럼 설계하자."
각 노드는 상태를 조금씩 바꾸고, 엣지가 다음 단계를 정한다.
🛠 언제 쓰면 좋은가
- 툴 호출이 있는 챗 에이전트
- 멀티 에이전트 협업 (Planner/Worker/Critic)
- 사람 승인이 필요한 자동화
- 오래 걸리는 작업의 중단·재개
📚 더 읽을거리
- LangChain: Prompts / Output Parsers / Structured Output
- LangGraph: Concepts / Human-in-the-loop / Persistence
- LangChain v1:
create_agent + Middleware 가이드
더 공부해볼 것
에이전트를 실전에서 쓰려면 알아두면 좋은 개념들
🪝 Middleware & Hooks
create_agent의 agent loop 각 지점에 로직 삽입.
- Node-style:
before_agent / before_model / after_model / after_agent
- Wrap-style:
wrap_model_call / wrap_tool_call
- 데코레이터(
@before_model) 또는 AgentMiddleware 클래스로 정의
🧰 Built-in Middleware
SummarizationMiddleware — 긴 대화 자동 요약HumanInTheLoopMiddleware — 툴 승인·편집·거절ModelCallLimit / ToolCallLimit — 루프 폭주 방지ModelFallbackMiddleware — 실패 시 백업 모델PIIMiddleware — 개인정보 마스킹·차단TodoListMiddleware, LLMToolSelectorMiddleware 등
🧠 Memory · Store
- Short-term:
Checkpointer 기반 thread 메모리
- Long-term:
Store — 사용자/도메인 지식 영속화
SummarizationMiddleware로 컨텍스트 관리
🌊 Streaming 모드
values — 매 스텝의 전체 Stateupdates — 노드가 반환한 변경분만messages — 토큰 단위 LLM 스트림custom — get_stream_writer로 임의 이벤트
🧵 Subgraph · Multi-agent
- 그래프를 다른 그래프의 노드로 중첩
- Supervisor / Swarm / Handoff 패턴
prebuilt: create_supervisor, create_swarm
🧪 Observability · 평가
- LangSmith — 트레이싱·데이터셋·평가
- Eval: 골든셋 + LLM-as-judge
- 프로덕션 배포: LangGraph Platform
참고: langchain.agents.middleware / langgraph.store / LangSmith 공식 문서