SQLite는 서버가 필요 없는 가볍고(self-contained) 독립적인 관계형 데이터베이스 관리 시스템(RDBMS)으로, 애플리케이션 내부에 파일 형태로 데이터베이스를 저장해 별도의 설치나 설정 없이 쉽게 사용할 수 있다는 특징이 있습니다. ANSI-SQL 표준을 대부분 지원하며, 트랜잭션, 인덱스, 뷰, 트리거 같은 기능도 제공하지만 MySQL이나 PostgreSQL 같은 대형 서버형 DBMS에 비해 동시성 처리나 고성능 분산 처리에는 한계가 있습니다. 작은 규모의 웹·모바일 앱, 프로토타입 개발, 내장형 기기 등에 널리 활용되며, 파일 하나만 복사하면 전체 데이터베이스를 옮길 수 있을 만큼 간편성이 뛰어납니다.
import requests
url = "https://storage.googleapis.com/benchmarks-artifacts/chinook/Chinook.db"
response = requests.get(url)
if response.status_code == 200:
with open("Chinook.db", "wb") as file:
file.write(response.content)
print("Chinook.db 다운로드")
else:
print(f"다운로드 실패: {response.status_code}")
!pip install langchain_community
from langchain_community.utilities import SQLDatabase
db = SQLDatabase.from_uri("sqlite:///Chinook.db")
print(f"Dialect: {db.dialect}")
print(f"Available tables: {db.get_usable_table_names()}")
print(f'Sample output: {db.run("SELECT * FROM Artist LIMIT 5;")}')
from langchain_community.utilities import SQLDatabase
import os
db_path = "./Chinook.db"
print("경로 존재 여부:", os.path.exists(db_path))
db = SQLDatabase.from_uri(f"sqlite:///{db_path}")
print(f"Dialect: {db.dialect}")
print(f"Available tables: {db.get_usable_table_names()}")
print(f'Sample output: {db.run("SELECT * FROM Artist LIMIT 5;")}')
import getpass
import os
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("OPENAI_API_KEY")
!pip install langchain_openai
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", temperature=0, max_retries=3)
sql_db_query: SQL 쿼리를 입력하면, 데이터베이스 조회 결과를 출력하는 도구로, 쿼리가 올바르지 않으면 오류 메시지가 반환됨.
sql_db_schema: 테이블 목록을 입력하면 해당 테이블의 스키마와 샘플 행이 출력되는 도구.
sql_db_list_tables: 데이터베이스 테이블 목록을 반환하는 도구.
sql_db_query_checker: 쿼리가 올바른지 확인하는 도구.
from langchain_community.agent_toolkits import SQLDatabaseToolkit
toolkit = SQLDatabaseToolkit(db=db, llm=llm)
tools = toolkit.get_tools()
for tool in tools:
print(f"{tool.name}: {tool.description}\n")
!pip install langgraph
from langgraph.prebuilt import create_react_agent
system_prompt = f"""
You are an agent designed to interact with a SQL database.
Given an input question, create a syntactically correct {db.dialect} query to run,
then look at the results of the query and return the answer. Unless the user
specifies a specific number of examples they wish to obtain, always limit your
query to at most 5 results.
You can order the results by a relevant column to return the most interesting
examples in the database. Never query for all the columns from a specific table,
only ask for the relevant columns given the question.
You MUST double check your query before executing it. If you get an error while
executing a query, rewrite the query and try again.
DO NOT make any DML statements (INSERT, UPDATE, DELETE, DROP etc.) to the
database.
To start you should ALWAYS look at the tables in the database to see what you
can query. Do NOT skip this step.
Then you should query the schema of the most relevant tables.
"""
agent = create_react_agent(
llm,
tools,
prompt=system_prompt,
)
agent
question = "2009년에 가장 많은 매출을 올린 영업 사원은 누구인가요?"
for step in agent.stream(
{"messages": [{"role": "user", "content": question}]},
stream_mode="values",
):
step["messages"][-1].pretty_print()
tools
2. Agent 구축
1. ToolNode 생성하기
from langgraph.prebuilt import ToolNode
list_tables_tool = next(tool for tool in tools if tool.name == "sql_db_list_tables")
list_tables_node = ToolNode([list_tables_tool], name="list_tables")
get_schema_tool = next(tool for tool in tools if tool.name == "sql_db_schema")
get_schema_node = ToolNode([get_schema_tool], name="get_schema")
run_query_tool = next(tool for tool in tools if tool.name == "sql_db_query")
run_query_node = ToolNode([run_query_tool], name="run_query")
2. DB 조회 시작을 결정할 첫 노드 생성하기
from langgraph.graph import END, START, MessagesState, StateGraph
def chatbot(state: MessagesState):
llm_with_tools = llm.bind_tools([list_tables_tool])
response = llm_with_tools.invoke(state["messages"]) # tool_calls : DB 조회가 필요 / AIMessage : DB 조회가 필요X
return {"messages": [response]}
from langchain_core.prompts import ChatPromptTemplate
generate_query_system_prompt = f"""
You are an agent designed to interact with a SQL database.
Given an input question, create a syntactically correct {db.dialect} query to run,
then look at the results of the query and return the answer. Unless the user
specifies a specific number of examples they wish to obtain, always limit your
query to at most 5 results.
You can order the results by a relevant column to return the most interesting
examples in the database. Never query for all the columns from a specific table,
only ask for the relevant columns given the question.
DO NOT make any DML statements (INSERT, UPDATE, DELETE, DROP etc.) to the database.
DO NOT wrap the response in any backticks or anything else(```). Respond with a SQL statement only!
"""
generate_query_user_prompt = """
User input: {question}
Schema: {schema}
If an error message is given, regenerate the query based on the error message.
History: {history}
SQL query:"""
def generate_query(state: MessagesState):
print("##### GENERATE QUERY #####")
print(state["messages"])
history = ""
for message in state["messages"][2:]:
history += message.content + "\n"
generate_query_msgs = [
("system", generate_query_system_prompt),
("user", generate_query_user_prompt),
]
generate_propmt = ChatPromptTemplate.from_messages(generate_query_msgs)
response = llm.invoke(generate_propmt.format_messages(
question=state["messages"][0].content,
schema=state["messages"][1].content,
history=history
))
print("generate_query", response)
return {"messages": [response]}
5. 생성한 쿼리문 실행하기
from langchain_core.messages import AIMessage
import re # 파일 상단 임포트
def _sanitize_sql(text: str) -> str:
m = re.search(r"```(?:sql)?\s*(.*?)```", text, flags=re.DOTALL | re.IGNORECASE)
return (m.group(1) if m else text).strip()
def check_query(state: MessagesState):
print("##### CHECK QUERY #####")
raw = state["messages"][-1].content
query = _sanitize_sql(raw) # <-- 코드펜스 제거 포인트
print("check_query", query)
response = run_query_tool.invoke({"query": query})
return {"messages": [AIMessage(content=str(response))]}
6. 질문 + 쿼리 실행 결과 기반 답변 생성하기
answer_system_prompt = """
You are a highly intelligent assistant trained to provide concise and accurate answers.
You will be given a context that has been retrieved from a database using a specific SQL query.
Your task is to analyze the context and answer the user’s question based on the information provided in the context.
ANSWER IN KOREAN.
"""
def answer(state: MessagesState):
print("##### ANSWER #####")
question = state["messages"][0].content
context = state["messages"][-1].content
generated_query = state["messages"][-2].content
print("context", context)
answer_msgs = [
("system", answer_system_prompt),
("user", "User Question: {question} SQL Query: {generated_query} Context: {context}"),
]
answer_propmt = ChatPromptTemplate.from_messages(answer_msgs)
response = llm.invoke(answer_propmt.format_messages(question=question, generated_query=generated_query, context=context))
print("response", response)
return {"messages": [response]}
graph_builder.add_node("call_get_schema", call_get_schema)
graph_builder.add_node("get_schema", get_schema_node)
graph_builder.add_node("generate_query", generate_query)
graph_builder.add_node("check_query", check_query)
graph_builder.add_node("answer", answer)
graph_builder.add_edge("list_tables", "call_get_schema")
graph_builder.add_edge("call_get_schema", "get_schema")
graph_builder.add_edge("get_schema", "generate_query")
graph_builder.add_edge("generate_query", "check_query")
def should_correct(state):
print(state["messages"][-1].content)
txt = state["messages"][-1].content
if "Error:" in txt or "error" in txt.lower(): # DB 조회 결과가 [-1] 에 저장
return "generate_query"
else:
return "answer"
graph_builder.add_conditional_edges(
"check_query",
should_correct,
{
"generate_query": "generate_query",
"answer": "answer",
},
)
graph_builder.add_edge("answer", END)
graph = graph_builder.compile()
graph
question = "2009년에 가장 많은 매출을 올린 영업 사원은 누구인가요?"
for step in graph.stream(
{"messages": [{"role": "user", "content": question}]},
stream_mode="values",
):
step["messages"][-1].pretty_print()
question = "DB 조회를 기반으로 2009년 가장 많은 양의 음반을 판매한 아티스트의 해당 앨범 판매 기간을 알려주세요. "
for step in graph.stream(
{"messages": [{"role": "user", "content": question}]},
stream_mode="values",
):
step["messages"][-1].pretty_print()
벡터 데이터베이스(Vector Database)는 텍스트, 이미지, 오디오와 같은 데이터를 고차원 벡터 형태로 변환해 저장하고, 이 벡터 간의 유사도를 빠르게 검색할 수 있도록 최적화된 데이터베이스입니다. 일반적인 관계형 데이터베이스가 정확한 값 기반 검색(SQL 쿼리 등)에 적합하다면, 벡터 데이터베이스는 의미적 유사성(semantic similarity)에 기반한 검색을 지원하여 예를 들어 "강아지"와 "개"처럼 다른 표현이라도 비슷한 의미의 데이터를 찾아낼 수 있습니다. 이를 위해 코사인 유사도, 내적(dot product), 유클리드 거리와 같은 수학적 거리 계산을 활용하며, 대규모 임베딩(embedding) 데이터를 효율적으로 관리하고 검색할 수 있어 추천 시스템, 검색 엔진, 생성형 AI의 RAG(Retrieval-Augmented Generation) 등에 널리 활용됩니다.
크로마디비
크로마디비(ChromaDB)는 대표적인 오픈소스 벡터 데이터베이스로, 문서·이미지·코드 등 다양한 데이터를 임베딩 벡터로 변환해 저장하고, 이를 빠르게 검색할 수 있도록 설계된 시스템입니다. 파이썬 기반으로 사용이 간편하며, LangChain 같은 LLM 프레임워크와 잘 통합되어 RAG(Retrieval-Augmented Generation) 구조를 쉽게 구축할 수 있습니다. 내부적으로는 벡터 인덱싱과 메타데이터 저장을 함께 지원하여, 단순히 유사도 검색뿐 아니라 조건 필터링과 결합된 검색도 가능합니다. 무료로 가볍게 실행할 수 있고, 로컬 환경부터 클라우드까지 유연하게 확장할 수 있어 학습용이나 실무용 AI 검색 엔진 구축에 많이 활용됩니다.
import getpass
import os
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("OPENAI_API_KEY")
청크(Chunk)는 긴 텍스트나 문서를 작은 단위로 나눈 조각을 의미하며, 주로 자연어 처리와 RAG(Retrieval-Augmented Generation) 같은 작업에서 사용됩니다. 대형 언어모델은 한 번에 처리할 수 있는 토큰 수에 한계가 있기 때문에 문서를 일정한 길이로 분할하여 임베딩 벡터로 변환하고, 이후 검색이나 질의 응답 시 필요한 청크만 불러와 모델에 전달하는 방식으로 효율성과 정확성을 높입니다. 청크는 단순히 일정 글자 수나 토큰 수로 나누기도 하지만, 문단·문장 단위 등 의미 단위로 나누어야 검색 품질이 좋아지며, 결국 청크는 방대한 데이터를 모델이 다룰 수 있는 크기로 잘게 나눈 최소 단위라고 할 수 있습니다.
SemanticChunker
SemanticChunker는 텍스트를 단순히 일정한 길이로 자르는 방식이 아니라, 문장의 의미적 맥락을 고려해 자연스럽게 분할하는 청크 생성 기법입니다. 즉, 문장이나 문단의 의미가 단절되지 않도록 문맥 단위로 텍스트를 나누어 임베딩과 검색의 정확도를 높여줍니다. 이를 통해 RAG(Retrieval-Augmented Generation) 구조에서 모델이 보다 관련성 높은 정보를 검색할 수 있으며, 불필요하게 잘려나간 조각이나 중복된 정보 전달을 줄일 수 있습니다. 따라서 SemanticChunker는 의미 기반 검색과 대규모 문서 처리에서 효율성과 정밀도를 동시에 향상시키는 중요한 도구로 활용됩니다.
from langchain_community.document_loaders import PyPDFLoader
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai.embeddings import OpenAIEmbeddings
file_path = "SPRi AI Brief_4월호_산업동향_250407_F.pdf"
loader = PyPDFLoader(file_path)
pages = []
async for page in loader.alazy_load():
pages.append(page)
text_splitter = SemanticChunker(OpenAIEmbeddings())
docs = text_splitter.split_documents(pages)
print(f"총 {len(docs)}개 만큼의 문서로 청킹되었습니다.")
print([len(i.page_content) for i in docs])
# 각 청크의 메타데이터 및 내용 출력
for i in docs:
print(i.metadata) # 문서의 메타데이터 출력 (예: 페이지 번호 등)
print(i.page_content) # 분할된 청크의 내용 출력
print("-" * 100) # 구분선 출력
3. 벡터 리트리버
벡터 리트리버(Vector Retriever)는 사용자의 질의(Query)를 임베딩 벡터로 변환한 뒤, 벡터 데이터베이스에 저장된 청크(Chunk) 벡터들과의 유사도를 계산하여 가장 관련성 높은 결과를 찾아주는 구성 요소입니다. 즉, “검색기” 역할을 하는데, 단순히 키워드 일치를 찾는 것이 아니라 의미적 유사성을 기반으로 정보를 불러옵니다. 이를 위해 코사인 유사도, 내적(dot product), 유클리드 거리 등의 수학적 방법을 활용하며, 검색된 결과는 LLM과 결합되어 RAG(Retrieval-Augmented Generation) 같은 구조에서 모델의 답변 품질을 높이는 데 쓰입니다. 쉽게 말해, 벡터 리트리버는 “의미를 이해하는 검색 엔진”이라고 할 수 있습니다.
앙상블 리트리버(Ensemble Retriever)는 하나의 검색 방식에만 의존하지 않고, 여러 종류의 리트리버를 조합해 더 정확하고 풍부한 검색 결과를 제공하는 방법입니다. 예를 들어 키워드 기반의 전통적 BM25 리트리버와 의미 기반의 벡터 리트리버를 함께 사용하면, 단어가 정확히 일치하는 문서뿐 아니라 의미적으로 관련 있는 문서도 함께 찾아낼 수 있습니다. 이렇게 서로 다른 리트리버의 강점을 결합하면 검색 누락을 줄이고, 다양한 관점에서 문서를 확보할 수 있어 RAG(Retrieval-Augmented Generation) 구조에서 더욱 신뢰도 높은 응답을 생성하는 데 유용합니다.
BM25 리트리버
BM25 리트리버는 전통적인 정보 검색 기법 중 하나로, 사용자의 질의(Query)와 문서 간의 키워드 일치 정도를 계산해 관련성이 높은 문서를 찾아주는 방식입니다. 기본적으로 단어 빈도(Term Frequency), 역문서 빈도(Inverse Document Frequency), 그리고 문서 길이를 고려해 점수를 매기며, 특정 단어가 질의에 많이 등장하거나 드문 단어일수록 가중치를 높게 주어 검색 정확도를 높입니다. 벡터 리트리버처럼 의미적 유사성을 직접 파악하지는 못하지만, 빠르고 해석 가능한 결과를 제공하기 때문에 대규모 문서 검색이나 키워드 중심 검색에서 여전히 많이 활용되며, 종종 벡터 리트리버와 결합해 앙상블 리트리버로 사용됩니다.
ensemble_result = ensemble_retriever.invoke(query)
bm25_result = bm25_retriever.invoke(query)
vector_result = vector_retriever.invoke(query)
print("[Ensemble Retriever]")
for doc in ensemble_result:
print(f"Content: {doc.page_content}")
print()
print("[BM25 Retriever]")
for doc in bm25_result:
print(f"Content: {doc.page_content}")
print()
print("[Vector Retriever]")
for doc in vector_result:
print(f"Content: {doc.page_content}")
print()
!pip install langgraph
from langgraph.graph import StateGraph, MessagesState
class State(MessagesState):
context: str
graph_builder = StateGraph(State)
from langchain_core.messages import HumanMessage
def retriever(state: State):
"""
Retrieve the relevant document and return the content.
"""
print("##### RETRIEVER #####")
query = state["messages"][0].content
ensemble_result = ensemble_retriever.invoke(query)
content = ensemble_result[0].page_content
print("[CONTEXT]\n", content)
return {"context" : content, "messages": [HumanMessage(content=content)]}
4. 랭체인 허브
랭체인 허브(LangChain Hub)는 개발자와 연구자들이 프롬프트(Prompt), 체인(Chain), 에이전트(Agent) 같은 LLM 관련 리소스를 공유하고 재사용할 수 있도록 만든 오픈 플랫폼입니다. 사용자는 자신이 만든 프롬프트 템플릿을 업로드해 다른 사람과 공유할 수 있고, 다른 사람이 만든 검증된 프롬프트를 hub.pull() 같은 방식으로 손쉽게 불러와 활용할 수 있습니다. 이를 통해 매번 새롭게 프롬프트를 설계할 필요 없이 빠르게 실험하고 협업할 수 있으며, 다양한 검색 기능을 제공해 특정 목적(Q&A, 요약, RAG 등)에 맞는 템플릿을 쉽게 찾을 수 있습니다. 결국 랭체인 허브는 LLM 애플리케이션 개발을 가속화하는 중앙 저장소 역할을 합니다.
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain import hub
llm = ChatOpenAI(model="gpt-5-nano", temperature=0)
def answer(state: State):
"""
Answer the question based on the retrieved document.
"""
print("##### ANSWER #####")
query = state["messages"][0].content
context = state["messages"][-1].content
# context = state["context"]
# prompt = ChatPromptTemplate.from_messages(
# [
# ("system",
# """
# You are an assistant for answering questions based on retrieved document context.
# Answer in Korean.
# Context: {context}"""
# ),
# ("human", "{question}"),
# ]
# )
prompt = hub.pull("rlm/rag-prompt")
# You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know. Use three sentences maximum and keep the answer concise.
# Question: {question}
# Context: {context}
# Answer:
response = llm.invoke(
prompt.format_messages(context=context, question=query)
)
return {"messages": [response]}
hub.pull("rlm/rag-prompt").pretty_print()
from langgraph.graph import START, END
graph_builder.add_sequence([retriever, answer])
graph_builder.add_edge(START, "retriever")
graph_builder.add_edge("answer", END)
graph = graph_builder.compile()
graph
response = graph.invoke({"messages": "카나나의 테크니컬 리포트는 어떤 내용인가요?"})
for mes in response['messages']:
mes.pretty_print()
from langchain.tools.retriever import create_retriever_tool
retriever_tool = create_retriever_tool(
ensemble_retriever,
"retrieve_AI_brief",
"Search and return information about AI Technology and Industry.",
)
tools = [retriever_tool]
from langgraph.graph import StateGraph, MessagesState
graph_builder = StateGraph(MessagesState)
from langgraph.prebuilt import ToolNode, tools_condition
tool_node = ToolNode(tools=tools)
graph_builder.add_node("retriever", tool_node)
from langgraph.graph import MessagesState
class State(MessagesState):
"""
Represents the state of our graph.
Attributes:
question: question
generation: LLM generation
document: retrieved document
"""
question: str
generation: str
document: str
2. Retriever 호출을 위한 Agent 와 Retriever
from langchain.tools.retriever import create_retriever_tool
retriever_tool = create_retriever_tool(
retriever,
"retrieve_ai_policy_april_2025",
"Search and return information from the April 2025 AI Policy Report, including global AI policies, legal regulations, major industry updates, technical research, education trends, and government strategies related to AI development.",
)
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
def agent(state: State):
"""
Invokes the agent model to generate a response based on the current state. Given
the question, it will decide to retrieve using the retriever tool, or simply end.
"""
print("##### HI ! #####")
messages = state["messages"]
llm_with_tools = llm.bind_tools([retriever_tool]) # , tool_choice="any"
response = llm_with_tools.invoke(messages)
print("response", response)
return {"messages": [response], "question" : messages[0].content}
from pydantic import BaseModel, Field
class Grade(BaseModel):
"""Binary score for relevance check."""
binary_score: str = Field(description="Relevance score 'yes' or 'no'")
from langchain import hub
from langchain_core.prompts import ChatPromptTemplate
def grade_documents(state: State):
"""
Determines whether the retrieved document is relevant to the question.
"""
print("##### CHECK RELEVANCE #####")
grader = llm.with_structured_output(Grade)
grader_prompt = ChatPromptTemplate.from_template(
"""
You are a grader assessing relevance of a retrieved document to a user question. \n
Here is the retrieved document: \n\n {context} \n\n
Here is the user question: {question} \n
If the document contains keyword(s) or semantic meaning related to the user question, grade it as relevant. \n
It does not need to be a stringent test. The goal is to filter out erroneous retrievals. \n
Give a binary score 'yes' or 'no' score to indicate whether the document is relevant to the question.
"""
)
chain = grader_prompt | grader
question = state["question"]
print("state[messages]", state["messages"])
document = state["document"]
print("question", question)
print("context", document)
score = chain.invoke({"question": question, "context": document})
print("context", document)
grade = score.binary_score
print("grade", grade)
if grade == "yes":
print("---GRADE: DOCUMENT RELEVANT---")
return {"document": document, "question": question}
else:
print("---GRADE: DOCUMENT NOT RELEVANT---")
return {"document": "", "question": question}
4. 답변 생성
def generate(state: State):
"""
Generate answer based on the retrieved document and the question.
"""
print("##### GENERATE #####")
question = state["question"]
document = state["document"]
print("question", question)
print("context", document)
prompt = hub.pull("rlm/rag-prompt")
response = llm.invoke(
prompt.format_messages(context=document, question=question)
)
print("response", response)
return {"documents": document, "question": question, "generation": response.content, "messages": [response]}
5. 질문 재작성
from langchain_core.messages import HumanMessage
def transform_query(state):
"""
Transform the query to produce a better question.
Args:
state (dict): The current graph state
Returns:
state (dict): Updates question key with a re-phrased question
"""
print("##### TRANSFORM QUERY #####")
question = state["question"]
system = """You a question re-writer that converts an input question to a better version that is optimized \n
for vectorstore retrieval. Look at the input and try to reason about the underlying semantic intent / meaning."""
re_write_prompt = ChatPromptTemplate.from_messages(
[
("system", system),
(
"user",
"Here is the initial question: \n\n {question} \n Formulate an improved question In Korean. ",
),
]
)
question_rewriter = re_write_prompt | llm
better_question = question_rewriter.invoke({"question": question})
print("question", question)
print("better_question", better_question.content)
return {"question": better_question.content, "messages": [better_question]}
6. Web Search Agent
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
web_search_agent = create_react_agent(llm, tools=[TavilySearchResults(max_results=3)])
7. 답변 생성 여부 결정
def decide_to_generate(state):
"""
Determines whether to generate an answer, or re-generate a question.
"""
print("##### ASSESS GRADED DOCUMENTS #####")
if state["document"] == "":
print(
"---DECISION: RETRIEVED DOCUMENT ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---"
)
return "transform_query"
else:
print("---DECISION: GENERATE---")
return "generate"
8. 답변 평가
답변 해결성 평가 (답변과 질문을 비교)
class GradeAnswer(BaseModel):
"""Binary score to assess answer addresses question."""
binary_score: str = Field(
description="Answer addresses the question, 'yes' or 'no'"
)
structured_llm_grader = llm.with_structured_output(GradeAnswer)
system = """You are a grader assessing whether an answer addresses / resolves a question \n
If ambiguous, return no. \n
Give a binary score 'yes' or 'no'. Yes' means that the answer resolves the question."""
answer_prompt = ChatPromptTemplate.from_messages(
[
("system", system),
("user", "User question: \n\n {question} \n\n LLM generation: {generation}"),
]
)
answer_grader = answer_prompt | structured_llm_grader
환각 평가 (문서와 답변을 비교하여 팩트 체크)
class GradeHallucinations(BaseModel):
"""Binary score for hallucination present in generation answer."""
binary_score: str = Field(
description="Answer is grounded in the facts, 'yes' or 'no'"
)
structured_llm_grader = llm.with_structured_output(GradeHallucinations)
system = """You are a grader assessing whether an LLM generation is grounded in / supported by a set of retrieved facts. \n
Give a binary score 'yes' or 'no'. 'Yes' means that the answer is grounded in / supported by the set of facts."""
hallucination_prompt = ChatPromptTemplate.from_messages(
[
("system", system),
("user", "Set of facts: \n\n {document} \n\n LLM generation: {generation}"),
]
)
hallucination_grader = hallucination_prompt | structured_llm_grader
환각 발생 :not supported
환각 X, 답변이 질문을 해결 X :not useful
환각 X, 답변이 질문을 해결 O :useful
def grade_generation_v_documents_and_question(state):
"""
Determines whether the generation is grounded in the document and answers question.
"""
print("##### CHECK HALLUCINATIONS #####")
question = state["question"]
document = state["document"]
generation = state["generation"]
score = hallucination_grader.invoke(
{"document": document, "generation": generation}
)
grade = score.binary_score
if grade == "yes": # 환각 없음
print("---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---")
score = answer_grader.invoke({"question": question, "generation": generation})
grade = score.binary_score
if grade == "yes": # 답변이 질문을 해결함
print("---DECISION: GENERATION ADDRESSES QUESTION---")
return "useful"
else: # 답변이 질문을 해결하지 않음
print("---DECISION: GENERATION DOES NOT ADDRESS QUESTION---")
return "not useful"
else: # 환각 있음
print("---DECISION: GENERATION IS NOT GROUNDED IN DOCUMENTS, RE-TRY---")
return "not supported"
from langgraph.errors import GraphRecursionError
inputs = {
"messages": "2024년 노벨문학상 수상자는 누구인가요?"
}
try:
response = graph.invoke(inputs)
except GraphRecursionError as e:
print("Recursion Error")
for msg in response["messages"]:
msg.pretty_print()
case 2: 내부 문서 검색이 필요한 경우
inputs = {
"messages": "카카오가 개발한 모델에 대해 알려주세요."
}
try:
response = graph.invoke(inputs)
except GraphRecursionError as e:
print("Recursion Error")
for msg in response["messages"]:
msg.pretty_print()
inputs = {
"messages": "마누스 AI 에이전트에 대한 장단점"
}
try:
response = graph.invoke(inputs)
except GraphRecursionError as e:
print("Recursion Error")
for msg in response["messages"]:
msg.pretty_print()
case 3: 부실한 사용자 쿼리를 재작성한 후 재 검색 및 답변
inputs = {
"messages": "구글 AI AI AI"
}
try:
response = graph.invoke(inputs)
except GraphRecursionError as e:
print("Recursion Error")
for msg in response["messages"]:
msg.pretty_print()
Reflection은 에이전트가 스스로 결과를 평가·비판 한 뒤 그 피드백을 상태(state)에 기록하고, 필요하면 수정 루프로 되돌아가 답을 개선하는 설계 패턴입니다. 보통 “작성 노드(답 생성) → 리플렉션 노드(자기평가) → 라우팅(조건부 엣지)”로 구성되며, 리플렉션 노드는 품질 기준(예: 정확성, 근거, 형식)을 점수·코멘트(score, critique)로 남깁니다. 라우터는 이 정보를 읽어 임계값 미달이면 작성 노드로 되감기, 충족하면 종료 노드로 이동합니다. 무한 루프를 막기 위해 max_iters 같은 반복 한도를 두며, 툴 호출과는 별개로 LLM의 자기검토 능력을 활용해 코드 생성, 질의응답, 체인드 리저닝 등의 정확도·일관성을 높이는 데 쓰입니다.
import getpass
import os
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("OPENAI_API_KEY")
!pip install langchain_openai
1. 가사 생성
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"당신은 5단락 노래가사를 훌륭하게 작성하는 작사 도우미입니다."
"사용자의 요청에 따라 최고의 가사를 작성하세요."
"사용자가 피드백을 제공할 경우, 이전 시도에서 개선된 수정본을 작성해 응답하세요.",
),
MessagesPlaceholder(variable_name="messages"),
]
)
llm = ChatOpenAI(model="gpt-5-nano")
generate = prompt | llm
lyric = ""
request = HumanMessage(
content="코딩에 대한 가사를 작성해주세요."
)
for chunk in generate.stream({"messages": [request]}):
print(chunk.content, end="")
lyric += chunk.content
2. 가사 개선
reflection_prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"당신은 가사를 채점하는 작사가입니다. 사용자가 제출한 작사에 대한 비평과 개선 사항을 작성하세요."
"가사의 길이, 깊이, 문체 등을 포함해 구체적인 개선 요청을 제공하세요.",
),
MessagesPlaceholder(variable_name="messages"),
]
)
reflect = reflection_prompt | llm
reflection = ""
for chunk in reflect.stream({"messages": [request, HumanMessage(content=lyric)]}):
print(chunk.content, end="")
reflection += chunk.content
for chunk in generate.stream(
{"messages": [request, AIMessage(content=lyric), HumanMessage(content=reflection)]}
):
print(chunk.content, end="")
3. Graph로 Reflection 구현
!pip install langgraph
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import END, StateGraph, START
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
class State(TypedDict):
messages: Annotated[list, add_messages]
from typing import Literal
from langgraph.graph import END
def should_continue(state: State) -> Literal["reflect", END]:
if len(state["messages"]) > 6:
return END
return "reflect"
graph_builder.add_conditional_edges("generate", should_continue)
“Reflexion: Language Agents with Verbal Reinforcement Learning”은, 2023년 3월 20일 최초 제출, 2023년 10월 10일 v4로 개정된 논문입니다. 저자는 Noah Shinn 외 5명이고, 핵심 내용은 언어 에이전트가 스스로 언어적 피드백(반성문)을 생성·메모리에 저장해 다음 시도에 반영함으로써 성능을 높이는 프레임워크를 제안했다는 점입니다. HumanEval 등에서 유의미한 향상을 보고합니다.
Experience (long-term memory): 누적된 반성문을 쌓아두는 장기 메모리(mem)입니다. 이후 시도에서 프롬프트에 이 기억을 넣어 같은 실수를 반복하지 않게 합니다.
Environment: 외부에서 관찰/보상(예: 유닛 테스트의 통과/실패, 웹툴의 응답 등)을 제공합니다. 외부 피드백이 있으면 Evaluator의 판단과 함께 사용됩니다.
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("TAVILY_API_KEY")
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-5-nano")
!pip install langchain_community
!pip install langchain-tavily
from langchain_tavily import TavilySearch
tavily_tool = TavilySearch(max_results=5)
1. 필요한 데이터 클래스 정의
Reflection- 놓친것 / 불필요한 것
from langchain_core.messages import HumanMessage, ToolMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from pydantic import BaseModel, Field
class Reflection(BaseModel):
missing: str = Field(description="누락되거나 부족한 부분에 대한 비평")
superfluous: str = Field(description="불필요한 부분에 대한 비평")
AnswerQuestion- 답변 / 답변에 대한 반성 / 개선하기 위한 검색 쿼리
class AnswerQuestion(BaseModel):
answer: str = Field(description="질문에 대한 10문장 이내의 자세한 답변")
search_queries: list[str] = Field(
description="현재 답변에 대한 비평을 해결하기 위한 추가 조사를 위한 1~3개의 웹 검색 쿼리"
)
reflection: Reflection = Field(description="답변에 대한 자기반성 내용")
import datetime
actor_prompt_template = ChatPromptTemplate.from_messages(
[
(
"system",
"""당신은 전문 연구자입니다.
1. {first_instruction}
2. <Reflect> 생성한 답변을 다시 되돌아보고 개선할 수 있도록 비판하세요.
3. <Recommend search queries> 답변의 질을 높이기 위해 추가적으로 조사해야 할 정보에 대한 웹 검색 쿼리를 추천하세요.""",
),
MessagesPlaceholder(variable_name="messages"),
(
"user",
"\n\n<Reflect> 사용자 원래 질문과 지금까지의 행동을 되돌아보세요."
),
]
)
initial_answer_chain = actor_prompt_template.partial(
first_instruction="질문에 대한 10문장 이내의 자세한 답변을 제공해주세요.", # 초기 답변
) | llm.bind_tools(tools=[AnswerQuestion], tool_choice="any")
class ReviseAnswer(AnswerQuestion):
"""Revise your original answer to your question. Provide an answer, reflection,
cite your reflection with references, and finally
add search queries to improve the answer."""
references: list[str] = Field(
description="업데이트된 답변에 사용된 인용 출처"
)
revise_instructions = """이전 답변을 새로운 정보를 바탕으로 수정하세요.
- 이전 비평 내용을 활용해 중요한 정보를 추가해야 합니다.
- 수정된 답변에는 반드시 숫자로 된 인용 표시를 포함하여 검증 가능하도록 해야 합니다.
- 답변 하단에 "참고문헌" 섹션을 추가하세요 (이 부분은 단어 수 제한에 포함되지 않습니다). 형식은 다음과 같습니다:
- [1] https://example.com
- [2] https://example.com
- 이전 비평 내용을 바탕으로 불필요한 정보를 제거하고, 최종 답변은 반드시 200자를 넘지 않도록 하세요.
"""
revision_chain = actor_prompt_template.partial(
first_instruction=revise_instructions,
) | llm.bind_tools(tools=[ReviseAnswer], tool_choice="any")
revisor = Responder(runnable=revision_chain)
from langchain_core.tools import StructuredTool
from langgraph.prebuilt import ToolNode
def run_queries(search_queries: list[str], **kwargs):
"""Run the generated queries."""
return tavily_tool.batch([{"query": query} for query in search_queries])
tool_node = ToolNode(
[
StructuredTool.from_function(run_queries, name=AnswerQuestion.__name__),
StructuredTool.from_function(run_queries, name=ReviseAnswer.__name__),
]
)
5. 그래프 생성하기
from langgraph.graph import END, StateGraph, START
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDict
class State(TypedDict):
messages: Annotated[list, add_messages]
MAX_ITERATIONS = 5
graph_builder = StateGraph(State)
graph_builder.add_node("draft", first_responder.respond)
graph_builder.add_node("execute_tools", tool_node) # 웹 검색 진행
graph_builder.add_node("revise", revisor.respond)
graph_builder.add_edge("draft", "execute_tools")
graph_builder.add_edge("execute_tools", "revise")
def _get_num_iterations(state: list):
i = 0
for m in state[::-1]:
if m.type not in {"tool", "ai"}:
break
i += 1
return i
def event_loop(state: list):
num_iterations = _get_num_iterations(state["messages"])
if num_iterations > MAX_ITERATIONS:
return END
return "execute_tools"
graph_builder.add_conditional_edges("revise", event_loop, ["execute_tools", END])
graph_builder.add_edge(START, "draft")
graph = graph_builder.compile()
graph
events = graph.stream(
{"messages": [HumanMessage(content="AI Agent가 무엇인가요?")]},
stream_mode="values",
)
for i, step in enumerate(events):
print(f"Step {i}")
step["messages"][-1].pretty_print()
3. Plan & Execute
import getpass
import os
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("OPENAI_API_KEY")
_set_env("TAVILY_API_KEY")
!pip install langchain_community
from langchain_community.tools.tavily_search import TavilySearchResults
# ReAct 패턴
# 추론 -> 행동 -> 관찰 -> 답변, 자동으로 생성
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
llm = ChatOpenAI(model="gpt-5-nano")
prompt = "You are a helpful assistant."
plan_executor = create_react_agent(llm, tools, prompt=prompt)
plan_executor
plan_executor.invoke({'messages':[('user','2025년 한국의 최저 시급은 얼마입니까?')]})
from pydantic import BaseModel, Field
from typing import List
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
class Plan(BaseModel):
"""Plan to follow in future"""
steps: List[str] = Field(
# 단계별 진행 예정
description="different steps to follow, should be in sorted order"
)
planner_prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"""For the given objective, come up with a simple step by step plan. \
This plan should involve individual tasks, that if executed correctly will yield the correct answer. Do not add any superfluous steps. \
The result of the final step should be the final answer. Make sure that each step has all the information needed - do not skip steps.""",
),
MessagesPlaceholder(variable_name="messages"),
]
)
planner = planner_prompt | llm.with_structured_output(Plan)
plan_result = planner.invoke(
{
"messages": [HumanMessage(
content="2025년 한국에서 개봉한 영화 중 가장 흥행한 영화는 무엇인가요?",
)]
}
)
plan_result.steps
replanner_prompt = ChatPromptTemplate.from_template(
"""For the given objective, come up with a simple step by step plan. \
This plan should involve individual tasks, that if executed correctly will yield the correct answer. Do not add any superfluous steps. \
The result of the final step should be the final answer. Make sure that each step has all the information needed - do not skip steps.
Your objective was this:
{input}
Your original plan was this:
{plan}
You have currently done the follow steps:
{past_steps}
Update your plan accordingly.
If no more steps are needed and you can return to the user, then respond with that.
Otherwise, fill out the plan. Only add steps to the plan that still NEED to be done.
Do not return previously done steps as part of the plan."""
)
from typing import Union
class Response(BaseModel):
"""Response to user."""
response: str
class Act(BaseModel):
"""Action to perform."""
action: Union[Response, Plan] = Field(
description="Action to perform. If you want to respond to user, use Response. "
"If you need to further use tools to get the answer, use Plan."
)
import operator
from typing import Annotated, List, Tuple
from typing_extensions import TypedDict
class PlanExecute(TypedDict):
input: str
plan: List[str]
past_steps: Annotated[List[Tuple], operator.add]
response: str
# 계획 생성
def plan_step(state: PlanExecute): # 계획을 생성하는 노드
plan = planner.invoke({"messages": [("user", state["input"])]})
return {"plan": plan.steps}
# 계획 실행
def execute_step(state: PlanExecute):
plan = state["plan"]
plan_str = "\n".join(f"{i+1}. {step}" for i, step in enumerate(plan))
task = plan[0]
task_formatted = f"""For the following plan:
{plan_str}\n\nYou are tasked with executing step {1}, {task}."""
agent_response = plan_executor.invoke(
{"messages": [("user", task_formatted)]}
)
return {
"past_steps": [(task, agent_response["messages"][-1].content)], # 실행 완료한 계획과 결과 저장
}
# 계획 수정
def replan_step(state: PlanExecute):
output = replanner.invoke(state)
if isinstance(output.action, Response): # 답변이 바로 가능한 상태
return {"response": output.action.response}
else:
return {"plan": output.action.steps}
from langgraph.graph import END
def should_end(state: PlanExecute):
if "response" in state and state["response"]:
return END
else:
return "agent"
config = {'recursion_limit':50}
inputs = {'input' :'2024년 노벨 문학상 수상자의 출신국가는 어디인가요?'}
for event in graph.stream(inputs, config=config, stream_mode='values'):
for k,v in event.items():
print(k,v)
print('*'*50)
4. 코드 수정을 반복하는 데이터 전처리 Agent
1. 클로드
클로드(Claude)는 앤트로픽(Anthropic)에서 개발한 대규모 언어 모델(LLM) 기반의 인공지능 챗봇으로, 사람과의 대화, 글쓰기, 요약, 코드 작성 등 다양한 작업을 수행할 수 있는 생성형 AI입니다. 이름은 인공지능의 선구자 클로드 섀넌(Claude Shannon)에서 따왔으며, “헌宪법 기반 AI(constitutional AI)” 접근법을 적용해 안전성과 투명성을 강화한 것이 특징입니다. 즉, 인간의 직접적인 지시보다는 미리 정해둔 원칙과 가이드라인을 통해 스스로 출력을 조율하도록 설계되었기 때문에, 사용자가 안심하고 활용할 수 있는 대화형 AI라는 점에서 주목받고 있습니다.
from pydantic import BaseModel, Field
class code(BaseModel):
"""Schema for code solutions."""
# 코드에 대한 접근방식
prefix: str = Field(description="Description of the problem and approach")
# 코드의 import 영역
imports: str = Field(description="Code block import statements")
# 코드 자체에 대한
code: str = Field(description="Code block not including import statements")
from langchain_core.prompts import ChatPromptTemplate
GENERATE_CODE_TEMPLATE = """
Given the following pandas `describe()` output of a dataset,
write a **directly executable Python code** to:
1. handle missing values,
2. convert categorical columns,
3. ...any additional preprocessing needed,
4. prepare the dataset for machine learning.
Here is the describe result of the dataset:
\n ------- \n {context} \n ------- \n
Do not wrap the code in a function and the response in any backticks or anything else. The code should be written as a flat script, so that it can be run immediately and any errors will be visible during execution.
Ensure any code you provide can be executed \n
with all required imports and variables defined. Structure your answer with a description of the code solution. \n
Then list the imports. And finally list the functioning code block.
"""
code_gen_prompt = ChatPromptTemplate.from_messages(
[
("user", GENERATE_CODE_TEMPLATE),
]
)
from langchain_anthropic import ChatAnthropic
llm_claude= ChatAnthropic(model="claude-sonnet-4-20250514")
from langgraph.graph import StateGraph, MessagesState
class State(MessagesState): # messages
"""
Represents the state of our graph.
Attributes:
error : Binary flag for control flow to indicate whether test error was tripped
context: Data summary
generation : Code solution
iterations : Number of tries
"""
error: str # yes or no
context: str
generation: str
iterations: int
graph_builder = StateGraph(State)
llm_with_tools = llm_gpt.bind_tools(tools=[describe_data]) # llm이 알아서 하도록
def add_context(state: State):
print("##### ADD CONTEXT #####")
if messages := state.get("messages", []):
message = messages[-1] # 마지막 message 꺼내서 저장
else:
raise ValueError("No message found in input")
for tool_call in message.tool_calls:
for tool in tools:
if tool.name == tool_call['name']:
describe_str = tool.invoke(tool_call['args'])
# Get context from describe_data tool
print("데이터 통계 (context) : ", describe_str[:100])
return {"context": describe_str}
graph_builder.add_node("add_context", add_context)
from langgraph.graph import END
def guardrail_route(
state: State,
):
"""
Use in the conditional_edge to route to the ToolNode if the last message
has tool calls. Otherwise, route to the end.
"""
if isinstance(state, list):
ai_message = state[-1]
elif messages := state.get("messages", []):
ai_message = messages[-1]
else:
raise ValueError(f"No messages found in input state to tool_edge: {state}")
if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
return "add_context"
return END
graph_builder.add_conditional_edges(
"chatbot",
guardrail_route,
{"add_context": "add_context", END: END},
)
from pydantic import BaseModel, Field
class code(BaseModel):
"""Schema for code solutions."""
prefix: str = Field(description="Description of the problem and approach")
imports: str = Field(description="Code block import statements")
code: str = Field(description="Code block not including import statements")
from langchain_core.prompts import ChatPromptTemplate
GENERATE_CODE_TEMPLATE = """
Given the following pandas `describe()` output of a dataset,
write a **directly executable Python code** to:
1. handle missing values,
2. convert categorical columns,
3. ...any additional preprocessing needed,
4. prepare the dataset for machine learning.
Here is the describe result of the dataset:
\n ------- \n {context} \n ------- \n
Do not wrap the code in a function and the response in any backticks or anything else. The code should be written as a flat script, so that it can be run immediately and any errors will be visible during execution.
Ensure any code you provide can be executed \n
with all required imports and variables defined. Structure your answer with a description of the code solution. \n
Then list the imports. And finally list the functioning code block.
"""
code_gen_prompt = ChatPromptTemplate.from_messages(
[
("user", GENERATE_CODE_TEMPLATE),
]
)
from langchain_core.prompts import ChatPromptTemplate
reflect_prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"""
You are given an error message that occurred while running a Python script, along with the original code that produced the error.
Provide a corrected version of the original code that resolves the issue.
Ensure the code runs without errors and maintains the intended functionality."""
),
(
"user",
"""
--- ERROR MESSAGE ---
{error}
--- ORIGINAL CODE ---
{code_solution}
----------------------
Ensure any code you provide can be executed \n
with all required imports and variables defined. Structure your answer with a description of the code solution. \n
Then list the imports. And finally list the functioning code block.""",
)
]
)
Tool Calling Agent는 자신이 가진 지식만 사용하는 것이 아니라, 외부 도구(API, 데이터베이스, 코드 실행기 등)를 호출해 문제를 해결하는 에이전트입니다. 사용자의 질문을 이해한 뒤 필요한 경우 적절한 툴을 선택하고, 입력값을 구성해 호출하며, 반환된 결과를 다시 가공해 최종 답변을 만듭니다. 쉽게 말해, 단순히 대화만 하는 AI가 아니라 “필요할 때 계산기, 검색엔진, 데이터 조회 도구 같은 도구를 직접 쓸 수 있는 AI”가 Tool Calling Agent입니다.
2. 웹 검색을 하는 챗봇
1. Tavily
Tavily는 웹을 실시간으로 검색해 AI가 최신·정확한 정보를 답변할 수 있도록 돕는 AI용 검색·브라우징 API 플랫폼입니다.
import getpass
import os
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("OPENAI_API_KEY")
_set_env("TAVILY_API_KEY")
from tavily import TavilyClient
tavily_client = TavilyClient()
response = tavily_client.search("What is AI Agent?", max_results=3) # , topic="news", days = 10
response
max_results=3 → 검색 결과의 최대 개수를 지정합니다. 3이므로, 검색 결과를 최대 3개까지 가져옵니다.
topic="news" → 뉴스 기사 위주로 검색
days=10 → 최근 10일 이내의 자료만 검색
response['results']
# get_search_context: 보통 문자열(string) 형태이며, 여러 개의 검색 결과에서 중요한 내용만 추려서 한 덩어리의 텍스트로 제공합니다.
context = tavily_client.get_search_context(query="What is AI Agent?")
context
# qna_search: Tavily가 반환한 최종 답변. 보통 문자열(string) 형태이며, 한두 문장 정도로 정리된 응답을 제공합니다.
answer = tavily_client.qna_search(query="What is AI Agent?")
answer
topic(optional, str): 검색 카테고리 / "general"(Default), "news", "finance"
include_answer(optional, bool): 쿼리에 대한 답변 포함 여부
include_raw_content(optional, bool): 결과 HTML 포함 여부
include_images(optional, bool): 쿼리 관련 이미지 목록 포함 여부
include_image_descriptions(optional, bool): 각 이미지에 대한 설명 텍스트 포함 여부
search_depth(optional, str): 검색 깊이 / "basic"(Default),"advanced"
time_range(optional, str): 필터링 날짜 범위 - "day", "week", "month", "year"
include_domains(optional, List[str]): 구체적으로 포함할 도메인 목록
exclude_domains(optional, List[str]): 구체적으로 제외할 도메인 목록
!pip install langchain_tavily
from langchain_tavily import TavilySearch
tool = TavilySearch(max_results=3)
tool.invoke("What's a 'node' in LangGraph?")
invoke_with_toolcall = tool.invoke({"args": {'query': "What's a 'node' in LangGraph?"}, "type": "tool_call", "id": "foo", "name": "tavily_search"})
invoke_with_toolcall
.invoke()는 LangChain Tool 호출 형식으로 입력을 받을 수 있습니다.
단순 문자열을 넣는 대신, JSON(딕셔너리) 구조로 툴 호출 이벤트(tool call event)처럼 전달하는 방식입니다.
invoke_with_toolcall.content
!pip install langchain_community
from langchain_community.tools.tavily_search import TavilySearchResults
tool = TavilySearchResults(max_results=2)
tool.invoke("What's a 'node' in LangGraph?")
invoke_with_toolcall = tool.invoke({"args": {'query': "What's a 'node' in LangGraph?"}, "type": "tool_call", "id": "foo", "name": "tavily"})
invoke_with_toolcall
# results에 들어 있는 정보
invoke_with_toolcall.content
# 모델의 모든 실행결과
invoke_with_toolcall.artifact
3. 도구 바인딩
from langchain_core.tools import tool
# 데코레이터
@tool
def add(a: int, b: int) -> int:
# 툴에 대한 설명이 없으면 에러난다.
"""Adds a and b
Arg:
a: first int
b: second int
"""
return a + b
@tool
def multiply(a: int, b: int) -> int:
"""Multiply a and b
Arg:
a: first int
b: second int
"""
return a * b
tools = [add, multiply]
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
def chatbot(state: State):
return {"messages": [llm_with_tools.invoke(state["messages"])]} # 일반적인 질문에 대한 일반 답변 or tool_calls
graph_builder.add_node("chatbot", chatbot)
4. ToolNode
ToolNode는 LangGraph에서도구 호출을 실제로 실행해 주는 노드입니다. LLM이 생성한 AIMessage.tool_calls를 읽어 각 호출의 도구 이름과 인자를 매칭해 실행하고, 결과를 ToolMessage로 반환하여 그래프의 상태(대화 기록)에 추가합니다. 보통 “LLM 노드 → ToolNode → LLM 노드” 형태의 루프에서 사용되며, tools_condition 같은 조건부 엣지와 함께 붙여 LLM이 도구를 요청할 때만 ToolNode가 동작하게 합니다. 요약하면, ToolNode는LLM의 툴 호출 계획을 실제 코드 실행으로 연결하는 브리지로, 도구 레지스트리(이름→함수/툴)만 넘겨주면 호출·에러 처리·결과 전달까지 표준화된 방식으로 처리해줍니다.
import json
from langchain_core.messages import ToolMessage
from langgraph.prebuilt import ToolNode
class BasicToolNode:
def __init__(self, tools: list) -> None:
self.tools_by_name = {tool.name: tool for tool in tools} # ["tavily_search" : TavilySearch()]
def __call__(self, inputs: dict):
if messages := inputs.get("messages", []):
message = messages[-1] # 마지막 message
else:
raise ValueError("No message found in input")
outputs = []
for tool_call in message.tool_calls: # 메시지에서 호출된 도구를 불러옴
tool_result = self.tools_by_name[tool_call["name"]].invoke( # Tool 호출 실행
tool_call["args"]
)
outputs.append( # Tool 호출 결과(ToolMessage) 추가
ToolMessage(
content=json.dumps(tool_result),
name=tool_call["name"],
tool_call_id=tool_call["id"],
)
)
return {"messages": outputs}
tool_node = BasicToolNode(tools=[tool])
# tool_node = ToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)
def route_tools(
state: State,
):
if isinstance(state, list):
ai_message = state[-1]
elif messages := state.get("messages", []):
ai_message = messages[-1]
else:
raise ValueError(f"No messages found in input state to tool_edge: {state}")
if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
return "tools"
return END
graph_builder.add_conditional_edges(
"chatbot",
route_tools,
{"tools": "tools", END: END},
)
# 엣지 연결
graph_builder.add_edge("tools", "chatbot") # 도구가 호출될 때마다 챗봇으로 돌아가 다음 단계를 결정
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()
graph
def stream_graph_updates(user_input: str):
for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}): # graph 노드 호출 결과 받아옴
for value in event.values():
print("Assistant:", value["messages"][-1].content) # AI 답변 출력
while True:
try:
user_input = input("User: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Goodbye!")
break
stream_graph_updates(user_input)
except:
user_input = "What do you know about LangGraph?"
print("User: " + user_input)
stream_graph_updates(user_input)
break
5. create_react_agent
create_react_agent는 LangChain에서 ReAct 패턴(Reason+Act)을 따르는 에이전트를 손쉽게 구성하는 팩토리로, LLM과 사용할 도구 목록, 그리고 적절한 프롬프트를 결합해 “생각→도구 호출→관찰→최종 답변”의 반복 루프를 수행하는 Agent 객체를 만들어줍니다. 이 에이전트는 질문을 해석해 필요한 도구를 선택하고 인자를 구성해 호출한 뒤, 결과를 반영해 다음 행동을 결정하며, 보통 AgentExecutor와 함께 실행하여 다단계 추론과 복수의 툴 호출을 자동으로 오케스트레이션합니다. 핵심은 프롬프트(지침), LLM, 툴 레지스트리(이름→함수/API), 출력 파서를 표준화해 붙여주는 것이며, OpenAI 스타일의 툴콜을 포함한 다양한 LLM과 호환되어 실용적인 “생각하며 도구를 쓰는” 에이전트를 빠르게 구성할 수 있게 해주는 점입니다.
response = agent.invoke({"messages": [{"role": "user", "content": "What is LangGraph?"}]})
3. 원하는 형태로 출력하는 챗봇
from pydantic import BaseModel, Field
from typing import Union
class MovieResponse(BaseModel):
title: str = Field(description = '영화제목')
director : str = Field(description = '감독이름')
genre : str = Field(description = '장르')
release_year : int = Field(description = '개봉연도')
model = ChatOpenAI(model="gpt-5-nano")
model_with_structured_output = model.with_structured_output(MovieResponse)
model_with_structured_output.invoke('메멘토 영화에 대해 설명해주세요')
# MovieResponse(title='메멘토', director='크리스토퍼 놀런', genre='심리 스릴러, 미스터리', release_year=2000)
class MovieResponse(BaseModel):
title: str = Field(description = '영화제목')
director : str = Field(description = '감독이름')
genre : str = Field(description = '장르')
release_year : int = Field(description = '개봉연도')
class ConversationalResponse(BaseModel):
response: str = Field(description = '사용자의 질문에 대해 친절하게 대화하듯 답변하는 문장')
class FinalResponse(BaseModel):
final_output:Union[MovieResponse, ConversationalResponse]
structured_llm = model.with_structured_output(FinalResponse)
structured_llm.invoke('메멘토 영화에 대해 설명해주세요')
from langchain_core.tools import tool # 재정의
from langgraph.graph import MessagesState
from typing import Literal
class State(MessagesState):
final_response:MovieResponse
# 내 메세지에 따라 분기하는 코드
@tool
def get_movieinfo(movie:Literal['메멘토','인터스텔라']):
""" 아래 설명은 영화에 대한 내용이야. 참고해줘 """
if movie == "메멘토":
return "메멘토는 단기 기억 상실증을 가진 주인공이 아내 살해 사건의 진실을 찾기 위해 메모와 문신에 의존해 사건을 추적하는 독특한 구조의 스릴러 영화입니다. "
elif movie == "인터스텔라":
return "인터스텔라는 인류의 미래를 구하기 위해 우주로 떠난 탐사대가 사랑과 시간, 과학의 한계를 넘어서며 펼치는 감동적인 SF 영화입니다."
else:
raise AssertionError("알수없는 영화")
tools = [get_movieinfo] # tool로 등록
model_with_tool = model.bind_tools(tools)
from langchain_core.messages import AnyMessage
from typing_extensions import TypedDict
class State(TypedDict):
messages: list[AnyMessage]
extra_field: int
from langchain_core.messages import HumanMessage
result = graph.invoke({"messages": [HumanMessage("안녕")]})
# 워크플로우가 끝날때까지 기다려
result
'''
{'messages': AIMessage(content='안녕하세요! 무엇을 도와드릴까요?', additional_kwargs={}, response_metadata={}),
'extra_field': 10}
'''
result["messages"]
2. 대화메시지 상태 누적 업데이트
add_messages는 기존 메시지에서 추가 메시지를 병합하는 데 사용하는 함수로, 새로 들어온 메시지를 추가할 때 사용할 수 있는 리듀서 역할
from typing_extensions import Annotated
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
extra_field: int
input_message = {"role": "user", "content": "안녕하세요."}
result = graph.invoke({"messages": [input_message]})
for message in result["messages"]:
# pretty_print()는 데이터나 객체를 보기 좋게(Pretty) 정리해서 출력하는 함수
message.pretty_print()
'''
================================ Human Message =================================
안녕하세요.
================================== Ai Message ==================================
안녕하세요! 무엇을 도와드릴까요?
'''
# stream : 중간 결과를 실시간으로 반환함
# stream_mode="values" 각 단계의 현재 상태 값 출력
for chunk in graph.stream({"messages": [input_message]}, stream_mode="values"):
print(chunk)
for state_key, state_value in chunk.items():
if state_key == "messages":
state_value[-1].pretty_print() # 최근걸로 가져오기 위해서
# Default) stream_mode="updates" 각 단계의 상태 업데이트만 출력
for chunk in graph.stream({"messages": [input_message]}, stream_mode="updates"):
print(chunk)
for node, value in chunk.items():
if node:
print(node)
if "messages" in value:
print(value['messages'].content)
# stream_mode="messages" 각 단계의 메시지 출력
for chunk_msg, metadata in graph.stream({"messages": [input_message]}, stream_mode="messages"):
print(chunk_msg)
print(chunk_msg.content)
print(metadata)
print(metadata["langgraph_node"])
astream: 비동기 방식으로 스트리밍 처리
async for chunk_msg, metadata in graph.astream({"messages": [input_message]}, stream_mode="messages"):
print(chunk_msg)
print(chunk_msg.content)
print(metadata)
print(metadata["langgraph_node"])
3. 노드와 엣지 연결
from typing_extensions import TypedDict
class State(TypedDict):
value_1: str
value_2: int
import operator
from typing import Annotated, Any
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
aggregate: Annotated[list, operator.add] # 업데이트 값이 뒤에 추가되도록 하는 operator.add 리듀서
def a(state: State):
print(f'Adding "A" to {state["aggregate"]}')
return {"aggregate": ["A"]}
def b(state: State):
print(f'Adding "B" to {state["aggregate"]}')
return {"aggregate": ["B"]}
def c(state: State):
print(f'Adding "C" to {state["aggregate"]}')
return {"aggregate": ["C"]}
def d(state: State):
print(f'Adding "D" to {state["aggregate"]}')
return {"aggregate": ["D"]}
graph_builder = StateGraph(State)
# 노드 추가
graph_builder.add_node(a)
graph_builder.add_node(b)
graph_builder.add_node(c)
graph_builder.add_node(d)
# 엣지 추가
# graph_builder.add_edge(START, "a")
# graph_builder.add_edge("a", "b") # a -> b
# graph_builder.add_edge("b", "c") # a -> c
# graph_builder.add_edge("c", "d") # b -> d
# graph_builder.add_edge("d", END)
# graph = graph_builder.compile()
# graph
# 엣지 추가
graph_builder.add_edge(START, "a")
graph_builder.add_edge("a", "b") # a -> b
graph_builder.add_edge("a", "c") # a -> c
graph_builder.add_edge("b", "d") # b -> d
graph_builder.add_edge("c", "d") # c -> d
graph_builder.add_edge("d", END)
graph = graph_builder.compile()
graph
graph.invoke({"aggregate":[]})
'''
Adding "A" to []
Adding "B" to ['A']
Adding "C" to ['A']
Adding "D" to ['A', 'B', 'C']
{'aggregate': ['A', 'B', 'C', 'D']}
'''
6. 조건부 엣지 병렬 연결
import operator
from typing import Annotated, Sequence
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
aggregate: Annotated[list, operator.add]
which: str
def a(state: State):
print(f'Adding "A" to {state["aggregate"]}')
return {"aggregate": ["A"]}
def b(state: State):
print(f'Adding "B" to {state["aggregate"]}')
return {"aggregate": ["B"]}
def c(state: State):
print(f'Adding "C" to {state["aggregate"]}')
return {"aggregate": ["C"]}
def d(state: State):
print(f'Adding "D" to {state["aggregate"]}')
return {"aggregate": ["D"]}
def e(state: State):
print(f'Adding "E" to {state["aggregate"]}')
return {"aggregate": ["E"]}
# bc 혹은 cd 로 라우트를 결정하는 함수
def route_bc_or_cd(state: State) -> Sequence[str]:
if state["which"] == "cd":
return ["c", "d"]
return ["b", "c"] # 노드 이름
intermediates = ["b", "c", "d"]
graph_builder.add_conditional_edges(
# 조건을 주는 add_conditional
"a",
route_bc_or_cd, # 콜백
intermediates, # 적용
)
for node in intermediates:
graph_builder.add_edge(node, "e")
graph_builder.add_edge("e", END)
graph = graph_builder.compile()
graph
graph.invoke({"aggregate": [], "which": "bc"})
'''
Adding "A" to []
Adding "B" to ['A']
Adding "C" to ['A']
Adding "E" to ['A', 'B', 'C']
{'aggregate': ['A', 'B', 'C', 'E'], 'which': 'bc'}
'''
graph.invoke({"aggregate": [], "which": "cd"})
'''
Adding "A" to []
Adding "C" to ['A']
Adding "D" to ['A']
Adding "E" to ['A', 'C', 'D']
{'aggregate': ['A', 'C', 'D', 'E'], 'which': 'cd'}
'''
7. 조건과 반복
import operator
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
aggregate: Annotated[list, operator.add]
def a(state: State):
print(f'Node A 처리 중 현재 상태값 : {state["aggregate"]}')
return {"aggregate": ["A"]}
def b(state: State):
print(f'Node B 처리 중 현재 상태값 : {state["aggregate"]}')
return {"aggregate": ["B"]}
graph_builder = StateGraph(State)
graph_builder.add_node(a)
graph_builder.add_node(b)
import requests, zlib, base64
from IPython.display import Image
# Mermaid 코드 추출
code = graph.get_graph().draw_mermaid()
# 압축·인코딩 후 Kroki 요청
encoded = base64.urlsafe_b64encode(zlib.compress(code.encode())).decode()
url = f"https://kroki.io/mermaid/png/{encoded}"
resp = requests.get(url)
display(Image(resp.content))
graph.invoke({"aggregate": []})
'''
Node A 처리 중 현재 상태값 : []
Node B 처리 중 현재 상태값 : ['A']
Node A 처리 중 현재 상태값 : ['A', 'B']
Node B 처리 중 현재 상태값 : ['A', 'B', 'A']
Node A 처리 중 현재 상태값 : ['A', 'B', 'A', 'B']
Node B 처리 중 현재 상태값 : ['A', 'B', 'A', 'B', 'A']
Node A 처리 중 현재 상태값 : ['A', 'B', 'A', 'B', 'A', 'B']
{'aggregate': ['A', 'B', 'A', 'B', 'A', 'B', 'A']}
'''
from langgraph.errors import GraphRecursionError
# GraphRecursionError 로 에러를 반환하는 방법
try:
graph.invoke({"aggregate": []}, config={"recursion_limit": 4})
except GraphRecursionError: # 반복 종료 조건에 도달할 수 없는 경우
print("Recursion Error")
'''
Node A 처리 중 현재 상태값 : []
Node B 처리 중 현재 상태값 : ['A']
Node A 처리 중 현재 상태값 : ['A', 'B']
Node B 처리 중 현재 상태값 : ['A', 'B', 'A']
'''
8. 조건에 따른 반복 처리하기
import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
aggregate: Annotated[list, operator.add]
def a(state: State):
print(f'Node A 처리 중 현재 상태값 : {state["aggregate"]}')
return {"aggregate": ["A"]}
def b(state: State):
print(f'Node B 처리 중 현재 상태값 : {state["aggregate"]}')
return {"aggregate": ["B"]}
def c(state: State):
print(f'Node C 처리 중 현재 상태값 : {state["aggregate"]}')
return {"aggregate": ["C"]}
def d(state: State):
print(f'Node D 처리 중 현재 상태값 : {state["aggregate"]}')
return {"aggregate": ["D"]}
graph_builder = StateGraph(State)
graph_builder.add_node(a)
graph_builder.add_node(b)
graph_builder.add_node(c)
graph_builder.add_node(d)
result = graph.invoke({"aggregate": []})
'''
Node A 처리 중 현재 상태값 : []
Node B 처리 중 현재 상태값 : ['A']
Node C 처리 중 현재 상태값 : ['A', 'B']
Node D 처리 중 현재 상태값 : ['A', 'B']
Node A 처리 중 현재 상태값 : ['A', 'B', 'C', 'D']
Node B 처리 중 현재 상태값 : ['A', 'B', 'C', 'D', 'A']
Node C 처리 중 현재 상태값 : ['A', 'B', 'C', 'D', 'A', 'B']
Node D 처리 중 현재 상태값 : ['A', 'B', 'C', 'D', 'A', 'B']
Node A 처리 중 현재 상태값 : ['A', 'B', 'C', 'D', 'A', 'B', 'C', 'D']
'''
9. 사용자 입력에 따른 반복 조건 설정
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.graph.message import add_messages
class State(TypedDict):
human_messages: Annotated[list[HumanMessage], add_messages]
ai_messages: Annotated[list[AIMessage], add_messages]
retry_num : int
def route(state: State):
if "반복" in state["human_messages"][-1].content:
return "retry"
else:
return END
graph_builder.add_edge(START, "chatbot")
graph_builder.add_conditional_edges("chatbot", route)
graph_builder.add_edge("retry", "chatbot")
graph = graph_builder.compile()
for chunk in graph.stream({"human_messages" : "반복", "retry_num": 0}, stream_mode="updates"):
print(chunk)
for node, value in chunk.items():
if node:
print(node)
if "messages" in value:
print(value['messages'].content)
랭그래프(LangGraph)는 LangChain 생태계에서 에이전트나 RAG 시스템을 단계별로 구성하고 실행할 수 있게 해주는 그래프 기반 오케스트레이션 프레임워크입니다. 기존 RAG가 직선형 파이프라인이었다면, 랭그래프는 노드(작업 단위)와 엣지(흐름)를 그래프 형태로 정의해 분기, 반복, 조건 처리, 에이전트 루프 같은 복잡한 흐름을 명확하게 표현하고 실행할 수 있습니다. 이를 통해 “검색 → 답변 생성 → 자기평가 → 재검색” 같은 Agentic RAG 워크플로우를 안정적으로 설계·관리할 수 있으며, 디버깅과 모니터링도 쉬워집니다.
그래프 구조
그래프 구조(Graph Structure)는 객체(노드, vertex)와 그 객체들을 잇는 관계(간선, edge)로 표현되는 데이터 구조입니다. 노드는 개체를, 간선은 개체 간의 연결이나 관계를 나타내며, 방향성이 있는 경우 방향 그래프, 없는 경우 무방향 그래프로 구분됩니다. 이 구조는 복잡한 관계망을 직관적으로 표현하고 탐색할 수 있어, 소셜 네트워크 분석, 추천 시스템, 경로 탐색, 지식 그래프 등 다양한 분야에서 활용됩니다.
2. 그래프
자료 구조 알고리즘은 왜 공부해야하는 걸까?
프로그램 개발 능력이 향상되기 때문에 확연히 실력차이가 난다. 생각하는 폭이 넓어지는게 가장 큰 장점이다.
1. 그래프 종류
무방향 그래프 : 방향이 없는 그래프, 간선을 통해 노드는 양방향으로 갈 수 있음
방향 그래프 : 간선에 방향이 있는 그래프, 보통 노드 A, B가 A->B로 가는 간선으로 연결되어 있는 경우 <A,B>로 표기
가중치 그래프 : 간선에 비용 또는 가중치가 할당된 그래프
연결 그래프 : 무방향 그래프에서 모든 노드에 대해 항상 경로가 존재하는 경우
비연결 그래프 : 무방향 그래프에서 특정 노드에 대해 경로가 존재하지 않는 경우
순환 그래프 : 단순 경로의 시작 노드와 종료 노드가 동일한 경우
비순환 그래프 : 사이클이 없는 그래
2. 그래프와 트리의 차이
3. 너비 우선 탐색(BFS , Breadth-First Search)
대표적인 그래프 탐색 알고리즘으로 정점들과 같은 레벨에 있는 노드들(형제 노드들)을 먼저 우선 탐색하는 방식이다.
이는 한 단계씩 내려가면서 해당 노드와 같은 레벨에 있는 노드들을 먼저 순회한다.
https://olrlobt.tistory.com/41
앞으로 방문해야하는 곳, 방문했던 곳을 메모리에 방을 만든다.
방문해야하는 곳
1 2 ➡️ 3, 4 ...
방문했던 곳
1 ➡️ 3, 4
2 ➡️ 5, 6
3 ➡️ 7
어차피 중복된거 삭제할거니까 !
def bfs(graph, start_node):
visited,need_visit = list(),list()
need_visit.append(start_node)
while need_visit:
node = need_visit.pop(0)
if node not in visited:
visited.append(node)
need_visit.extend(graph[node])
return visited
노드는 LangGraph 워크플로우 안에서 실행되는 개별 작업 단위입니다. 예를 들어 “질문을 임베딩하기”, “벡터DB에서 관련 문서 검색하기”, “LLM으로 답변 생성하기” 같은 단계가 각각 하나의 노드가 됩니다. 즉, 노드는 전체 워크플로우를 이루는 핵심 블록이라고 할 수 있습니다.
2. 엣지(Edge)
엣지는 노드와 노드를 연결하는 흐름을 의미합니다. 한 노드의 결과가 다음 노드의 입력으로 이어지도록 만들어 주며, 워크플로우가 순차적으로 진행될 수 있게 합니다. 예를 들어 “검색 결과 → 답변 생성” 같은 연결이 엣지입니다.
3. 상태(State)
상태는 워크플로우 실행 중 유지되고 공유되는 데이터 저장소와 같습니다. 사용자 질문, 검색 결과, 현재까지의 답변 초안 등이 상태에 담겨 각 노드가 읽고 쓸 수 있습니다. 이 덕분에 그래프 전체가 일관된 맥락을 유지할 수 있습니다.
4. 조건 분기와 루프(Flow Control)
조건 분기는 특정 상황에 따라 워크플로우의 진행 경로를 바꾸는 기능입니다. 예를 들어 “검색 결과가 충분하면 답변 생성, 부족하면 재검색” 같은 분기 처리가 가능합니다. 루프는 특정 단계를 반복 실행하도록 해주며, 답변이 불충분할 경우 재검색-재생성을 반복하는 Agentic RAG 같은 구조를 구현할 수 있게 합니다.
👉 이렇게 네 가지 구성요소(노드, 엣지, 상태, 조건 분기·루프)가 맞물려 LangGraph는 단순한 RAG부터 복잡한 Agentic RAG까지 유연하게 표현할 수 있는 강력한 워크플로우 프레임워크가 됩니다.
4. 상태
State는 에이전트가 현재 어떤 정보를 가지고 있는지를 표현하는 데이터 구조로, 파이썬에서는 TypedDict나 Pydantic BaseModel을 사용해 정의할 수 있습니다. TypedDict는 단순히 키와 값의 타입을 명시하는 수준이라 잘못된 타입이 들어가도 실행은 되지만, 타입 검사기(mypy 등)에서만 잡아줍니다. 반면 Pydantic BaseModel은 실제 실행 시점에 타입을 엄격하게 검사하여 잘못된 데이터가 들어오면 오류를 발생시킵니다. 따라서 State를 정의할 때는 상황에 따라 단순히 구조만 명시할 것인지, 혹은 런타임에서 데이터 유효성까지 보장할 것인지를 고려해 TypedDict와 Pydantic을 선택하여 활용합니다.
from typing import TypedDict
class User(TypedDict): # 타입을 고정하는건 아님!
id: int
name: str
email: str
user1: User = {
'id': 1,
'name': '김사과',
'email': 'apple@apple.com'
}
print(user1)
# {'id': 1, 'name': '김사과', 'email': 'apple@apple.com'}
from pydantic import BaseModel
class User(BaseModel):
id: int
name: str
email: str
user_data = {
'id': 1,
'name': '김사과',
'email': 'apple@apple.com'
}
user1 = User(**user_data) # 딕셔너리에 접근해서 각각의 키와 값을 변수로 처리
print(user1)
# id=1 name='김사과' email='apple@apple.com'
# 타입이 안맞기때문에 에러가 분명히 난다.
user_data = {
'id': 1,
'name': 12345678,
'email': 'apple@apple.com'
}
user = User(**user_data)
print(user)
# ... For further information visit https://errors.pydantic.dev/2.11/v/string_type
State는 입력과 출력의 스키마를 정의하고 이를 업데이트하는리듀서 함수와 함께 사용됩니다. LangGraph에서는 TypedDict 등을 이용해 입력과 출력 상태를 명확히 정의하고, 각 노드가 상태를 받아 새로운 값을 반환하면 그래프가 이를 이어받아 전체 흐름을 완성합니다. 즉, State는 질문과 답변 같은 에이전트의 맥락 정보를 구조적으로 관리하며, 노드 실행 결과를 반영해 에이전트의 상태를 단계적으로 업데이트하는 핵심 역할을 합니다.
!pip install langgraph
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
# 입력을 위한 스키마 정의
class InputState(TypedDict):
question: str
# 출력을 위한 스키마 정의
class OutputState(TypedDict):
answer: str
# 입력과 출력을 합한 종합 스키마 정의
class OverallState(InputState, OutputState):
pass
# 입력을 처리하고 답변을 생성하는 노드 정의
def answer_node(state: InputState):
return {"answer": "bye", "question": state["question"]} # 상태 업데이트
graph_builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)
graph_builder.add_node(answer_node) # 답변 노드 추가
graph_builder.add_edge(START, "answer_node") # 시작 엣지 추가
graph_builder.add_edge("answer_node", END) # 끝 엣지 추가
graph = graph_builder.compile() # Compile the graph
# 입력 invoke 및 결과 출력
print(graph.invoke({"question": "hi"}))
# {'answer': 'bye'}
graph
State는 에이전트가 사용하는 데이터 구조로, 단순히 값만 담는 것이 아니라 값이 어떻게 업데이트될지(리듀서 함수)까지 함께 정의할 수 있습니다. 기본 TypedDict를 사용하면 단순히 키와 값의 타입만 지정하지만, Annotated와 리듀서 함수를 함께 쓰면 상태 업데이트 시 단순 덮어쓰기 대신 지정된 동작(예: add로 리스트 이어 붙이기, add_messages로 대화 메시지 누적하기)을 수행합니다. 즉, State는 에이전트의 현재 정보를 표현할 뿐 아니라, 노드 실행 결과가 기존 상태와 어떻게 병합·갱신될지를 규칙으로 내장한 확장된 상태 관리 구조입니다.
from typing_extensions import TypedDict
class State(TypedDict):
value1: int
value2: list[str]
from typing import Annotated
from typing_extensions import TypedDict
from operator import add
class State(TypedDict):
value1: int
value2: Annotated[list[str], add]
from typing_extensions import Annotated
from typing_extensions import TypedDict
def add(left, right):
# 만약 State안에 함수를 콜백으로 등록해놓으면 문자열로 들어왔을 때, 계속 연산해줄 수 있다.
return left + right
class State(TypedDict):
value1: int
value2: Annotated[list[str], add]
from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDict
class State(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
5. 노드
노드(Node)는 에이전트가 실제로 수행해야 할 논리나 작업을 구현한 함수이며, 그래프 안에서 상태(State)를 입력받아 새로운 값이나 업데이트를 반환하는 단위입니다. LangGraph에서는 add_node("노드이름", 함수) 형태로 노드를 등록하며, 각 노드는 입력 상태를 받아 특정 로직을 실행한 뒤 상태에 반영할 딕셔너리를 반환합니다. 이렇게 정의된 노드들은 그래프의 흐름 속에서 순차적으로 연결되어 실행되며, 에이전트가 질문에 답하거나 결과를 가공하는 등의 역할을 수행합니다.
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from operator import add
class State(TypedDict):
messages: Annotated[list[str], add]
graph = StateGraph(State)
엣지(Edges)는 그래프에서 노드 간의 실행 흐름을 연결하는 경로로, 에이전트가 어떤 순서로 작업을 이어갈지를 정의합니다. 가장 단순한 형태인 기본 엣지(Normal Edge)는 특정 노드가 끝나면 곧바로 다음 노드로 이동하도록 설정하며, graph.add_edge("node_a", "node_b")와 같이 사용합니다. 반면 조건부 엣지(Conditional Edge)는 라우팅 함수를 통해 현재 상태나 결과를 검사한 뒤, 조건에 따라 다른 노드로 분기시켜 실행 흐름을 유연하게 제어합니다. 즉, 엣지는 그래프에서 “다음에 무엇을 할지”를 결정짓는 연결선 역할을 하며, 조건을 두어 에이전트의 실행 경로를 동적으로 설계할 수 있게 합니다.
class State(TypedDict):
input: str
output: str
router_builder = StateGraph(State)
