Gemini × LangChain Integration — Build RAG, Chains & Agents
LangChain provides an optimal framework for leveraging Gemini's capabilities in production applications. This guide walks you through everything from basic setup to advanced agent construction.
Environment Setup and Configuration
Installing Required Packages
pip install langchain langchain-google-genai chromadb python-dotenvAPI Key Configuration
import os
from dotenv import load_dotenv
load_dotenv()
google_api_key = os.getenv("GOOGLE_API_KEY")
os.environ["GOOGLE_API_KEY"] = google_api_keyObtain your Google API key from [Google AI Studio](https://aistudio.google.com/app/apikey). In production, always manage credentials securely using environment variables.
Basic Chat Model Setup
Initializing ChatGoogleGenerativeAI
from langchain_google_genai import ChatGoogleGenerativeAI
# Basic model initialization
model = ChatGoogleGenerativeAI(
model="gemini-2.5-pro",
temperature=0.7,
max_tokens=1024
)
# Simple conversation
response = model.invoke("How do I implement a simple calculator in Python?")
print(response.content)Streaming Output
# Streaming-enabled initialization
model_streaming = ChatGoogleGenerativeAI(
model="gemini-2.5-flash",
streaming=True,
temperature=0.5
)
# Token-by-token output
for chunk in model_streaming.stream("Explain machine learning fundamentals in 5 minutes"):
print(chunk.content, end="", flush=True)When streaming is enabled, API responses are returned incrementally, reducing overall processing time. However, error handling becomes more complex, so choose based on your use case.
Prompt Templates and Chain Building
Basic Prompt Template
from langchain.prompts import ChatPromptTemplate
# Template definition
template = """You are a {role}.
Provide helpful and professional answers to user questions.
Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
# Chain construction using LangChain Expression Language
chain = prompt | model
# Execution
result = chain.invoke({
"role": "Python Expert",
"question": "What are the fundamentals of asynchronous programming?"
})
print(result.content)Multi-Step Chains
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import StrOutputParser
# Step 1: Requirements Analysis
analysis_template = """Analyze the following requirements and list key implementation points.
Requirements:
{requirement}
Analysis:"""
analysis_prompt = ChatPromptTemplate.from_template(analysis_template)
analysis_chain = analysis_prompt | model | StrOutputParser()
# Step 2: Implementation Proposal
impl_template = """Based on the analysis below, propose a Python implementation approach.
Analysis:
{analysis}
Implementation Approach:"""
impl_prompt = ChatPromptTemplate.from_template(impl_template)
impl_chain = impl_prompt | model | StrOutputParser()
# Pipeline construction
full_chain = {
"requirement": lambda x: x,
"analysis": analysis_chain
} | (lambda x: impl_chain.invoke(x))
# Execution
requirement = "I want to implement an API client with caching functionality"
result = full_chain(requirement)
print(result)Building RAG Pipelines
Preparing the Data Store
from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_google_genai import GoogleGenerativeAIEmbeddings
from langchain.vectorstores import Chroma
# Load documents
loader = TextLoader("documents/gemini_guide.txt")
documents = loader.load()
# Text splitting
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
splits = splitter.split_documents(documents)
# Initialize embedding model
embeddings = GoogleGenerativeAIEmbeddings(
model="models/embedding-001"
)
# Create vector database
vectorstore = Chroma.from_documents(
documents=splits,
embedding=embeddings,
persist_directory="./chroma_db"
)Building the RAG Chain
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.prompts import ChatPromptTemplate
# Configure retriever
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 3}
)
# RAG prompt
rag_prompt = ChatPromptTemplate.from_template("""Answer the question based on the provided documents.
<context>
{context}
</context>
Question: {input}
Answer:""")
# Build RAG chain
combine_docs_chain = create_stuff_documents_chain(model, rag_prompt)
rag_chain = create_retrieval_chain(retriever, combine_docs_chain)
# Execute query
result = rag_chain.invoke({
"input": "What are the advanced features of Gemini?"
})
print(result["answer"])ChromaDB is optimal for development environments as an in-memory vector store. For production, consider using specialized services like Vertex AI Vector Search or Weaviate.
Building ReAct Agents
Defining Tools
from langchain.tools import tool
from typing import Any
@tool
def search_web(query: str) -> str:
"""Search the web for information"""
# Example implementation (use actual API in production)
return f"Search results for '{query}'..."
@tool
def calculate(expression: str) -> float:
"""Calculate a mathematical expression"""
try:
return eval(expression)
except Exception as e:
return f"Calculation error: {str(e)}"
@tool
def get_current_time() -> str:
"""Get the current time"""
from datetime import datetime
return datetime.now().isoformat()
tools = [search_web, calculate, get_current_time]Building and Running the Agent
from langchain.agents import initialize_agent
from langchain.agents import AgentType
# Initialize ReAct agent
agent = initialize_agent(
tools=tools,
llm=model,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True,
max_iterations=10,
handle_parsing_errors=True
)
# Run agent
response = agent.run(
"What time will it be in one hour? Also search for the weather forecast at that time."
)
print(response)Complex Workflow with Custom Tools
from langchain.tools import tool
from pydantic import BaseModel, Field
class SearchInput(BaseModel):
query: str = Field(description="Search keywords")
filters: str = Field(default="", description="Search filters")
@tool(args_schema=SearchInput)
def advanced_search(query: str, filters: str = "") -> str:
"""Advanced search functionality"""
result = f"Query: {query}"
if filters:
result += f" (Filters: {filters})"
return result
# Update tools for agent
custom_tools = [advanced_search, calculate, get_current_time]
agent = initialize_agent(
tools=custom_tools,
llm=model,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)Agents risk infinite loops. Always set `max_iterations` and `max_execution_time` to prevent unexpected behavior.
Memory Management and Context Retention
Managing Conversation History
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
# Initialize memory
memory = ConversationBufferMemory()
# Build conversation chain
conversation = ConversationChain(
llm=model,
memory=memory,
verbose=True
)
# Execute multiple conversations
response1 = conversation.predict(input="My name is Alice.")
response2 = conversation.predict(input="What is my name?")
response3 = conversation.predict(input="Tell me about Python.")
print(memory.buffer)Token-Limited Memory
from langchain.memory import ConversationTokenBufferMemory
# Token-limited memory
memory = ConversationTokenBufferMemory(
llm=model,
max_token_limit=1000
)
conversation = ConversationChain(
llm=model,
memory=memory,
verbose=True
)
# Usage is the same
response = conversation.predict(input="This demonstrates continuous conversation support.")Error Handling
Exception Handling Best Practices
from langchain_google_genai import ChatGoogleGenerativeAI
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def safe_invoke(model: ChatGoogleGenerativeAI, prompt: str) -> str:
"""Safe model invocation"""
try:
response = model.invoke(prompt)
return response.content
except Exception as e:
logger.error(f"API error: {str(e)}")
return "Sorry, an error occurred."
# Usage example
result = safe_invoke(model, "This is a test question")
print(result)Retry Logic
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
def invoke_with_retry(model, prompt):
"""Invocation with retry support"""
return model.invoke(prompt).content
try:
result = invoke_with_retry(model, "A retry-enabled query")
except Exception as e:
logger.error(f"Max retries reached: {str(e)}")Performance Optimization
Batch Processing
from typing import List
def batch_invoke(model: ChatGoogleGenerativeAI, prompts: List[str]) -> List[str]:
"""Process multiple prompts in batch"""
results = []
for prompt in prompts:
try:
response = model.invoke(prompt)
results.append(response.content)
except Exception as e:
logger.warning(f"Processing failed for: {prompt[:50]}...")
results.append("")
return results
# Usage example
prompts = [
"Explain Python type hints",
"List 5 benefits of async programming",
"FastAPI error handling methods"
]
results = batch_invoke(model, prompts)
for prompt, result in zip(prompts, results):
print(f"Q: {prompt[:30]}...\nA: {result[:100]}...\n")Asynchronous Processing
import asyncio
from langchain_google_genai import ChatGoogleGenerativeAI
from typing import List
async def async_invoke(model: ChatGoogleGenerativeAI, prompts: List[str]) -> List[str]:
"""Asynchronous batch processing"""
tasks = [
asyncio.to_thread(model.invoke, prompt)
for prompt in prompts
]
responses = await asyncio.gather(*tasks)
return [r.content for r in responses]
# Execution
if __name__ == "__main__":
prompts = [
"Machine learning fundamentals",
"Data analysis best practices",
"API design anti-patterns"
]
results = asyncio.run(async_invoke(model, prompts))Looking back
The combination of LangChain and Gemini enables powerful applications:
- RAG Pipelines: Intelligent Q&A systems leveraging internal knowledge bases
- Agents: Autonomous agents that automatically utilize multiple tools
- Chains: Building and managing complex processing workflows
- Memory Management: Retaining conversation context over extended interactions
Always consider API rate limits, error handling, and token management in your implementations.
A note from the field
Running the Dolice Labs tooling as an indie developer, I learned to use LangChain thinly: build the minimum yourself, hand only what you must to the framework, and log every intermediate step. When RAG quality drops, I inspect the retrieved context first — if the answer isn't even in the candidates, the fix is chunking or embeddings, not the prompt.