This content originally appeared on DEV Community and was authored by Ajmal Hasan
Part 1: Python Fundamentals and LangChain Basics
Introduction: Your Journey to GenAI Mastery Begins Here
Welcome to the world of Generative AI development! If you’re completely new to this field, you’re about to embark on an exciting journey that will transform you from a complete beginner to a proficient GenAI developer. This comprehensive three-part series is designed specifically for beginners who want to learn everything sequentially and build a solid foundation.
Generative AI is revolutionizing industries from healthcare to finance, and Python stands at the forefront of this revolution. Whether you’re a seasoned programmer from another language or writing “Hello, World!” for the first time, this guide will take you through everything you need to know.
Setting Up Your Python Environment for GenAI Development
Before diving into the exciting world of AI agents and language models, we need to establish a proper development environment. This foundation is crucial for your success.
Installing Python
First, ensure you have Python 3.8 or higher installed on your system. Python stands out as the language of choice for AI development due to its simplicity, extensive community support, and powerful libraries.
# Check your Python version
python --version
If you need to install Python, download it from the official Python website or consider using Anaconda or Miniconda, which include essential tools for data science and AI development.
Creating Virtual Environments
Virtual environments are essential for managing project dependencies and preventing conflicts between different projects. Think of them as isolated Python installations for each project.
# Create a virtual environment
python -m venv genai_env
# Activate on Windows
genai_env\Scripts\activate
# Activate on macOS/Linux
source genai_env/bin/activate
When activated, your command prompt will show the environment name, indicating you’re working within the isolated environment.
Installing Essential Libraries
With your virtual environment active, install the core libraries we’ll use throughout this series:
# Core LangChain installation
pip install langchain langchain-core langchain-community
# LangGraph for advanced agent workflows
pip install langgraph
# Essential ML and data libraries
pip install numpy pandas openai python-dotenv
# For RAG systems (we'll cover this in Part 2)
pip install chromadb faiss-cpu
LangChain requires Python 3.7 or later and comes with modular packages that let you install only what you need.
Understanding the GenAI Development Stack
Before jumping into code, let’s understand the key components of modern GenAI applications:
Large Language Models (LLMs): The brain of your AI applications, capable of understanding and generating human-like text.
LangChain: A framework that simplifies building applications with LLMs by providing modular components and chains.
LangGraph: An extension of LangChain for building stateful, multi-actor applications and complex agent workflows.
Vector Databases: Storage systems for embeddings that enable semantic search and retrieval.
Prompt Engineering: The art of crafting effective instructions for LLMs to get desired outputs.
Your First LangChain Application
Let’s build a simple but powerful application that demonstrates core LangChain concepts. We’ll create a text translator that showcases prompt templates and LLM integration.
Setting Up API Keys
First, you’ll need an API key from a language model provider. For this example, we’ll use OpenAI:
import os
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
# Set your API key (create a .env file with OPENAI_API_KEY=your-key-here)
os.environ["OPENAI_API_KEY"] = "your-api-key-here"
Creating Your First Chain
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
# Initialize the language model
llm = ChatOpenAI(temperature=0.7)
# Create a prompt template
prompt = ChatPromptTemplate.from_template(
"Translate the following English text to {language}: {text}"
)
# Create an output parser
output_parser = StrOutputParser()
# Build the chain using LCEL (LangChain Expression Language)
chain = prompt | llm | output_parser
# Use the chain
result = chain.invoke({
"language": "Spanish",
"text": "Hello, how are you today?"
})
print(result) # Output: "Hola, ¿cómo estás hoy?"
This example demonstrates several fundamental concepts:
Prompt Templates: Reusable structures for creating prompts with dynamic inputs. They help maintain consistency and make your prompts more maintainable.
LangChain Expression Language (LCEL): The declarative syntax using the pipe operator (|) to chain components together. This creates optimized, streamable, and debuggable chains.
Output Parsers: Components that structure the LLM’s response into the format you need.
Understanding Prompt Templates
Prompt templates are the foundation of effective GenAI applications. They provide a consistent, reproducible way to formulate prompts and improve your chances of receiving high-quality outputs.
from langchain_core.prompts import PromptTemplate
# Simple string prompt template
simple_prompt = PromptTemplate.from_template(
"Write a {adjective} story about {topic} in {length} words."
)
# Chat prompt template with system and human messages
from langchain_core.prompts import ChatPromptTemplate
chat_prompt = ChatPromptTemplate([
("system", "You are a helpful AI assistant that provides clear, concise answers."),
("human", "Explain {concept} in simple terms for a beginner.")
])
# Using the templates
story_prompt = simple_prompt.invoke({
"adjective": "funny",
"topic": "a robot learning to cook",
"length": "100"
})
explanation_prompt = chat_prompt.invoke({
"concept": "machine learning"
})
Building More Complex Chains
As your applications grow in complexity, you’ll need to combine multiple operations. LangChain makes this straightforward:
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
# Create a more complex chain that analyzes and translates simultaneously
analysis_prompt = ChatPromptTemplate.from_template(
"Analyze the sentiment and main topics of this text: {text}"
)
translation_prompt = ChatPromptTemplate.from_template(
"Translate this text to {language}: {text}"
)
# Parallel processing chain
parallel_chain = RunnableParallel({
"analysis": analysis_prompt | llm | StrOutputParser(),
"translation": translation_prompt | llm | StrOutputParser(),
"original": RunnablePassthrough()
})
# Execute parallel operations
result = parallel_chain.invoke({
"text": "I love learning about artificial intelligence!",
"language": "French"
})
print("Original:", result["original"]["text"])
print("Analysis:", result["analysis"])
print("Translation:", result["translation"])
Error Handling and Best Practices
Production-ready applications need robust error handling:
from langchain_core.runnables import RunnableLambda
def safe_llm_call(inputs):
try:
return llm.invoke(inputs)
except Exception as e:
return f"Error occurred: {str(e)}"
# Create a chain with error handling
safe_chain = prompt | RunnableLambda(safe_llm_call) | StrOutputParser()
Understanding Memory and State
While basic chains are stateless, real applications often need to remember previous interactions:
from langchain.memory import ConversationBufferMemory
from langchain.schema import HumanMessage, AIMessage
# Simple conversation memory example
memory = ConversationBufferMemory()
# Add messages to memory
memory.chat_memory.add_user_message("What is machine learning?")
memory.chat_memory.add_ai_message("Machine learning is a method where computers learn patterns from data...")
# Retrieve conversation history
history = memory.buffer
print(history)
Key Takeaways from Part 1
You’ve successfully completed the foundational layer of GenAI development! Here’s what you’ve learned:
Environment Setup: Created an isolated Python environment with all necessary libraries for GenAI development.
LangChain Basics: Understanding how to create and use prompt templates, chains, and output parsers.
LCEL Syntax: Using the pipe operator to create declarative, optimized chains.
Error Handling: Building robust applications that gracefully handle failures.
Memory Concepts: Understanding how applications can maintain conversation state.
In Part 2, we’ll build upon this foundation to explore Retrieval-Augmented Generation (RAG), where you’ll learn to create AI systems that can access and reason over your own documents and data sources. This is where GenAI development becomes truly powerful for real-world applications.
Part 2: Retrieval-Augmented Generation (RAG) with LangChain
Introduction: Expanding AI Capabilities with External Knowledge
Welcome back! In Part 1, you mastered Python environment setup and LangChain basics. Now we’re entering the exciting world of Retrieval-Augmented Generation (RAG), where your AI applications can access and reason over external knowledge sources.
RAG is one of the most practical applications of GenAI in enterprise settings because it allows LLMs to work with private data that wasn’t part of their original training. By the end of this part, you’ll build a complete RAG system that can answer questions based on your own documents.
Understanding RAG Architecture
RAG combines the power of large language models with external knowledge retrieval. This dual mechanism allows models to blend their trained knowledge with new, relevant information, producing richer and more contextually accurate responses.
A typical RAG application has two main components:
Indexing: A pipeline for ingesting data from sources and indexing it (usually happens offline).
Retrieval and Generation: The actual RAG chain that takes user queries, retrieves relevant data, and passes that context to the model.
The complete sequence looks like this:
Indexing Pipeline
- Load: Use document loaders to ingest data from various sources
- Split: Break large documents into smaller, manageable chunks
- Store: Create embeddings and store them in a vector database
Retrieval and Generation Pipeline
- Retrieve: Find relevant document chunks based on user query
- Generate: Combine query and retrieved context for LLM response
Document Loading and Processing
Let’s start by learning how to load and process different types of documents. LangChain provides hundreds of document loaders for various data sources.
from langchain_community.document_loaders import (
TextLoader,
PyPDFLoader,
CSVLoader,
WebBaseLoader
)
# Loading different document types
def load_documents():
documents = []
# Load text files
text_loader = TextLoader("company_policies.txt")
documents.extend(text_loader.load())
# Load PDF documents
pdf_loader = PyPDFLoader("employee_handbook.pdf")
documents.extend(pdf_loader.load())
# Load CSV data
csv_loader = CSVLoader("employee_data.csv")
documents.extend(csv_loader.load())
# Load web pages
web_loader = WebBaseLoader([
"https://company.com/policies",
"https://company.com/benefits"
])
documents.extend(web_loader.load())
return documents
# Load all documents
raw_documents = load_documents()
print(f"Loaded {len(raw_documents)} documents")
Text Splitting and Chunking
Large documents need to be split into smaller chunks for effective retrieval and to fit within LLM context windows:
from langchain_text_splitters import (
RecursiveCharacterTextSplitter,
TokenTextSplitter
)
# Initialize text splitter
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # Maximum chunk size in characters
chunk_overlap=200, # Overlap between chunks for context continuity
length_function=len, # Function to measure chunk length
separators=["\n\n", "\n", " ", ""] # Priority order for splitting
)
# Split documents into chunks
chunks = text_splitter.split_documents(raw_documents)
print(f"Split into {len(chunks)} chunks")
print(f"Sample chunk: {chunks[0].page_content[:200]}...")
The overlap between chunks ensures that context isn’t lost when information spans chunk boundaries.
Embeddings and Vector Stores
Embeddings convert text into numerical representations that capture semantic meaning. Vector stores enable efficient similarity search over these embeddings.
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
import os
# Initialize embeddings model
embeddings = OpenAIEmbeddings()
# Create vector store from documents
def create_vector_store(chunks, embeddings_model):
# Chroma is a popular, lightweight vector database
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings_model,
persist_directory="./vector_db" # Persist database to disk
)
return vectorstore
# Create and populate vector store
vectorstore = create_vector_store(chunks, embeddings)
# Test similarity search
query = "What is our vacation policy?"
relevant_docs = vectorstore.similarity_search(query, k=3)
for i, doc in enumerate(relevant_docs):
print(f"Document {i+1}: {doc.page_content[:100]}...")
Building Your First RAG Chain
Now let’s combine retrieval with generation to create a complete RAG system:
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
# Create retriever from vector store
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 3} # Retrieve top 3 relevant chunks
)
# Define RAG prompt template
rag_prompt = ChatPromptTemplate.from_template("""
You are a helpful assistant that answers questions based on the provided context.
Context: {context}
Question: {question}
Provide a clear, accurate answer based only on the information in the context.
If the context doesn't contain enough information to answer the question,
say so clearly.
""")
# Function to format retrieved documents
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
# Initialize LLM
llm = ChatOpenAI(temperature=0)
# Build RAG chain using LCEL
rag_chain = (
{
"context": retriever | format_docs,
"question": RunnablePassthrough()
}
| rag_prompt
| llm
| StrOutputParser()
)
# Test the RAG system
question = "What is our remote work policy?"
answer = rag_chain.invoke(question)
print(f"Question: {question}")
print(f"Answer: {answer}")
Advanced RAG Techniques
Let’s enhance our RAG system with more sophisticated features:
Multi-Vector Retrieval
Sometimes you want to store multiple vectors per document for better retrieval:
from langchain.retrievers import MultiVectorRetriever
from langchain.storage import InMemoryByteStore
from langchain_core.documents import Document
# Create document store for original documents
docstore = InMemoryByteStore()
# Create multi-vector retriever
multi_vector_retriever = MultiVectorRetriever(
vectorstore=vectorstore,
byte_store=docstore,
id_key="doc_id"
)
# Add documents with summaries for better retrieval
def create_summaries(docs, llm):
summaries = []
for doc in docs:
summary_prompt = f"Summarize this document in 2-3 sentences:\n{doc.page_content}"
summary = llm.invoke(summary_prompt).content
summaries.append(summary)
return summaries
# Generate summaries (you could also create hypothetical questions)
summaries = create_summaries(chunks[:5], llm) # Example with first 5 chunks
Conversational RAG
Add conversation memory to maintain context across multiple questions:
from langchain.memory import ConversationBufferMemory
# Create conversational RAG prompt
conversational_prompt = ChatPromptTemplate.from_template("""
You are a helpful assistant that answers questions based on context and conversation history.
Context: {context}
Conversation History: {chat_history}
Current Question: {question}
Provide a clear, accurate answer considering both the context and previous conversation.
""")
# Initialize conversation memory
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
# Enhanced RAG chain with memory
conversational_rag_chain = (
{
"context": retriever | format_docs,
"question": RunnablePassthrough(),
"chat_history": lambda x: memory.buffer
}
| conversational_prompt
| llm
| StrOutputParser()
)
# Function to run conversational RAG
def chat_with_rag(question):
response = conversational_rag_chain.invoke(question)
# Add to memory
memory.chat_memory.add_user_message(question)
memory.chat_memory.add_ai_message(response)
return response
# Test conversational capabilities
print(chat_with_rag("What are our office hours?"))
print(chat_with_rag("Are there any exceptions to those hours?")) # Follow-up question
Working with Different Vector Stores
While Chroma is great for development, you might want other options for production:
# FAISS - Great for high-performance similarity search
from langchain_community.vectorstores import FAISS
faiss_store = FAISS.from_documents(chunks, embeddings)
# Qdrant - Production-ready vector database
from langchain_community.vectorstores import Qdrant
qdrant_store = Qdrant.from_documents(
chunks,
embeddings,
url="http://localhost:6333", # Qdrant server URL
collection_name="company_docs"
)
# Pinecone - Managed vector database service
from langchain_community.vectorstores import Pinecone
import pinecone
# Initialize Pinecone (requires API key)
# pinecone_store = Pinecone.from_documents(chunks, embeddings, index_name="company-docs")
RAG Evaluation and Optimization
Measuring RAG performance is crucial for production systems:
def evaluate_rag_response(question, generated_answer, reference_docs):
"""
Simple evaluation framework for RAG responses
"""
# Check if answer contains key information from reference docs
key_terms = set()
for doc in reference_docs:
# Extract important terms (this is simplified)
key_terms.update(doc.page_content.lower().split())
answer_terms = set(generated_answer.lower().split())
overlap = len(key_terms.intersection(answer_terms))
# Calculate relevance score
relevance_score = overlap / len(key_terms) if key_terms else 0
return {
"relevance_score": relevance_score,
"answer_length": len(generated_answer),
"sources_used": len(reference_docs)
}
# Example evaluation
test_question = "What is our vacation policy?"
test_docs = retriever.get_relevant_documents(test_question)
test_answer = rag_chain.invoke(test_question)
eval_result = evaluate_rag_response(test_question, test_answer, test_docs)
print(f"Evaluation: {eval_result}")
Building a Complete RAG Application
Let’s put everything together into a production-ready RAG application:
class CompanyRAGSystem:
def __init__(self, document_paths, openai_api_key):
os.environ["OPENAI_API_KEY"] = openai_api_key
# Initialize components
self.embeddings = OpenAIEmbeddings()
self.llm = ChatOpenAI(temperature=0)
self.memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
# Load and process documents
self.documents = self._load_documents(document_paths)
self.vectorstore = self._create_vector_store()
self.retriever = self.vectorstore.as_retriever(search_kwargs={"k": 3})
# Build RAG chain
self.rag_chain = self._build_rag_chain()
def _load_documents(self, paths):
documents = []
for path in paths:
if path.endswith('.pdf'):
loader = PyPDFLoader(path)
elif path.endswith('.txt'):
loader = TextLoader(path)
elif path.endswith('.csv'):
loader = CSVLoader(path)
documents.extend(loader.load())
# Split documents
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
return text_splitter.split_documents(documents)
def _create_vector_store(self):
return Chroma.from_documents(
self.documents,
self.embeddings,
persist_directory="./company_rag_db"
)
def _build_rag_chain(self):
prompt = ChatPromptTemplate.from_template("""
Answer questions about our company based on the provided context and conversation history.
Context: {context}
Chat History: {chat_history}
Question: {question}
Provide accurate, helpful answers based on the available information.
""")
return (
{
"context": self.retriever | self._format_docs,
"question": RunnablePassthrough(),
"chat_history": lambda x: self._get_chat_history()
}
| prompt
| self.llm
| StrOutputParser()
)
def _format_docs(self, docs):
return "\n\n".join(doc.page_content for doc in docs)
def _get_chat_history(self):
return self.memory.buffer
def ask(self, question):
"""Main interface for asking questions"""
response = self.rag_chain.invoke(question)
# Update memory
self.memory.chat_memory.add_user_message(question)
self.memory.chat_memory.add_ai_message(response)
return response
# Usage example
rag_system = CompanyRAGSystem(
document_paths=["company_handbook.pdf", "policies.txt"],
openai_api_key="your-api-key"
)
# Ask questions
response1 = rag_system.ask("What is our vacation policy?")
response2 = rag_system.ask("How many days can I take off?") # Follow-up
print(response1)
print(response2)
Key Takeaways from Part 2
You’ve now mastered Retrieval-Augmented Generation! Here’s what you accomplished:
Document Processing: Learned to load, split, and process various document types for RAG systems.
Vector Embeddings: Understanding how text becomes searchable through semantic embeddings.
RAG Architecture: Built complete systems that combine retrieval with generation.
Advanced Features: Implemented conversational memory, multi-vector retrieval, and evaluation frameworks.
Production Systems: Created a reusable RAG class suitable for real applications.
In Part 3, we’ll explore the most advanced topic: building intelligent agents using LangGraph. You’ll learn to create AI systems that can plan, use tools, and handle complex multi-step workflows autonomously.
Part 3: Advanced AI Agent Architecture with LangChain and LangGraph
Introduction: Building Intelligent, Autonomous AI Systems
Welcome to the final and most exciting part of your GenAI journey! You’ve mastered Python fundamentals, LangChain basics, and RAG systems. Now we’re entering the cutting-edge world of AI agents – autonomous systems that can reason, plan, and take actions to accomplish complex goals.
Unlike the linear chains we’ve built so far, agents can make decisions about what actions to take based on the situation. They can use tools, maintain memory, collaborate with other agents, and even engage in human-in-the-loop workflows. By the end of this part, you’ll build a complete multi-agent system that can handle complex, real-world tasks.
Understanding AI Agents vs. Traditional Chains
Traditional LangChain applications follow a predetermined sequence: input → processing → output. Agents, however, use an LLM as a reasoning engine to determine what actions to take and when to take them.
Key differences:
Chains: Fixed sequence of operations (like our RAG system)
Agents: Dynamic decision-making based on context and available tools
Agent Architecture Components:
- LLM Brain: Makes decisions about what actions to take
- Tools: External capabilities the agent can use (search, calculations, APIs)
- Memory: Maintains context across interactions
- Planning: Can break down complex tasks into steps
- Execution: Actually performs the chosen actions
Building Your First Agent
Let’s start with a simple agent that can search the web and perform calculations:
import os
from langchain.agents import initialize_agent, AgentType
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain_community.utilities import SerpAPIWrapper
import math
# Set up API keys
os.environ["OPENAI_API_KEY"] = "your-openai-key"
os.environ["SERPAPI_API_KEY"] = "your-serpapi-key" # For web search
# Initialize LLM
llm = ChatOpenAI(temperature=0, model="gpt-4")
# Create tools for the agent
search = SerpAPIWrapper()
def calculator(expression: str) -> str:
"""Perform mathematical calculations. Use Python syntax."""
try:
result = eval(expression)
return f"The result is: {result}"
except Exception as e:
return f"Error in calculation: {e}"
# Define agent tools
tools = [
Tool(
name="Search",
func=search.run,
description="Useful for searching current information on the internet"
),
Tool(
name="Calculator",
func=calculator,
description="Useful for mathematical calculations. Input should be a valid Python expression."
)
]
# Create the agent
agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.OPENAI_FUNCTIONS,
verbose=True, # Shows the agent's thinking process
handle_parsing_errors=True
)
# Test the agent
response = agent.run("What's the current stock price of Apple, and if I bought 100 shares, how much would that cost?")
print(response)
Transitioning to LangGraph: Modern Agent Architecture
While the traditional AgentExecutor works, LangGraph provides much more control and flexibility for building production agents. Let’s rebuild our agent using LangGraph:
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import HumanMessage, AIMessage
from typing import Annotated, TypedDict
import operator
# Define the agent's state
class AgentState(TypedDict):
messages: Annotated[list, operator.add] # Conversation history
current_task: str # Current task being processed
tool_results: dict # Results from tool calls
def create_research_agent():
"""Create a research agent using LangGraph"""
# Agent node - makes decisions
def agent_node(state: AgentState):
messages = state["messages"]
# Get the latest human message
last_message = messages[-1]
# Agent decides what to do based on the message
if "search" in last_message.content.lower() or "find" in last_message.content.lower():
# Needs to search for information
return {
"messages": [AIMessage(content="I'll search for that information.")],
"current_task": "search"
}
elif any(op in last_message.content for op in ["+", "-", "*", "/", "calculate"]):
# Needs to perform calculation
return {
"messages": [AIMessage(content="I'll calculate that for you.")],
"current_task": "calculate"
}
else:
# Can answer directly
response = llm.invoke([HumanMessage(content=last_message.content)])
return {
"messages": [response],
"current_task": "complete"
}
# Search tool node
def search_node(state: AgentState):
last_message = state["messages"][-1]
search_query = last_message.content
try:
search_result = search.run(search_query)
result_message = AIMessage(content=f"Search results: {search_result}")
return {
"messages": [result_message],
"tool_results": {"search": search_result},
"current_task": "complete"
}
except Exception as e:
error_message = AIMessage(content=f"Search failed: {e}")
return {
"messages": [error_message],
"current_task": "complete"
}
# Calculator tool node
def calculator_node(state: AgentState):
last_message = state["messages"][-1]
# Extract mathematical expression
expression = last_message.content
try:
result = calculator(expression)
result_message = AIMessage(content=result)
return {
"messages": [result_message],
"tool_results": {"calculation": result},
"current_task": "complete"
}
except Exception as e:
error_message = AIMessage(content=f"Calculation failed: {e}")
return {
"messages": [error_message],
"current_task": "complete"
}
# Routing function - decides which node to go to next
def route_decision(state: AgentState):
current_task = state.get("current_task", "")
if current_task == "search":
return "search_tool"
elif current_task == "calculate":
return "calculator_tool"
else:
return "end"
# Build the graph
graph = StateGraph(AgentState)
# Add nodes
graph.add_node("agent", agent_node)
graph.add_node("search_tool", search_node)
graph.add_node("calculator_tool", calculator_node)
# Define the flow
graph.add_edge(START, "agent")
# Add conditional routing from agent
graph.add_conditional_edges(
"agent",
route_decision,
{
"search_tool": "search_tool",
"calculator_tool": "calculator_tool",
"end": END
}
)
# Tools go back to agent or end
graph.add_edge("search_tool", END)
graph.add_edge("calculator_tool", END)
# Add memory
memory = MemorySaver()
# Compile the graph
return graph.compile(checkpointer=memory)
# Create and test the agent
research_agent = create_research_agent()
# Test with different types of queries
config = {"configurable": {"thread_id": "1"}}
# Search query
result1 = research_agent.invoke({
"messages": [HumanMessage(content="Search for the latest news about artificial intelligence")]
}, config)
print("Search Result:", result1["messages"][-1].content)
# Calculation query
result2 = research_agent.invoke({
"messages": [HumanMessage(content="Calculate 25 * 4 + 100")]
}, config)
print("Calculation Result:", result2["messages"][-1].content)
Building a Multi-Agent System
The real power of LangGraph comes with multi-agent systems where specialized agents collaborate:
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing import Literal
# Define shared state for multi-agent system
class MultiAgentState(TypedDict):
messages: Annotated[list, operator.add]
current_agent: str
task_completed: bool
research_data: dict
analysis_results: dict
# Research Agent - specializes in finding information
class ResearchAgent:
def __init__(self, llm, search_tool):
self.llm = llm
self.search = search_tool
self.name = "Researcher"
def research(self, state: MultiAgentState) -> dict:
last_message = state["messages"][-1].content
# Perform research
research_prompt = f"""
You are a research specialist. Research this topic thoroughly: {last_message}
Provide comprehensive, factual information.
"""
try:
# Search for information
search_results = self.search.run(last_message)
# Analyze and summarize
analysis_prompt = f"""
Based on this search data: {search_results}
Provide a well-structured research summary including:
1. Key findings
2. Important statistics
3. Recent developments
"""
summary = self.llm.invoke([HumanMessage(content=analysis_prompt)])
return {
"messages": [AIMessage(content=f"Research completed by {self.name}: {summary.content}")],
"research_data": {"summary": summary.content, "raw_data": search_results},
"current_agent": "analyst"
}
except Exception as e:
return {
"messages": [AIMessage(content=f"Research failed: {e}")],
"current_agent": "end"
}
# Analysis Agent - specializes in analyzing data
class AnalysisAgent:
def __init__(self, llm):
self.llm = llm
self.name = "Analyst"
def analyze(self, state: MultiAgentState) -> dict:
research_data = state.get("research_data", {})
analysis_prompt = f"""
You are a data analyst. Analyze this research data: {research_data.get('summary', '')}
Provide:
1. Key insights
2. Trends and patterns
3. Implications
4. Recommendations
"""
analysis = self.llm.invoke([HumanMessage(content=analysis_prompt)])
return {
"messages": [AIMessage(content=f"Analysis completed by {self.name}: {analysis.content}")],
"analysis_results": {"insights": analysis.content},
"current_agent": "writer",
"task_completed": False
}
# Writing Agent - specializes in creating final reports
class WritingAgent:
def __init__(self, llm):
self.llm = llm
self.name = "Writer"
def write_report(self, state: MultiAgentState) -> dict:
research_data = state.get("research_data", {})
analysis_results = state.get("analysis_results", {})
writing_prompt = f"""
You are a professional writer. Create a comprehensive report based on:
Research: {research_data.get('summary', '')}
Analysis: {analysis_results.get('insights', '')}
Write a well-structured, professional report with:
1. Executive Summary
2. Main Findings
3. Analysis
4. Conclusions
"""
report = self.llm.invoke([HumanMessage(content=writing_prompt)])
return {
"messages": [AIMessage(content=f"Final report by {self.name}: {report.content}")],
"current_agent": "supervisor",
"task_completed": True
}
# Supervisor Agent - coordinates the workflow
class SupervisorAgent:
def __init__(self, llm):
self.llm = llm
self.name = "Supervisor"
def supervise(self, state: MultiAgentState) -> dict:
if state.get("task_completed", False):
return {
"messages": [AIMessage(content="Task completed successfully!")],
"current_agent": "end"
}
else:
# Determine next step
current_agent = state.get("current_agent", "researcher")
return {
"messages": [AIMessage(content=f"Coordinating workflow, next: {current_agent}")],
"current_agent": current_agent
}
def create_multi_agent_system():
"""Create a multi-agent research system"""
# Initialize agents
llm = ChatOpenAI(temperature=0, model="gpt-4")
search = SerpAPIWrapper()
researcher = ResearchAgent(llm, search)
analyst = AnalysisAgent(llm)
writer = WritingAgent(llm)
supervisor = SupervisorAgent(llm)
# Build the graph
graph = StateGraph(MultiAgentState)
# Add agent nodes
graph.add_node("researcher", researcher.research)
graph.add_node("analyst", analyst.analyze)
graph.add_node("writer", writer.write_report)
graph.add_node("supervisor", supervisor.supervise)
# Define routing logic
def route_to_next_agent(state: MultiAgentState) -> Literal["researcher", "analyst", "writer", "supervisor", "end"]:
current_agent = state.get("current_agent", "researcher")
if current_agent == "end":
return "end"
elif current_agent == "researcher":
return "researcher"
elif current_agent == "analyst":
return "analyst"
elif current_agent == "writer":
return "writer"
else:
return "supervisor"
# Set up the workflow
graph.add_edge(START, "researcher")
# Add conditional routing
graph.add_conditional_edges(
"researcher",
lambda state: state["current_agent"],
{
"analyst": "analyst",
"end": END
}
)
graph.add_conditional_edges(
"analyst",
lambda state: state["current_agent"],
{
"writer": "writer",
"end": END
}
)
graph.add_conditional_edges(
"writer",
lambda state: state["current_agent"],
{
"supervisor": "supervisor",
"end": END
}
)
graph.add_conditional_edges(
"supervisor",
lambda state: state["current_agent"],
{
"end": END
}
)
return graph.compile(checkpointer=MemorySaver())
# Test the multi-agent system
multi_agent_system = create_multi_agent_system()
result = multi_agent_system.invoke({
"messages": [HumanMessage(content="Research the impact of artificial intelligence on healthcare")],
"current_agent": "researcher",
"task_completed": False
})
# Print the full conversation
for message in result["messages"]:
print(f"\n{message.content}")
Human-in-the-Loop Agent Systems
For critical applications, you want human oversight and approval:
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
import uuid
class HumanApprovalState(TypedDict):
messages: Annotated[list, operator.add]
pending_action: str
approved: bool
human_input: str
def create_human_approval_agent():
"""Create an agent that requires human approval for actions"""
def agent_planning(state: HumanApprovalState):
"""Agent plans what action to take"""
last_message = state["messages"][-1]
# Determine action needed
if "email" in last_message.content.lower():
action = "send_email"
elif "file" in last_message.content.lower():
action = "create_file"
else:
action = "general_response"
return {
"messages": [AIMessage(content=f"I plan to: {action}")],
"pending_action": action,
"approved": False
}
def human_approval(state: HumanApprovalState):
"""Wait for human approval"""
action = state["pending_action"]
# In a real application, this would be a UI interaction
print(f"\nAgent wants to perform: {action}")
print("Do you approve? (yes/no)")
# For demo, we'll simulate approval
approval = input().lower().strip()
if approval in ["yes", "y", "approve"]:
return {
"messages": [AIMessage(content="Action approved by human")],
"approved": True
}
else:
return {
"messages": [AIMessage(content="Action rejected by human")],
"approved": False
}
def execute_action(state: HumanApprovalState):
"""Execute the approved action"""
if not state.get("approved", False):
return {
"messages": [AIMessage(content="Cannot execute - action not approved")]
}
action = state["pending_action"]
if action == "send_email":
result = "Email sent successfully"
elif action == "create_file":
result = "File created successfully"
else:
result = "General response provided"
return {
"messages": [AIMessage(content=f"Action completed: {result}")]
}
# Build graph with human approval step
graph = StateGraph(HumanApprovalState)
graph.add_node("planning", agent_planning)
graph.add_node("human_approval", human_approval)
graph.add_node("execute", execute_action)
# Define flow
graph.add_edge(START, "planning")
graph.add_edge("planning", "human_approval")
# Conditional execution based on approval
graph.add_conditional_edges(
"human_approval",
lambda state: "execute" if state.get("approved", False) else "end",
{"execute": "execute", "end": END}
)
graph.add_edge("execute", END)
return graph.compile(checkpointer=MemorySaver())
# Test human approval system
human_agent = create_human_approval_agent()
result = human_agent.invoke({
"messages": [HumanMessage(content="Please send an email to the team")],
"pending_action": "",
"approved": False
})
Advanced Agent Patterns
Agent with Tools and Memory
class AdvancedAgentSystem:
def __init__(self):
self.llm = ChatOpenAI(temperature=0, model="gpt-4")
self.memory = MemorySaver()
self.tools = self._setup_tools()
def _setup_tools(self):
"""Setup various tools for the agent"""
return {
"calculator": calculator,
"search": search.run,
"file_writer": self._write_file,
"current_time": self._get_current_time
}
def _write_file(self, filename: str, content: str) -> str:
"""Write content to a file"""
try:
with open(filename, 'w') as f:
f.write(content)
return f"Successfully wrote to {filename}"
except Exception as e:
return f"Failed to write file: {e}"
def _get_current_time(self) -> str:
"""Get current time"""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def create_agent_graph(self):
"""Create the main agent graph"""
def agent_node(state: AgentState):
"""Main agent reasoning node"""
messages = state["messages"]
last_message = messages[-1].content
# Agent decides what tool to use
tool_decision_prompt = f"""
Given this user request: {last_message}
Available tools:
- calculator: for math calculations
- search: for web searches
- file_writer: to create files
- current_time: to get current time
Which tool should be used? Respond with just the tool name, or "none" if no tool needed.
"""
tool_choice = self.llm.invoke([HumanMessage(content=tool_decision_prompt)])
tool_name = tool_choice.content.strip().lower()
if tool_name in self.tools:
return {
"messages": [AIMessage(content=f"Using tool: {tool_name}")],
"current_task": tool_name
}
else:
# Answer directly
response = self.llm.invoke([HumanMessage(content=last_message)])
return {
"messages": [response],
"current_task": "complete"
}
def tool_execution_node(state: AgentState):
"""Execute the chosen tool"""
tool_name = state["current_task"]
last_message = state["messages"][-2].content # Original user message
try:
if tool_name == "calculator":
# Extract mathematical expression
result = self.tools[tool_name](last_message)
elif tool_name == "search":
result = self.tools[tool_name](last_message)
elif tool_name == "current_time":
result = self.tools[tool_name]()
else:
result = "Tool execution not implemented"
return {
"messages": [AIMessage(content=result)],
"current_task": "complete"
}
except Exception as e:
return {
"messages": [AIMessage(content=f"Tool execution failed: {e}")],
"current_task": "complete"
}
# Build the graph
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tool_execution", tool_execution_node)
graph.add_edge(START, "agent")
# Conditional routing
graph.add_conditional_edges(
"agent",
lambda state: "tool_execution" if state["current_task"] != "complete" else "end",
{"tool_execution": "tool_execution", "end": END}
)
graph.add_edge("tool_execution", END)
return graph.compile(checkpointer=self.memory)
# Create and test advanced agent
advanced_agent_system = AdvancedAgentSystem()
advanced_agent = advanced_agent_system.create_agent_graph()
# Test various capabilities
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
tests = [
"What's the current time?",
"Calculate 15 * 24 + 300",
"Search for recent developments in quantum computing",
"Write a summary to a file called summary.txt"
]
for test_query in tests:
print(f"\nQuery: {test_query}")
result = advanced_agent.invoke({
"messages": [HumanMessage(content=test_query)],
"current_task": "",
"tool_results": {}
}, config)
print(f"Response: {result['messages'][-1].content}")
Key Takeaways from Part 3
Congratulations! You’ve completed the full GenAI development journey. Here’s what you’ve mastered:
Agent Fundamentals: Understanding the difference between chains and agents, and when to use each approach.
LangGraph Architecture: Building stateful, controllable agent systems using modern graph-based approaches.
Multi-Agent Systems: Creating collaborative systems where specialized agents work together to solve complex problems.
Human-in-the-Loop: Implementing systems that require human oversight and approval for critical actions.
Production Patterns: Building robust, memory-enabled agents with comprehensive tool integration and error handling.
Advanced Orchestration: Understanding how to coordinate complex workflows with multiple decision points and conditional logic.
Next Steps in Your GenAI Journey
You now have the complete foundation to build production-ready GenAI applications. Consider exploring:
LangSmith: For observability, debugging, and optimization of your agent systems.
LangServe: For deploying your agents as scalable API services.
Advanced RAG: Implementing more sophisticated retrieval patterns and evaluation frameworks.
Custom Tools: Building domain-specific tools for your agents to use.
Enterprise Integration: Connecting your agents to existing business systems and databases.
Evaluation and Monitoring: Implementing comprehensive testing and monitoring for production agent deployments.
The world of GenAI is evolving rapidly, but with the solid foundation you’ve built through this series, you’re well-equipped to adapt to new developments and build the next generation of intelligent applications. Remember that the best way to deepen your understanding is through hands-on practice – start building, experimenting, and iterating on real projects!
This content originally appeared on DEV Community and was authored by Ajmal Hasan