LangGraphワークフローテンプレート（v6）

LangGraphは現代のAIエージェント開発のための強力なフレームワークであり、状態ベースのワークフローを実装するのに最適化されています。このガイドでは、実際の開発者がよく直面する問題を解決する5つの主要なテンプレートを提供します。

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. テンプレート1：簡単な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. テンプレート2: マルチ-ツールエージェント（計画→実行→観察→決定）

は、複数のツールを活用できる複雑なエージェントのためのワークフローです.

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. テンプレート3: 人間-仲介ワークフロー（一時停止→検討→続行）

は、人間の介入が必要な作業を処理するためのワークフローです.


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)

おすすめ購読元