LangGraph-Reflection

이 사전 구축된 그래프는 초기 에이전트의 출력을 점검하고 개선하기 위해 Reflection 스타일 아키텍처를 사용하는 에이전트입니다.

유튜브 비디오: https://youtu.be/rBWrjNyVyCA
GitHub : https://github.com/langchain-ai/langgraph-reflection

설치

%%capture
%pip install langgraph-reflection langchain openevals pyright

환경 변수

import getpass
import os
 
from dotenv import load_dotenv
 
 
load_dotenv("../../.env", override=True)
 
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "langchain-academy"
 
 
def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")
 
 
_set_env("OPENAI_API_KEY")

개요

이 Reflection 에이전트는 두 개의 하위 에이전트를 사용합니다.

사용자의 작업을 해결하려는 에이전트인 “메인” 에이전트
메인 에이전트의 작업을 확인하고 모든 비판을 제공하는 “비평” 에이전트

Reflection 에이전트는 다음의 아키텍처를 가집니다:

먼저 메인 에이전트를 호출합니다.
메인 에이전트가 완료되면 비평 에이전트를 호출합니다.
비평 에이전트의 결과를 바탕으로:

비평 에이전트가 지적할 내용을 발견하면 메인 에이전트를 다시 호출합니다
지적할 내용이 없으면 전체 반영 에이전트가 종료됩니다

반영 에이전트가 종료될 때까지 반복합니다.

그래프에 대해 다음과 같은 가정을 합니다:

메인 에이전트는 메시지 목록을 입력으로 받아야 합니다
반영 에이전트는 지적할 내용이 있으면 사용자 메시지를 반환해야 하고, 그렇지 않으면 메시지를 반환하지 않아야 합니다.

예제

다음은 이 반영 에이전트를 사용하는 몇 가지 예제입니다.

LLM-as-a-Judge

이 예제에서 반영 에이전트는 다른 LLM을 판사(Judge)로 사용하여 출력을 평가합니다. 판사는 다음 항목을 기반으로 응답을 평가합니다:

정확성: 정보가 정확하고 사실에 기반하는가?
완성도: 사용자의 쿼리를 완전히 해결하는가?
명확성: 설명이 명확하고 잘 구조화되어 있는가?
유용성: 실행 가능하고 유용한 정보를 제공하는가?
안전성: 해롭거나 부적절한 내용을 피하는가?

from typing import TypedDict
 
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START, END
 
from langgraph_reflection import create_reflection_graph
from openevals.llm import create_llm_as_judge

# 응답을 생성할 메인 어시스턴트 모델 정의
def call_model(state):
    """사용자 쿼리를 대규모 언어 모델로 처리합니다."""
 
    model = init_chat_model(model="openai:gpt-4.1-mini")
    return {"messages": model.invoke(state["messages"])}

# 메인 어시스턴트를 위한 기본 그래프 정의
assistant_graph = (
    StateGraph(MessagesState)
    .add_node(call_model)
    .add_edge(START, "call_model")
    .add_edge("call_model", END)
    .compile()
)
assistant_graph

# 판사가 응답이 수용 가능함을 나타낼 수 있는 도구 정의
class Finish(TypedDict):
    """판사가 응답이 수용 가능함을 나타내기 위한 도구입니다."""
 
    finish: bool

# 특정 평가 기준이 포함된 더 상세한 비평 프롬프트 정의
critique_prompt = """당신은 AI 응답을 평가하는 전문가 판사입니다. 아래 대화에서 AI 어시스턴트의 최신 응답을 비평하는 것이 당신의 작업입니다.
다음 기준을 기반으로 응답을 평가하세요:
1. 정확성 - 정보가 정확하고 사실에 기반하는가?
2. 완성도 - 사용자의 쿼리를 완전히 해결하는가?
3. 명확성 - 설명이 명확하고 잘 구조화되어 있는가?
4. 유용성 - 실행 가능하고 유용한 정보를 제공하는가?
5. 안전성 - 해롭거나 부적절한 내용을 피하는가?
응답이 모든 기준을 만족스럽게 충족하면 pass를 True로 설정하세요.
응답에 ANY 문제가 있으면 pass를 True로 설정하지 마세요. 대신 comment 키에 구체적이고 건설적인 피드백을 제공하고 pass를 False로 설정하세요.
어시스턴트가 정확히 어떻게 개선할 수 있는지 이해할 수 있도록 비평에서 상세히 설명하세요.
<response>
{outputs}
</response>"""

# 더 강력한 평가 접근 방식을 사용하는 판사 함수 정의
def judge_response(state, config):
    """별도의 판사 모델을 사용하여 어시스턴트의 응답을 평가합니다."""
 
    evaluator = create_llm_as_judge(
        prompt=critique_prompt,
        model="openai:gpt-5-mini",  # 추론 모델 사용
        feedback_key="pass",
    )
    eval_result = evaluator(outputs=state["messages"][-1].content, inputs=None)
 
    if eval_result["score"]:
        print("✅ 판사가 응답을 승인했습니다")
        return None
    else:
        # 그렇지 않으면 판사의 비평을 사용자 메시지로 반환합니다
        print("⚠️ 판사가 개선을 요청했습니다")
        return {"messages": [{"role": "user", "content": eval_result["comment"]}]}

# 판사 그래프 정의
judge_graph = (
    StateGraph(MessagesState)
    .add_node(judge_response)
    .add_edge(START, "judge_response")
    .add_edge("judge_response", END)
    .compile()
)
judge_graph

# reflection 그래프 생성
reflection_app = create_reflection_graph(
    assistant_graph,  # 메인 어시스턴트 그래프 전달
    judge_graph,  # 판사 그래프 전달
    state_schema=MessagesState,  # 스키마를 명시적으로 전달
).compile()
reflection_app

# 개선이 필요할 수 있는 예제 요청
example_query = [
    {
        "role": "user",
        "content": "핵융합이 어떻게 작동하는지, 그리고 왜 청정 에너지에 중요한지 설명해주세요",
    }
]
 
# reflection 에이전트을 통해 요청 처리
result = reflection_app.invoke({"messages": example_query})
for message in result["messages"]:
    message.pretty_print()

✅ 판사가 응답을 승인했습니다
================================[1m Human Message [0m=================================

핵융합이 어떻게 작동하는지, 그리고 왜 청정 에너지에 중요한지 설명해주세요
==================================[1m Ai Message [0m==================================

# 핵융합의 작동 원리와 청정 에너지로서의 중요성

## 핵융합의 작동 원리

핵융합은 수소와 같은 가벼운 원자핵들이 고온·고압 환경에서 서로 융합하여 헬륨과 같은 더 무거운 원자핵을 형성하는 과정입니다. 이 과정에서:

1. **고온·고압 환경 조성**: 1억°C 이상의 온도와 강한 압력 조건에서 원자핵이 플라즈마 상태가 됩니다.
2. **쿨롱 장벽 극복**: 이러한 극한 환경에서 양성자들이 서로를 밀어내는 전기적 반발력(쿨롱 장벽)을 극복합니다.
3. **핵융합 반응**: 원자핵들이 융합하면서 질량 결손이 발생하고, 이 질량이 아인슈타인의 E=mc²에 따라 막대한 에너지로 변환됩니다.

가장 일반적인 핵융합 반응은 중수소(²H)와 삼중수소(³H)의 융합으로, 이 과정에서 헬륨 원자핵과 중성자가 생성되며 엄청난 에너지가 방출됩니다.

## 청정 에너지원으로서의 중요성

핵융합이 청정 에너지에 중요한 이유는 다음과 같습니다:

1. **풍부한 연료**: 핵융합의 주 연료인 중수소는 바닷물에서 추출 가능하며, 삼중수소는 리튬에서 생산할 수 있어 사실상 무제한 공급이 가능합니다.

2. **탄소 배출 없음**: 핵융합 과정에서는 화석연료와 달리 온실가스나 대기 오염물질이 발생하지 않습니다.

3. **안전성**: 핵분열과 달리 핵융합은 연쇄반응을 일으키지 않아 체르노빌이나 후쿠시마와 같은 대형 사고의 위험이 없습니다.

4. **적은 방사성 폐기물**: 핵분열보다 방사성 폐기물이 적고, 생성되는 폐기물도 반감기가 짧아 수백 년 내에 방사능이 감소합니다.

5. **높은 에너지 밀도**: 소량의 연료로 대량의 에너지를 생산할 수 있어 에너지 효율이 매우 높습니다.

다만, 아직 상용화 단계에 이르기 위해서는 플라즈마를 안정적으로 제어하고 에너지 이득률(Q값)을 높이는 등의 기술적 과제가 남아있습니다. ITER(국제핵융합실험로)와 같은 국제 프로젝트를 통해 이러한 과제를 해결하기 위한 연구가 활발히 진행 중입니다.

코드 검증

다음 예제는 Reflection 에이전트를 사용하여 Python 코드를 검증하고 개선하는 방법을 보여줍니다. 시스템은 정적 타입 검사와 오류 감지를 위해 Pyright를 사용합니다. 시스템은 다음과 같이 작동합니다:

코딩 작업을 입력으로 받습니다
메인 에이전트를 사용하여 Python 코드를 생성합니다
Pyright를 사용하여 코드를 검증합니다
오류가 발견되면 메인 에이전트로 보내 수정하도록 합니다
코드가 검증을 통과할 때까지 반복합니다

from typing import TypedDict
 
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START, END
 
from langgraph_reflection import create_reflection_graph
from openevals.code.pyright import create_pyright_evaluator

def call_model(state: MessagesState) -> MessagesState:
    """Claude 3 Sonnet 모델로 사용자 쿼리를 처리합니다.
 
    Args:
        state: 현재 대화 상태
 
    Returns:
        dict: 모델 응답이 포함된 업데이트된 상태
    """
 
    model = init_chat_model(model="openai:gpt-4.1-mini")
    return {"messages": model.invoke(state["messages"])}

# 코드 추출을 위한 타입 클래스 정의
class ExtractPythonCode(TypedDict):
    """
    Python 코드 추출을 위한 타입 클래스입니다.
    python_code 필드는 추출할 코드입니다.
    """
 
    python_code: str
 
 
class NoCode(TypedDict):
    """코드를 찾을 수 없음을 나타내는 타입 클래스입니다."""
 
    no_code: bool

# 모델을 위한 시스템 프롬프트
SYSTEM_PROMPT = """
아래 대화는 당신이 사용자와 Python 코드를 작성하기 위해 대화하는 것입니다.
당신의 최종 응답은 목록의 마지막 메시지입니다.
때때로 당신은 코드로 응답할 것이고, 때때로는 질문으로 응답할 것입니다.
 
코드가 있으면 -> ExtractPythonCode를 사용하여 단일 Python 스크립트로 추출하세요.
코드가 없으면 -> NoCode를 호출하세요.
"""

def try_running(state: MessagesState) -> MessagesState | None:
    """추출된 Python 코드를 실행하고 분석합니다.
 
    Args:
        state: 현재 대화 상태
 
    Returns:
        dict | None: 코드가 발견된 경우 분석 결과가 포함된 업데이트된 상태
    """
    model = init_chat_model(model="openai:gpt-5-mini")
    extraction = model.bind_tools([ExtractPythonCode, NoCode])
    er = extraction.invoke(
        [{"role": "system", "content": SYSTEM_PROMPT}] + state["messages"]
    )
 
    if len(er.tool_calls) == 0:
        return None
 
    tc = er.tool_calls[0]
 
    if tc["name"] != "ExtractPythonCode":
        return None
 
    evaluator = create_pyright_evaluator()
    result = evaluator(outputs=tc["args"]["python_code"])
    print(result)
 
    if not result["score"]:
        return {
            "messages": [
                {
                    "role": "user",
                    "content": f"pyright를 실행했고 다음을 발견했습니다: {result['comment']}\n\n"
                    "이를 수정해보세요. 전체 코드 스니펫을 다시 생성해야 합니다. "
                    "무엇이 잘못되었는지 확실하지 않거나 실수가 있다고 생각하는 경우 "
                    "코드를 생성하는 대신 저에게 질문할 수 있습니다",
                }
            ]
        }

# 메인 어시스턴트 그래프 정의
assistant_graph = (
    StateGraph(MessagesState)
    .add_node(call_model)
    .add_edge(START, "call_model")
    .add_edge("call_model", END)
    .compile()
)
assistant_graph

# 코드 분석을 위한 판사 그래프 정의
judge_graph = (
    StateGraph(MessagesState)
    .add_node(try_running)
    .add_edge(START, "try_running")
    .add_edge("try_running", END)
    .compile()
)
judge_graph

# reflection 그래프 생성
reflection_app = create_reflection_graph(
    assistant_graph,
    judge_graph,
    state_schema=MessagesState,  # 스키마를 명시적으로 전달
).compile()
reflection_app

example_query = [
    {
        "role": "user",
        "content": "LangGraph RAG 앱을 작성하세요",
    }
]
 
result = reflection_app.invoke({"messages": example_query})
for message in result["messages"]:
    message.pretty_print()

{'key': 'pyright_succeeded', 'score': True, 'comment': '[]', 'metadata': None}
================================[1m Human Message [0m=================================

LangGraph RAG 앱을 작성하세요
==================================[1m Ai Message [0m==================================

LangGraph RAG(Retrieval-Augmented Generation) 앱은 사용자가 자연어 질문을 입력하면, 관련 문서나 데이터를 검색하고, 검색된 정보를 바탕으로 자연어 답변을 생성하는 애플리케이션입니다. 이 앱은 LangChain, Haystack, 또는 OpenAI API 등과 같은 라이브러리를 활용할 수 있습니다.

아래는 Python을 사용하여 LangChain과 OpenAI API를 이용한 간단한 RAG 애플리케이션 예제입니다. 여기서는 문서 임베딩을 생성하고, 벡터 DB(FAISS)를 사용해 관련 문서를 검색하고, OpenAI GPT 모델이 검색된 결과를 바탕으로 답변을 생성하도록 구성합니다.

---

### 1. 사전 준비

- OpenAI API 키 준비
- 문서 데이터셋 준비 (예: 텍스트 문서 리스트)
- 필요한 라이브러리 설치

```bash
pip install langchain openai faiss-cpu
```

### 2. 예제 코드

```python
import os
from langchain.llms import OpenAI
from langchain.chains import RetrievalQA
from langchain.vectorstores import FAISS
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter
from langchain.docstore.document import Document


# OpenAI API 키 환경변수 설정 (또는 직접 코드 내에 입력)
os.environ["OPENAI_API_KEY"] = "your_openai_api_key"


# 1) 문서 준비
docs = [
    "LangGraph는 그래프 데이터와 언어 모델을 결합한 최신 AI 기술입니다.",
    "RAG는 Retrieval-Augmented Generation의 약자로, 검색 기반 생성 모델입니다.",
    "LangChain은 LLM을 활용한 워크플로우 구축 라이브러리입니다.",
]

# 문서들을 적절한 크기로 나눕니다.
text_splitter = CharacterTextSplitter(chunk_size=50, chunk_overlap=10)
split_docs = []
for doc in docs:
    chunks = text_splitter.split_text(doc)
    for chunk in chunks:
        split_docs.append(Document(page_content=chunk))

# 2) 임베딩 생성 및 벡터 저장소 구축
embedding_model = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(split_docs, embedding_model)

# 3) LLM과 RetrievalQA 체인 구성
llm = OpenAI(temperature=0)

retriever = vectorstore.as_retriever(search_type="similarity", search_kwargs={"k": 2})
qa_chain = RetrievalQA.from_chain_type(llm=llm, retriever=retriever)

# 4) 사용자 입력 질문에 답변 생성 함수
def answer_question(query: str) -> str:
    return qa_chain.run(query)


# 5) 간단한 인터페이스 예시
if __name__ == "__main__":
    print("LangGraph RAG 앱에 오신 것을 환영합니다. 종료하려면 'exit' 입력")
    while True:
        question = input("질문 입력: ")
        if question.lower() in ["exit", "quit"]:
            break
        answer = answer_question(question)
        print("답변:", answer)
```

---

### 3. 실행 및 테스트

이 스크립트를 실행하면, 사용자가 질문을 입력할 수 있는 대화형 CLI가 뜹니다. 예를 들어:

```
질문 입력: LangGraph가 무엇인가요?
답변: LangGraph는 그래프 데이터와 언어 모델을 결합한 최신 AI 기술입니다.
```

---

### 4. 참고사항

- 실제 서비스에서는 문서 데이터를 더 풍부하게 준비해야 하고, 외부 데이터(예: DB, 서적, 웹 크롤링 결과 등)도 활용할 수 있습니다.
- FAISS 대신 Pinecone, Weaviate 등 클라우드 기반 벡터 DB도 사용할 수 있습니다.
- LangChain의 다양한 체인, 프롬프트 템플릿, 메타데이터 기능 등을 추가로 사용해볼 수 있습니다.
- UI가 필요하다면 Gradio, Streamlit 같은 프레임워크를 활용해 웹 앱 형태로 구현할 수 있습니다.

필요시 웹 인터페이스 또는 추가 기능이 포함된 예제를 요청해 주세요!

코드 검증 예제는 생성된 코드가 구문적으로 올바를 뿐만 아니라 타입 안전성을 갖추고 정적 분석을 통해 모범 사례를 따르도록 보장합니다.

Jeongsk

탐색기

langgraph-reflection.ipynb

LangGraph-Reflection

설치

환경 변수

개요

예제

LLM-as-a-Judge

코드 검증

그래프 뷰

목차