LangGraph 워크플로우 템플릿 (v6)
LangGraph者，為現代AI之輔，尤善於狀態之流，適於作工之序。此篇所載，乃解眾工所遇之困，列五要範，以為指南。
1. LangGraph之體例
LangGraph之構，有要者五：
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Annotated
import operator
# 상태 정의
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
# 기타 상태 필드
# 노드 정의
def retrieve_node(state):
# 문서 검색 로직
pass
def generate_node(state):
# 생성 로직
pass
# 엣지 정의
def route_to_next(state):
# 다음 노드 결정 로직
pass
# 워크플로우 정의
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("generate", generate_node)
workflow.add_edge("retrieve", "generate")
workflow.add_conditional_edges("generate", route_to_next)
workflow.set_entry_point("retrieve")
workflow.set_finish_point(END)
# 체크포인팅 설정
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
2. 範一：簡易之RAG輔 (索→作→驗)
此模板，合文档检索与生成，乃RAG之基础流程也。
from langchain_core.messages import HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
import operator
from typing import Annotated, List
from langgraph.graph import StateGraph, END
class RAGState(TypedDict):
query: str
retrieved_docs: List[str]
generated_response: str
validation_result: bool
# 검색 노드
def retrieve_node(state: RAGState):
# Chroma에서 문서 검색
vectorstore = Chroma(
persist_directory="./chroma_db",
embedding_function=OpenAIEmbeddings()
)
docs = vectorstore.similarity_search(state["query"], k=3)
return {"retrieved_docs": [doc.page_content for doc in docs]}
# 생성 노드
def generate_node(state: RAGState):
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant that answers questions based on provided context."),
("human", """Context: {context}
Question: {question}
Answer:""")
])
model = ChatOpenAI(model="gpt-4")
chain = prompt | model
response = chain.invoke({
"context": "\n".join(state["retrieved_docs"]),
"question": state["query"]
})
return {"generated_response": response.content}
# 검증 노드
def validate_node(state: RAGState):
# 생성된 응답 검증
if len(state["generated_response"]) < 10:
return {"validation_result": False}
# 간단한 키워드 기반 검증
keywords = ["answer", "question", "context", "response"]
valid = any(keyword in state["generated_response"].lower() for keyword in keywords)
return {"validation_result": valid}
# 워크플로우 정의
def create_rag_workflow():
workflow = StateGraph(RAGState)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("generate", generate_node)
workflow.add_node("validate", validate_node)
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", "validate")
# 검증 결과에 따라 처리
def route_validation(state: RAGState):
if not state["validation_result"]:
return "retry"
return "end"
workflow.add_conditional_edges(
"validate",
route_validation,
{
"retry": "retrieve",
"end": END
}
)
workflow.set_entry_point("retrieve")
return workflow.compile()
# 사용 예시
# app = create_rag_workflow()
# result = app.invoke({"query": "Python 디자인 패턴에 대해 설명해주세요"})
3. 模板二：多工器代理（谋→行→察→决）
此模板，适于用众器之复代理，其流程繁也。
from typing import Literal
from langgraph.graph import StateGraph, END
from langchain_core.messages import ToolMessage
class ToolAgentState(TypedDict):
messages: Annotated[list, operator.add]
plan: List[str]
current_tool: str
tool_result: str
# 계획 노드
def plan_node(state: ToolAgentState):
# 도구 사용 계획 생성
prompt = ChatPromptTemplate.from_messages([
("system", "You are a task planner. Analyze the following request and create a plan of actions."),
("human", "Request: {request}\nPlan:")
])
model = ChatOpenAI(model="gpt-4")
chain = prompt | model
response = chain.invoke({"request": state["messages"][-1].content})
return {"plan": response.content.split('\n')}
# 실행 노드
def execute_node(state: ToolAgentState):
# 도구 실행
tool_name = state["plan"][0] # 첫 번째 계획 항목
# 실제 도구 실행 로직 (예: 파일 시스템 조작)
if tool_name == "file_search":
# 파일 검색 로직
result = "Found files matching criteria"
elif tool_name == "code_analysis":
# 코드 분석 로직
result = "Code analysis complete"
else:
result = "Tool not recognized"
return {"current_tool": tool_name, "tool_result": result}
# 관찰 노드
def observe_node(state: ToolAgentState):
# 도구 결과 관찰
messages = state["messages"]
tool_message = ToolMessage(
content=state["tool_result"],
tool_call_id=state["current_tool"]
)
return {"messages": [tool_message]}
# 결정 노드
def decide_node(state: ToolAgentState):
# 다음 단계 결정
if len(state["plan"]) > 1:
return "continue"
return "end"
# 워크플로우 정의
def create_tool_workflow():
workflow = StateGraph(ToolAgentState)
workflow.add_node("plan", plan_node)
workflow.add_node("execute", execute_node)
workflow.add_node("observe", observe_node)
workflow.add_node("decide", decide_node)
workflow.add_edge("plan", "execute")
workflow.add_edge("execute", "observe")
workflow.add_edge("observe", "decide")
def route_decision(state: ToolAgentState):
if state["plan"] and len(state["plan"]) > 1:
return "plan"
return "end"
workflow.add_conditional_edges(
"decide",
route_decision,
{
"plan": "plan",
"end": END
}
)
workflow.set_entry_point("plan")
return workflow.compile()
# 사용 예시
# app = create_tool_workflow()
# result = app.invoke({"messages": [HumanMessage(content="파일 시스템에서 모든 Python 파일을 검색해주세요")]})
4. 模板三：人介流程（暂止→察核→续行）
此模板乃处理需人力介入之作业之流程也.
python
from langgraph.graph import StateGraph, END, START
from langchain_core.messages import AIMessage
class HumanInLoopState(TypedDict):
messages: Annotated[list, operator.add]
action_needed: bool
user_feedback: str
processed_result: str
# 작업 생성 노드
def create_task_node(state: HumanInLoopState):
# 작업 생성 및 인간 개입 필요성 판단
prompt = ChatPromptTemplate.from_messages([
("system", "Generate task based on user request and determine if human review is needed."),
("human", "Request: {request}\nGenerate task:")
])
model = ChatOpenAI(model="gpt-4")
chain = prompt | model
response = chain.invoke({"request": state["messages"][-1].content})
# 간단한 규칙으로 인간 개입 필요성 판단
needs_review = "review" in response.content.lower() or "approve" in response.content.lower()
return {
"messages": [AIMessage(content=response.content)],
"action_needed":
---
📥 **Get the full guide on Gumroad**: https://gumroad.com/l/auto ($5)