목차
- 문제와 접근
- AdaptiveState 설계
- init_state 초기화 함수
- _Route 스키마와 classify_node
- 분류 전용 LangGraph
- direct 노드와 조건 분기
- 사내 문서 FAISS 검색기
- rag_node 구현
- direct/rag 통합 그래프
- 핵심 개념 총정리
- 자주 나오는 실수/주의사항
- 분류와 답변을 분리해서 확인하기
1. 문제와 접근
분류기를 실제 처리 경로에 연결하면 route 값과 State의 전달 규칙이 중요해진다. 직접 답변과 문서 검색을 같은 그래프 안에서 처리하되, 각 경로가 어떤 근거로 답했는지를 구분한다. 초기화와 반환 구조를 맞추어 경로를 바꾸어도 결과를 동일한 방식으로 확인할 수 있게 한다.
1
2
3
4
| direct : LLM이 바로 답변
rag : 사내 문서 검색 후 답변
web : 최신/실시간 정보 검색
multihop : 여러 단계 추론이 필요한 질문
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| import os
from typing import TypedDict, Annotated, Literal
from operator import add
from dotenv import load_dotenv
load_dotenv()
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.documents import Document
from langchain_community.vectorstores import FAISS
from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel, Field
llm = ChatOpenAI(model="gpt-4o-mini")
|
2. AdaptiveState 설계
라우팅, 검색 결과, 멀티홉 상태를 모두 담는 state를 만든다.
1
2
3
4
5
6
7
8
9
10
| class AdaptiveState(TypedDict):
query: str
route: str
documents: Annotated[list[str], add]
web_results: Annotated[list[str], add]
sub_queries: Annotated[list[str], add]
retrieved: Annotated[list[str], add]
hop_count: int
max_hops: int
answer: str
|
Annotated[list, add]는 노드가 반환한 리스트를 기존 리스트에 누적할 때 사용한다.
3. init_state 초기화 함수
초기화 함수는 요청 사이에 남으면 안 되는 값을 명확히 하는 자리다. 이전 검색 문서나 답변을 새 질문에서 재사용할 의도가 없다면 초기값을 정하고, 같은 상태 객체를 여러 요청에서 공유하지 않는지 확인한다.
그래프 실행에 필요한 기본값을 한 번에 만든다.
1
2
3
4
5
6
7
8
9
10
11
12
| def init_state(q: str, max_hops: int = 3) -> dict:
return {
"query": q,
"route": "",
"documents": [],
"web_results": [],
"sub_queries": [],
"retrieved": [],
"hop_count": 0,
"max_hops": max_hops,
"answer": "",
}
|
4. _Route 스키마와 classify_node
LLM의 라우팅 결과를 Pydantic 스키마로 제한한다.
1
2
3
4
5
6
7
8
9
| class _Route(BaseModel):
"""쿼리를 처리 경로로 분류."""
route: Literal["direct", "rag", "web", "multihop"] = Field(
description="direct=상식, rag=사내문서, web=최신정보, multihop=복합추론"
)
router_llm = llm.with_structured_output(_Route)
|
1
2
3
4
5
6
7
8
9
10
| def classify_node(state: AdaptiveState) -> dict:
sys = (
"질문을 분류하세요. direct=LLM 상식, rag=사내 문서, "
"web=최신/실시간, multihop=여러 단계 추론이 필요한 복합 질문."
)
r = router_llm.invoke([
SystemMessage(content=sys),
HumanMessage(content=state["query"]),
])
return {"route": r.route}
|
5. 분류 전용 LangGraph
먼저 classify 노드만 가진 최소 그래프로 라우팅 결과를 확인한다.
1
2
3
4
5
| builder = StateGraph(AdaptiveState)
builder.add_node("classify", classify_node)
builder.add_edge(START, "classify")
builder.add_edge("classify", END)
app_v2 = builder.compile()
|
1
2
3
4
5
6
7
8
9
10
| queries = [
"파이썬 리스트가 뭐야?",
"우리 회사 휴가 정책은?",
"오늘 코스피 지수는?",
"봉준호 2019년 영화의 음악감독 출생연도는?",
]
for q in queries:
r = app_v2.invoke(init_state(q))
print(f"{q[:25]:<25} -> route={r['route']}")
|
6. direct 노드와 조건 분기
direct로 분류된 질문은 LLM이 바로 답한다.
1
2
3
4
5
6
7
8
9
10
| def direct_node(state: AdaptiveState) -> dict:
r = llm.invoke([
SystemMessage(content="간결히 답변하세요."),
HumanMessage(content=state["query"]),
])
return {"answer": r.content}
def route_fn(state: AdaptiveState) -> str:
return state["route"]
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| builder = StateGraph(AdaptiveState)
builder.add_node("classify", classify_node)
builder.add_node("direct", direct_node)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route_fn, {
"direct": "direct",
"rag": END,
"web": END,
"multihop": END,
})
builder.add_edge("direct", END)
app_v3 = builder.compile()
|
7. 사내 문서 FAISS 검색기
RAG route에서 사용할 간단한 사내 문서 코퍼스를 만든다.
1
2
3
4
5
6
7
8
9
10
11
12
13
| DOCS = [
"휴가 정책: 입사 1년 미만은 월 1일, 1년 이상은 연 15일 유급 휴가가 부여됩니다.",
"출장 규정: 국내 출장은 일비 5만원, 해외는 일비 10만원이 지급됩니다.",
"재택근무: 주 2일까지 재택근무 가능. 팀장 사전 승인 필요.",
"야근 식대: 오후 8시 이후 야근 시 1만원 식대 지급.",
"교육비 지원: 직무 관련 도서/강의 연 30만원 한도 지원.",
"건강검진: 매년 1회 종합검진 무료 제공.",
"복장 규정: 비즈니스 캐주얼. 고객 미팅 시 정장 권장.",
]
docs = [Document(page_content=d) for d in DOCS]
vstore = FAISS.from_documents(docs, OpenAIEmbeddings())
retriever = vstore.as_retriever(search_kwargs={"k": 2})
|
8. rag_node 구현
RAG 경로에서는 검색 결과가 비어 있거나 질문과 무관한 경우를 별도로 확인한다. 답변 문자열만 반환하면 직접 답한 경우와 근거를 찾은 경우가 구분되지 않으므로, 경로와 사용한 문서 정보를 결과와 함께 추적하는 구성을 검토한다.
질문으로 문서를 검색하고, 검색된 문서를 근거로 답변한다.
1
2
3
4
5
6
7
8
9
10
11
| def rag_node(state: AdaptiveState) -> dict:
hits = retriever.invoke(state["query"])
ctx = "\n".join(h.page_content for h in hits)
r = llm.invoke([
SystemMessage(content=f"다음 사내 문서 근거로 답변해주세요:\n{ctx}"),
HumanMessage(content=state["query"]),
])
return {
"documents": [h.page_content for h in hits],
"answer": "[rag] " + r.content,
}
|
필기에는 반환 키가 document로 적힌 부분이 있는데, state에는 documents가 있으므로 정리본에서는 맞춰 적었다.
9. direct/rag 통합 그래프
분류 결과에 따라 direct 또는 rag 노드로 이동한다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| builder = StateGraph(AdaptiveState)
builder.add_node("classify", classify_node)
builder.add_node("direct", direct_node)
builder.add_node("rag", rag_node)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route_fn, {
"direct": "direct",
"rag": "rag",
"web": END,
"multihop": END,
})
builder.add_edge("direct", END)
builder.add_edge("rag", END)
app_v3 = builder.compile()
|
1
2
3
| r = app_v3.invoke(init_state("우리 회사 휴가 정책은?"))
r["route"]
# -> 'rag'
|
전체 흐름
1
2
3
4
5
6
7
8
| START
↓
classify
↓ route_fn
├─ direct -> END
├─ rag -> END
├─ web -> END
└─ multihop -> END
|
10. 핵심 개념 총정리
| 구성요소 | 역할 |
|---|
| AdaptiveState | 라우팅/검색/답변 상태를 담는 TypedDict |
| _Route | LLM 분류 결과를 direct/rag/web/multihop 중 하나로 제한 |
| classify_node | 질문을 route로 분류 |
| route_fn | 조건부 엣지에서 다음 노드 선택 |
| direct_node | LLM 단독 답변 |
| FAISS retriever | 사내 문서 검색 |
| rag_node | 검색 문서를 근거로 답변 생성 |
11. 자주 나오는 실수/주의사항
실수 1: state 키와 반환 키 불일치
1
2
3
4
5
| # 틀린 코드
return {"document": [...]}
# 올바른 코드
return {"documents": [...]}
|
실수 2: 조건부 엣지 mapping 누락
1
2
3
4
5
6
| builder.add_conditional_edges("classify", route_fn, {
"direct": "direct",
"rag": "rag",
"web": END,
"multihop": END,
})
|
route_fn이 반환할 수 있는 모든 route를 mapping에 넣어야 한다.
실수 3: Annotated[list, add]에 문자열 반환
1
2
| # documents는 list 누적 필드이므로 리스트를 반환해야 함
return {"documents": [h.page_content for h in hits]}
|
실수 4: web/multihop 미구현 상태
1
2
| web, multihop은 route만 준비되어 있고 실제 처리 노드는 아직 구현 전이다.
END로 보내면 답변 없이 끝날 수 있다.
|
12. 분류와 답변을 분리해서 확인하기
최종 답변이 기대와 다르면 route 선택부터 확인하고 다음으로 검색 결과를 본다. 분류가 잘못된 상태에서 RAG 프롬프트를 조정해도 해당 질문이 그 경로를 거치지 않으면 결과는 바뀌지 않는다.
같은 질문을 의도한 경로에 직접 넣은 결과와 라우터를 통과한 결과를 비교하면 문제를 좁히기 쉽다. 경로별 독립 확인을 남겨 두면 새 route를 추가할 때도 기존 처리의 변화를 살펴볼 수 있다.