Author(s): Michalzarnecki

Originally published on Towards AI.

Hi. So far in this series we’ve built a basic graph, and then a graph with an LLM and a conditional loop. We also covered different types of AI agents.
Now we’ll do a practical example of an agent that has access to an external knowledge source — Retrieval-Augmented Generation (RAG).

What does this flow look like?

The workflow is simple and very realistic:

1) Retrieval node

We start with a Retrieval node that pulls context from a vector database — for example, document chunks that best match the user question.

2) Generation node

Next comes the Generation node. Here the LLM receives the question + retrieved context and generates an answer.

3) Evaluation node

Finally we have an Evaluation node where we check whether the answer actually addresses the question.

If the evaluation fails → the graph goes back to the generation stage, forming a retry loop.

Why use a graph instead of a standard chain?

In a classic chain it’s hard to add a clean “check → retry” mechanism.
A graph lets us describe this logic directly:

• if the answer is good → we finish,
• if not → we go back and generate again.

Now let’s build it step by step in the notebook.

Install libraries, import dependencies and load configuration

!pip install -q langgraph langchain langchain-openai langchain-community faiss-cpu python-dotenv
!apt install libgraphviz-dev
!pip install pygraphviz

from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

Preparing the document database

docs = [
 "LangChain is a framework for working with large language models.",
 "Langgraph is a framework that allows you to build workflows in the form of state graphs.",
 "Retrieval-Augmented Generation (RAG) combines Context retrieval and response generation.",
 "FAISS is a library for finding the closest vectors in embeddings."
]

# Division into chunks
splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=10)
splits = splitter.create_documents(docs)

# Index in vector database
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(splits, embedding=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

Define a state

class State(TypedDict):
 question: str
 context: str
 answer: str
 pass_eval: bool

Retrieval node

def retrieval(state: State) -> State:
 docs = retriever.invoke(state["question"])
 context = "\\n".join([d.page_content for d in docs])
 return {"context": context}

Generation node

prompt = ChatPromptTemplate.from_messages([
 ("system", "You are an assistant who locates information in documents. Respond only based on CONTEXT:\\n{context}"),
 ("user", "{question}")
])

def generation(state: State) -> State:
 response = (prompt | llm).invoke({
 "context": state["context"],
 "question": state["question"]
 })
 return {"answer": response.content}

Evaluation node

eval_prompt = ChatPromptTemplate.from_messages([
 ("system", "Evaluate answer."),
 ("user", "Question: {question}\\nAnswer: {answer}\\nDoes the answer contain a needed information? Answer only 'yes' or 'no'.")
])

def eval_node(state: State) -> State:
 response = (eval_prompt | llm).invoke({
 "question": state["question"],
 "answer": state["answer"]
 })
 result = response.content.strip().lower().startswith("yes")
 return {"pass_eval": result}

Finish node

def finish(state: State) -> State:
 if state["pass_eval"]:
 print(f"✅ Answer approved: {state['answer']}")
 else:
 print("❌ The answer does not contain a solution.")
 return state

Prepare the graph

graph = StateGraph(State)

graph.add_node("retrieval", retrieval)
graph.add_node("generation", generation)
graph.add_node("eval", eval_node)
graph.add_node("finish", finish)

graph.set_entry_point("retrieval")
graph.add_edge("retrieval", "generation")
graph.add_edge("generation", "eval")

def check_eval(state: State):
 return "finish" if state["pass_eval"] else "generation"

graph.add_conditional_edges("eval", check_eval, ["finish", "generation"])
graph.add_edge("finish", END)

app = graph.compile(debug=True)

Visualize the graph

# Graph visualization
from IPython.display import Image, display

png_bytes = app.get_graph().draw_png()
display(Image(png_bytes))

Run the graph

app.invoke({"question": "What is a LangGraph?"})

output:

[values] {'question': 'What is a LangGraph?'}
[updates] {'retrieval': {'context': 'Langgraph is a framework that allows you to build workflows in the form of state graphs.\\nLangChain is a framework for working with large language models.'}}
[values] {'question': 'What is a LangGraph?', 'context': 'Langgraph is a framework that allows you to build workflows in the form of state graphs.\\nLangChain is a framework for working with large language models.'}
[updates] {'generation': {'answer': 'LangGraph is a framework that allows you to build workflows in the form of state graphs.'}}
[values] {'question': 'What is a LangGraph?', 'context': 'Langgraph is a framework that allows you to build workflows in the form of state graphs.\\nLangChain is a framework for working with large language models.', 'answer': 'LangGraph is a framework that allows you to build workflows in the form of state graphs.'}
[updates] {'eval': {'pass_eval': True}}
[values] {'question': 'What is a LangGraph?', 'context': 'Langgraph is a framework that allows you to build workflows in the form of state graphs.\\nLangChain is a framework for working with large language models.', 'answer': 'LangGraph is a framework that allows you to build workflows in the form of state graphs.', 'pass_eval': True}
✅ Answer approved: LangGraph is a framework that allows you to build workflows in the form of state graphs.
[updates] {'finish': {'question': 'What is a LangGraph?', 'context': 'Langgraph is a framework that allows you to build workflows in the form of state graphs.\\nLangChain is a framework for working with large language models.', 'answer': 'LangGraph is a framework that allows you to build workflows in the form of state graphs.', 'pass_eval': True}}
[values] {'question': 'What is a LangGraph?', 'context': 'Langgraph is a framework that allows you to build workflows in the form of state graphs.\\nLangChain is a framework for working with large language models.', 'answer': 'LangGraph is a framework that allows you to build workflows in the form of state graphs.', 'pass_eval': True}
{'question': 'What is a LangGraph?',
 'context': 'Langgraph is a framework that allows you to build workflows in the form of state graphs.\\nLangChain is a framework for working with large language models.',
 'answer': 'LangGraph is a framework that allows you to build workflows in the form of state graphs.',
 'pass_eval': True}

That’s all int this chapter dedicated to AI RAG agent built in LangGraph. In the next chapter we will build fully functional application of AI agent article publisher.

see next chapter

see previous chapter

see the full code from this article in the GitHub repository

Join thousands of data leaders on the AI newsletter. Join over 80,000 subscribers and keep up to date with the latest developments in AI. From research to projects and ideas. If you are building an AI startup, an AI-related product, or a service, we invite you to consider becoming a sponsor.

Published via Towards AI