티스토리 뷰
노드, 엣지, 상태관리를 통해 Agentic Work Flow를 가진다.
노드 = 각각의 상태. 실행하고자 하는 하나의 함수라고 보면 된다.
엣지 = 어디서부터 어디로 보낼 것인가?
두 개 이상의 엣지 구현 가능
상태 = 현재의 상태 값을 저장 및 전달하는 데 활용
순환 연산 기능으로 쉽게 흐름을 제어
Conditional Edge : 조건부 흐름 제어를 할 수 있게 해주는 엣지. 분기 처리가 가능하다.
Human-in-the-loop : 사람이 직접 개입해서 정하게 하는 것도 가능하다.
상태(State)
TypeDict : 일반 파이썬 dict에 타입헌팅을 추가. 쉽게 말하면 dictionary
모든 값을 다 채우지 않아도 된다.
새로운 노드에서 값을 덮어쓰기(Overwrite) 방식으로 채운다.
어떤 식으로 업데이트가 되는가?
첫 번째 노드에서 두 번째 노드로 전달이 될 때, Value를 전부 채울 필요는 없다.
두 번째 노드에서 필요한 상태 값을 조회아며 동작에 활용할 수 있다.
이전에 업데이트한 Value가 그대로 다음 노드로 전달되고, 새로이 업데이트 된 Value만 갱신된다.
어떤 특정 단계에서 있는 값이 그대로 유지가 되는 게 도움이 될 때가 있다 -> 첫번째 입력된 질문이 유지되어야 할 때.
Relevance Check를 할 때 사용할 수도 있다.
Context, Question, Answer, Score가 있으면,
노드 1, 노드2, 노드3, 노드4를 거치면서 Question, Context, Answer, Score 순으로 Value가 채워진다.
만약 질문 재작성을 요청하려면 노드 1에서의 Question만 바꿔주면 되고,
노드1은 그대로 두고 context 재작성(Retriever)을 요청해도 되고,
노드1,2는 그대로 두고 Answer만 재작성(프롬프트)를 요청해도 된다.
노드
노드는 함수로 정의하며, Langchain에서는 Runable이 된다.
상태 객체를 입력으로 받아야 하며, Return 하는 것도 동일한 상태 객체를 반환해야 한다.
def retrieve_document(state: GraphState) -> GraphState:
# Question에 대한 문서 검색을 retriever로 수행합니다.
retrieved_docs = pdf_retriever.invoke(state["question"])
# 검색된 문서를 context 키에 저장합니다.
return GraphState(context=format_docs(retrieved_docs))상태 객체가 들어오면 결국 dict로 생각하면 된다. 여기에는 key : value 쌍으로 우리가 필요한 정보들이 존재한다.
state에 있는 question이 검색된 문서가 된다.
그러면 당연히 dict이므로 해당하는 context라는 키에다 넣어서 반환해주면 된다.
다만 Conditional Edge의 경우 다를 수 있음
중간에 Langchain이 꼭 들어가지 않아도 된다. 알아서 동작을 위한 코드를 구현할 수 있다.
def llm_answer(state: GraphState) -> GraphState:
return GraphState(answer=pdf_chain.invoke({"question":state["question"], "context":state["contenxt"]}))
Graph 생성 후에 노드 추가
이전에 정의한 함수를 Graph에 추가한다.
add_node("노드이름", 함수) 로 추가한다.
from langgraph.graph import END, StateGraph
from langgraph.checkpoint.memory import MemorySaver
# langgraph.graph에서 StateGraph와 END를 가져옵니다.
workflow = StateGraph(GraphState)
workflow.add_node("retriever", retrieve_document) # 에이전트 노드를 추가합니다.
workflow.add_node("llm_answer", llm_answer)
엣지(Edge)
노드에서 노드간의 연결
add_edge("노드 이름","노드 이름")
from > to
차례대로 연결해주면 된다.
workflow.add_edge("retriever", "llm_answer")
workflow.add_edge("llm_answer", "relevance_check")
조건부 엣지(Conditional Edge)
LangGraph 기능의 핵심
노드에 조건부 엣지를 추가하여 분기를 수행할 수 있다.
add_conditional_edges("노드이름", 조건부 판단 함수, dict로 다음 단계 결정)
workflow.add_conditional_edges(
"relevance_check", # 관련성 체크 노드에서 나온 결과를 is_relevant 함수에 전달합니다. 어느 노드에 추가할 것인가?
is_relevant, #해당 노드가 반환하는 건 String을 반환해야 한다(grounded,notGrounded,notSure)
{
"grounded": "END", # 관련성이 있으면 종료합니다.
"notGrounded": "llm_answer", # 관련성이 없으면 다시 답변을 생성합니다.
"notSure": "llm_answer", # 관련성 체크 결과가 모호하다면 다시 답변을 생성합니다.
},
)
시작점 지정
set_entry_point("노드이름")
지정한 시작점부터 Graph가 시작한다.
workflow.set_entry_point("retrieve")여기서는 retrieve 노드에서 Graph가 시작하도록 설정했다.
체크포인터(memory)
Checkpointer는 각 노드간의 실행결과를 추적하기 위한 메모리이다(대화에 대한 기록과 유사한 개념)
체크포인터를 활용하여 특정 시점(Snapshot)으로 되돌리기 기능도 가능하다
compile(checkpointer=memory) 지정하여 그래프를 작성한다.
# 기록을 위한 메모리 저장소를 설정합니다.
memory = MemorySaver()
# 그래프를 컴파일합니다.
app = workflow.compile(checkpointer=memory)compile은 말 그대로 그래프를 생성하는 것
노드의 흐름을 추적하거나 스냅샷 기능을 사용해서 되돌리기 기능을 활용하기 위해 메모리를 넣어준다.
그래프 시각화
from IPython.display import Image, display
try:
display(
Image(app.get_graph(xray=True).draw_mermaid_png())
) # 실행 가능한 객체의 그래프를 mermaid 형식의 PNG로 그려서 표시합니다.
# xray=True는 추가적인 세부 정보를 포함합니다.
except:
# 이 부분은 추가적인 의존성이 필요하며 선택적으로 실행됩니다.
pass
그래프 실행
import pprint
from langgraph.errors import GraphRecursionError
from langchain_core.runnables import RunnableConfig
config = RunnableConfig(recursion_limit=100, configurable={"thread_id": "TODO"})
# GraphState 객체를 활용하여 질문을 입력합니다.
inputs = GraphState(goal="랭체인(LangChain) 밋업에서 발표자료를 준비하기")
# app.stream을 통해 입력된 메시지에 대한 출력을 스트리밍합니다.
try:
for output in app.stream(inputs, config=config):
# 출력된 결과에서 키와 값을 순회합니다.
for key, value in output.items():
# 노드의 이름과 해당 노드에서 나온 출력을 출력합니다.
pprint.pprint(f"[NODE] {key}")
for k, v in value.items():
pprint.pprint(f"<{k}> {v}")
pprint.pprint("===" * 10)
# 출력 값을 예쁘게 출력합니다.
# pprint.pprint(value, indent=2, width=80, depth=None)
except GraphRecursionError as e:
pprint.pprint(f"Recursion limit reached: {e}")그래프를 실행할 때에는 RunnableConfig라는 걸 만들어서 실행한다.
config에 있는 다양한 옵션들을 넣어서 실행한다.
recursion limit은 최대 노드 실행 개수를 의미한다. 전체 100개 노드까지 거칠 수 있다는 뜻
thread_id는 그래프 실행 아이디를 기록하고 추적하기 위한 용도
input은 상태로 시작하고, question에 질문만 입력하고 생타를 첫번째 노드에 전달한다.
원하는 키 값을 담아서 invoke를 해준다.
invoke(상태, config)를 전달하여 실행하게 된다.
output = app.stream(inputs, config=config)로 나온다.
이 녀석도 그냥 dict로 생각하면 된다.
key 값으로 원하는 값을 뽑아낸다.
마지막 그래프 상태의 체크를 해준다.
Grouned Check
그라운드 체크를 통해서 고객사에게 더 나은 경험을 제공하는 것을 권장한다.
Upstage의 Groundedness Check를 넣어도 된다.
context와 answer만 넣어도 자동으로 해준다.
from rag.utils import format_docs
from langchain_upstage import UpstageGroundednessCheck
# 업스테이지 문서 관련성 체크 기능을 설정합니다. https://upstage.ai
upstage_ground_checker = UpstageGroundednessCheck()
# 업스테이지 문서 관련성 체크를 실행합니다.
upstage_ground_checker.run(
{
"context": format_docs(
pdf_retriever.invoke("삼성전자가 개발한 생성AI 의 이름은")
),
"answer": "삼성전자가 개발한 생성AI 의 이름은 `빅스비 AI` 입니다.",
}
)
그러면 relevance_check는 다음과 같이 된다.
def relevance_check(state: GraphState) -> GraphState:
# 관련성 체크를 실행합니다. 결과: grounded, notGrounded, notSure
response = upstage_ground_checker.run(
{"context": state["context"], "answer": state["answer"]}
)
return GraphState(
relevance=response,
context=state["context"],
answer=state["answer"],
question=state["question"],
)
Type hinting
from typing import TypedDict,Annotated, Sequence
import operator
# GraphState 상태를 저장하는 용도로 사용합니다.
class GraphState(TypedDict):
question: Annotated[str, operator.add] # 질문
context: Annotated[Sequence[Documnet], operator.add] # 문서의 검색 결과
answer: Annotated[Sequence[Documnet], operator.add] # 답변
sql_query: Annotated[str, operator.add] # 쿼리
binary_score: Annotated[str, operator.add]
해당하는 Document로 만드는 것
operator.add는 type hinting을 위해서 쓰인다.
양질의 강의를 제공해주신 테디노트님께 감사드립니다.
Reference:
https://teddylee777.github.io/langgraph/langgraph-agentic-rag/
https://www.youtube.com/watch?v=4JdzuB702wI