Introduction to Sales Outreach Agents with LangGraph
A sales outreach agent is an AI-powered system that automates the process of researching prospects, personalizing outreach messages, and managing follow-ups across multiple channels. LangGraph, built on top of LangChain, provides a stateful, graph-based framework for constructing such agents with precise control over execution flow, state management, and tool integration.
Unlike traditional linear pipelines, LangGraph allows you to define complex, conditional workflows as directed graphs—making it ideal for sales outreach where decisions depend on prospect data, channel preferences, and prior interactions. This tutorial will walk you through building a complete sales outreach agent from scratch, covering research, personalization, channel selection, and follow-up logic.
Why LangGraph for Sales Outreach Matters
Sales outreach typically involves multiple interconnected steps that don't always follow a straight line. You might need to research a company, check if the prospect is active on LinkedIn, decide between email or social outreach, handle rate limits, and schedule follow-ups based on engagement signals. Traditional sequential code becomes unwieldy with branching logic, retries, and state dependencies.
LangGraph addresses these challenges by offering:
- Explicit state management — a single state object flows through all nodes, giving you full visibility into what data is available at each step
- Conditional routing — edges can evaluate runtime conditions to dynamically choose the next node
- Human-in-the-loop — you can pause execution for approval before sending messages
- Streaming and observability — watch agent decisions in real-time and debug complex flows
- Tool integration — seamlessly connect to CRM APIs, email services, LinkedIn, and research tools
Prerequisites and Installation
Before building the agent, install the required packages. You'll need LangGraph, LangChain, and a few utility libraries for API calls and email integration.
pip install langgraph langchain langchain-openai langchain-community
pip install requests python-dotenv sendgrid
pip install tavily-python # for web research capabilities
Set up your environment variables for API keys:
# .env file
OPENAI_API_KEY=your-openai-api-key
TAVILY_API_KEY=your-tavily-api-key # for web search
SENDGRID_API_KEY=your-sendgrid-api-key # for email sending
LINKEDIN_API_TOKEN=your-linkedin-token # optional, for social outreach
Core Architecture Overview
The sales outreach agent consists of several key nodes in a directed graph:
- Research Node — gathers prospect information using web search and CRM lookups
- Profile Analysis Node — extracts key insights (industry, pain points, recent news)
- Personalization Node — crafts tailored messaging based on research
- Channel Selection Node — decides best outreach channel (email, LinkedIn, phone)
- Send Node — executes the outreach via the chosen channel
- Follow-up Scheduler Node — plans next steps based on response status
The state flows between these nodes, accumulating data and decisions at each step.
Step 1: Define the Agent State
The state is the central data structure that persists across all nodes. Define it using Python's TypedDict for type safety.
from typing import TypedDict, List, Optional, Literal, Dict
from datetime import datetime
class SalesOutreachState(TypedDict):
# Input fields
prospect_name: str
prospect_company: str
prospect_email: Optional[str]
prospect_linkedin_url: Optional[str]
# Research output
company_info: Optional[str]
recent_news: Optional[List[str]]
pain_points: Optional[List[str]]
# Personalization
personalized_message: Optional[str]
message_subject: Optional[str]
tone: str # professional, casual, direct
# Channel decision
selected_channel: Optional[Literal["email", "linkedin", "phone"]]
channel_reasoning: Optional[str]
# Execution
message_sent: bool
send_timestamp: Optional[str]
response_status: Optional[Literal["pending", "replied", "no_response", "bounced"]]
# Follow-up
follow_up_scheduled: bool
follow_up_date: Optional[str]
follow_up_message: Optional[str]
# Control flow
requires_human_approval: bool
human_approved: bool
error_message: Optional[str]
Step 2: Build the Research Node
The research node uses web search and a mock CRM lookup to gather intelligence about the prospect and their company. In production, you'd connect to real APIs like Clearbit, Apollo, or your CRM.
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.prompts import ChatPromptTemplate
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
tavily_search = TavilySearchResults(max_results=5)
def research_node(state: SalesOutreachState) -> SalesOutreachState:
"""
Gathers company and prospect information using web search.
"""
print(f"🔍 Researching {state['prospect_name']} at {state['prospect_company']}...")
# Search for company information
company_query = f"{state['prospect_company']} company overview products services"
company_results = tavily_search.invoke(company_query)
company_info = "\n".join([r["content"] for r in company_results[:3]])
# Search for recent news
news_query = f"{state['prospect_company']} recent news announcements 2024"
news_results = tavily_search.invoke(news_query)
recent_news = [r["content"] for r in news_results[:3]]
# Analyze pain points using LLM
pain_point_prompt = ChatPromptTemplate.from_messages([
("system", "You are a sales intelligence analyst. Based on the company information and news, identify 3-5 potential business pain points or challenges this company might face. Return them as a concise list."),
("user", "Company: {company}\nInfo: {info}\nRecent News: {news}\n\nIdentify their likely pain points.")
])
pain_point_chain = pain_point_prompt | llm
pain_points_response = pain_point_chain.invoke({
"company": state["prospect_company"],
"info": company_info,
"news": "\n".join(recent_news)
})
# Parse pain points from response
pain_points = [
line.strip().lstrip("- ")
for line in pain_points_response.content.split("\n")
if line.strip() and len(line.strip()) > 5
]
updated_state = state.copy()
updated_state["company_info"] = company_info
updated_state["recent_news"] = recent_news
updated_state["pain_points"] = pain_points
return updated_state
Step 3: Build the Personalization Node
This node crafts a personalized message based on research findings. It generates both the subject line and body, adapting tone based on the prospect's industry signals.
personalization_prompt = ChatPromptTemplate.from_messages([
("system", """You are an expert sales copywriter specializing in B2B outreach.
Given the prospect information and research, craft a personalized outreach message.
Guidelines:
- Reference specific findings from research (company news, pain points)
- Keep it concise (3-4 sentences max for email, slightly longer for LinkedIn)
- Use the specified tone
- Include a clear, low-pressure call-to-action
- Never use generic templates—every message must feel custom-crafted
Return the output in this format:
SUBJECT: [compelling subject line]
BODY: [message body]"""),
("user", """Prospect: {prospect_name}
Company: {prospect_company}
Company Info: {company_info}
Recent News: {recent_news}
Pain Points: {pain_points}
Tone: {tone}
Craft a personalized outreach message.""")
])
def personalization_node(state: SalesOutreachState) -> SalesOutreachState:
"""
Creates a personalized message based on research findings.
"""
print(f"✍️ Crafting personalized message for {state['prospect_name']}...")
chain = personalization_prompt | llm
response = chain.invoke({
"prospect_name": state["prospect_name"],
"prospect_company": state["prospect_company"],
"company_info": state.get("company_info", "No company info available"),
"recent_news": "\n".join(state.get("recent_news", ["No recent news"])),
"pain_points": "\n".join(state.get("pain_points", ["Not identified"])),
"tone": state.get("tone", "professional")
})
# Parse the response
content = response.content
subject = ""
body = ""
if "SUBJECT:" in content and "BODY:" in content:
subject_part = content.split("SUBJECT:")[1].split("BODY:")[0].strip()
body = content.split("BODY:")[1].strip()
subject = subject_part
else:
body = content
updated_state = state.copy()
updated_state["personalized_message"] = body
updated_state["message_subject"] = subject
return updated_state
Step 4: Build the Channel Selection Node
This node intelligently selects the best outreach channel based on available contact information, prospect seniority signals, and industry preferences.
channel_prompt = ChatPromptTemplate.from_messages([
("system", """You are a sales strategist deciding the optimal outreach channel.
Consider these factors:
- If LinkedIn URL is available and prospect is senior-level (VP+), LinkedIn InMail often works better
- If only email is available, use email
- If both available, analyze industry: tech/startup favors LinkedIn, traditional industries favor email
- Phone outreach is best for urgent, time-sensitive opportunities
Respond with exactly one of: "email", "linkedin", or "phone"
Then provide a brief reason starting with "REASON:" """),
("user", """Prospect: {prospect_name}
Company: {prospect_company}
Email available: {has_email}
LinkedIn available: {has_linkedin}
Company Info: {company_info}
Select the best outreach channel.""")
])
def channel_selection_node(state: SalesOutreachState) -> SalesOutreachState:
"""
Selects the optimal outreach channel based on available data.
"""
print(f"📡 Selecting channel for {state['prospect_name']}...")
has_email = bool(state.get("prospect_email"))
has_linkedin = bool(state.get("prospect_linkedin_url"))
# If no contact methods available, flag for enrichment
if not has_email and not has_linkedin:
updated_state = state.copy()
updated_state["error_message"] = "No contact methods available. Requires manual enrichment."
updated_state["requires_human_approval"] = True
return updated_state
chain = channel_prompt | llm
response = chain.invoke({
"prospect_name": state["prospect_name"],
"prospect_company": state["prospect_company"],
"has_email": has_email,
"has_linkedin": has_linkedin,
"company_info": state.get("company_info", "Unknown")
})
content = response.content.lower()
# Parse channel selection
channel = "email" # default
if "linkedin" in content:
channel = "linkedin"
elif "phone" in content:
channel = "phone"
# Extract reasoning
reason = ""
if "REASON:" in content.upper():
reason = content.split("REASON:")[-1].strip()
updated_state = state.copy()
updated_state["selected_channel"] = channel
updated_state["channel_reasoning"] = reason
return updated_state
Step 5: Build the Send Node with Human Approval Gate
The send node executes the actual outreach. For sensitive or high-value prospects, it can pause for human approval before sending. This demonstrates LangGraph's human-in-the-loop capability.
import json
from datetime import datetime
def send_node(state: SalesOutreachState) -> SalesOutreachState:
"""
Sends the outreach message via the selected channel.
Supports email (via SendGrid) and mock LinkedIn/phone implementations.
"""
print(f"📨 Sending message via {state['selected_channel']}...")
channel = state["selected_channel"]
message = state.get("personalized_message", "")
subject = state.get("message_subject", "")
updated_state = state.copy()
try:
if channel == "email":
# Real SendGrid integration
if state.get("prospect_email"):
# In production, uncomment the SendGrid API call
# sendgrid_client = SendGridAPIClient(os.getenv("SENDGRID_API_KEY"))
# mail = Mail(
# from_email="sales@yourcompany.com",
# to_emails=state["prospect_email"],
# subject=subject,
# plain_text_content=message
# )
# response = sendgrid_client.send(mail)
print(f" ✅ Email sent to {state['prospect_email']}")
print(f" Subject: {subject}")
print(f" Body preview: {message[:100]}...")
updated_state["message_sent"] = True
updated_state["send_timestamp"] = datetime.now().isoformat()
else:
updated_state["error_message"] = "No email address available"
elif channel == "linkedin":
# Mock LinkedIn InMail (in production, use LinkedIn API)
print(f" 📨 [LinkedIn] Would send InMail to {state['prospect_name']}")
print(f" Message preview: {message[:100]}...")
updated_state["message_sent"] = True
updated_state["send_timestamp"] = datetime.now().isoformat()
elif channel == "phone":
# Mock phone outreach (in production, integrate with dialer API)
print(f" 📞 [Phone] Would call {state['prospect_name']}")
print(f" Talk track preview: {message[:100]}...")
updated_state["message_sent"] = True
updated_state["send_timestamp"] = datetime.now().isoformat()
updated_state["response_status"] = "pending"
except Exception as e:
updated_state["error_message"] = f"Send failed: {str(e)}"
updated_state["message_sent"] = False
return updated_state
def human_approval_gate(state: SalesOutreachState) -> str:
"""
Conditional edge function: routes to send node if approved,
or pauses for human review if requires_human_approval is True.
"""
if state.get("requires_human_approval") and not state.get("human_approved"):
print("⏸️ Pausing for human approval...")
return "pause_for_approval"
return "proceed_to_send"
Step 6: Build the Follow-Up Scheduler Node
After sending, the agent plans follow-ups based on response status. This node generates a follow-up message and schedules it for the appropriate timeframe.
followup_prompt = ChatPromptTemplate.from_messages([
("system", """You are a sales follow-up specialist. Based on the outreach status,
create an appropriate follow-up strategy.
- If no response after 3 days: send a gentle follow-up referencing the original message
- If the prospect replied: suggest a meeting time and thank them
- If the message bounced: find alternative contact methods
- Always be respectful and never pushy
Return format:
FOLLOWUP_DATE: [suggested date, 3-5 days from now]
FOLLOWUP_MESSAGE: [the follow-up message]"""),
("user", """Original message: {original_message}
Channel: {channel}
Response status: {status}
Plan the follow-up strategy.""")
])
def followup_scheduler_node(state: SalesOutreachState) -> SalesOutreachState:
"""
Schedules appropriate follow-up based on outreach outcome.
"""
print(f"📅 Planning follow-up for {state['prospect_name']}...")
status = state.get("response_status", "pending")
# If message wasn't sent, skip follow-up
if not state.get("message_sent"):
updated_state = state.copy()
updated_state["follow_up_scheduled"] = False
return updated_state
chain = followup_prompt | llm
response = chain.invoke({
"original_message": state.get("personalized_message", ""),
"channel": state.get("selected_channel", "email"),
"status": status
})
content = response.content
# Parse follow-up details
followup_date = ""
followup_message = ""
if "FOLLOWUP_DATE:" in content:
date_part = content.split("FOLLOWUP_DATE:")[1].split("FOLLOWUP_MESSAGE:")[0].strip()
followup_date = date_part
if "FOLLOWUP_MESSAGE:" in content:
followup_message = content.split("FOLLOWUP_MESSAGE:")[1].strip()
updated_state = state.copy()
updated_state["follow_up_scheduled"] = True
updated_state["follow_up_date"] = followup_date
updated_state["follow_up_message"] = followup_message
print(f" Follow-up scheduled for: {followup_date}")
print(f" Follow-up preview: {followup_message[:80]}...")
return updated_state
Step 7: Assemble the Graph
Now wire all nodes together using LangGraph's StateGraph. Define nodes, edges, and conditional routing logic.
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
# Create the graph
builder = StateGraph(SalesOutreachState)
# Add nodes
builder.add_node("research", research_node)
builder.add_node("personalize", personalization_node)
builder.add_node("select_channel", channel_selection_node)
builder.add_node("send", send_node)
builder.add_node("schedule_followup", followup_scheduler_node)
# Add edges
builder.add_edge(START, "research")
builder.add_edge("research", "personalize")
builder.add_edge("personalize", "select_channel")
# Conditional edge: check for human approval or enrichment needs
def after_channel_selection(state: SalesOutreachState) -> str:
if state.get("error_message") and "No contact methods" in state.get("error_message", ""):
return END # Stop, needs manual intervention
return "send"
builder.add_conditional_edges(
"select_channel",
after_channel_selection,
{
"send": "send",
END: END
}
)
builder.add_edge("send", "schedule_followup")
builder.add_edge("schedule_followup", END)
# Compile with memory checkpoint for state persistence
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)
Step 8: Run the Agent
Execute the graph with an initial state. The agent will flow through all nodes, and you can inspect the final state.
import asyncio
async def run_sales_outreach():
# Initialize with prospect data
initial_state: SalesOutreachState = {
"prospect_name": "Jane Smith",
"prospect_company": "Acme Cloud Solutions",
"prospect_email": "jane.smith@acmecloud.com",
"prospect_linkedin_url": "https://linkedin.com/in/janesmith",
"prospect_email": "jane.smith@acmecloud.com",
"tone": "professional",
"requires_human_approval": False,
"human_approved": True,
"message_sent": False,
"follow_up_scheduled": False,
"company_info": None,
"recent_news": None,
"pain_points": None,
"personalized_message": None,
"message_subject": None,
"selected_channel": None,
"channel_reasoning": None,
"send_timestamp": None,
"response_status": None,
"follow_up_date": None,
"follow_up_message": None,
"error_message": None
}
# Run the graph with a thread ID for state tracking
config = {"configurable": {"thread_id": "outreach-001"}}
print("🚀 Starting Sales Outreach Agent...\n")
async for event in graph.astream(initial_state, config):
# Stream events to see real-time progress
for node_name, node_output in event.items():
if isinstance(node_output, dict):
# Print key state changes
if "personalized_message" in node_output and node_output["personalized_message"]:
print(f"\n📝 Generated message:")
print(f" Subject: {node_output.get('message_subject', 'N/A')}")
print(f" Body: {node_output['personalized_message'][:150]}...")
if "selected_channel" in node_output:
print(f"\n📡 Channel: {node_output['selected_channel']}")
if "follow_up_date" in node_output:
print(f"\n📅 Follow-up: {node_output['follow_up_date']}")
# Retrieve final state
final_state = graph.get_state(config).values
print("\n" + "="*60)
print("✅ OUTREACH COMPLETE")
print("="*60)
print(f"Prospect: {final_state['prospect_name']} @ {final_state['prospect_company']}")
print(f"Channel used: {final_state['selected_channel']}")
print(f"Message sent: {final_state['message_sent']}")
print(f"Follow-up scheduled: {final_state['follow_up_scheduled']}")
if final_state['follow_up_date']:
print(f"Next follow-up: {final_state['follow_up_date']}")
return final_state
# Run synchronously if preferred
if __name__ == "__main__":
asyncio.run(run_sales_outreach())
Step 9: Adding Human-in-the-Loop Approval
For high-value prospects or sensitive messaging, you can pause execution and wait for human review. This is one of LangGraph's most powerful features.
def human_approval_node(state: SalesOutreachState) -> SalesOutreachState:
"""
This node is reached when the graph pauses for approval.
In production, this would integrate with Slack, email, or a UI.
"""
print("\n⏸️ === HUMAN APPROVAL REQUIRED ===")
print(f"Prospect: {state['prospect_name']}")
print(f"Company: {state['prospect_company']}")
print(f"Channel: {state['selected_channel']}")
print(f"Subject: {state.get('message_subject', 'N/A')}")
print(f"Message:\n{state.get('personalized_message', 'N/A')}")
print("⏸️ === Waiting for approval... ===\n")
# In a real system, this would trigger a notification
# and wait for external input. For demo, simulate approval.
updated_state = state.copy()
# Simulate: uncomment next line for real pause
# updated_state["human_approved"] = await wait_for_human_input()
updated_state["human_approved"] = True
updated_state["requires_human_approval"] = False
return updated_state
# Add to graph
builder.add_node("human_approval", human_approval_node)
def approval_router(state: SalesOutreachState) -> str:
if state.get("requires_human_approval") and not state.get("human_approved"):
return "human_approval"
return "send"
builder.add_conditional_edges(
"select_channel",
approval_router,
{
"human_approval": "human_approval",
"send": "send"
}
)
builder.add_edge("human_approval", "send")
Complete Graph with All Features
Here's the final, complete graph assembly with all nodes, conditional edges, and the human-in-the-loop gate integrated:
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
def build_complete_sales_agent():
"""
Builds the complete sales outreach agent graph.
"""
builder = StateGraph(SalesOutreachState)
# Register all nodes
builder.add_node("research", research_node)
builder.add_node("personalize", personalization_node)
builder.add_node("select_channel", channel_selection_node)
builder.add_node("human_approval", human_approval_node)
builder.add_node("send", send_node)
builder.add_node("schedule_followup", followup_scheduler_node)
# Main flow edges
builder.add_edge(START, "research")
builder.add_edge("research", "personalize")
builder.add_edge("personalize", "select_channel")
# Conditional: after channel selection, check approval and errors
def post_channel_router(state: SalesOutreachState) -> str:
# Check for critical errors
if state.get("error_message"):
if "No contact methods" in state.get("error_message", ""):
return "end_with_error"
# Check if human approval is needed
if state.get("requires_human_approval") and not state.get("human_approved"):
return "human_approval"
return "send"
builder.add_conditional_edges(
"select_channel",
post_channel_router,
{
"send": "send",
"human_approval": "human_approval",
"end_with_error": END
}
)
builder.add_edge("human_approval", "send")
builder.add_edge("send", "schedule_followup")
builder.add_edge("schedule_followup", END)
# Compile with checkpointing
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)
return graph
# Build and export the agent
sales_outreach_agent = build_complete_sales_agent()
Testing with Multiple Scenarios
Test the agent with different prospect profiles to verify routing logic works correctly.
async def test_scenarios():
"""Test the agent with various prospect configurations."""
# Scenario 1: Full information, email + LinkedIn
print("\n" + "="*60)
print("SCENARIO 1: Well-researched prospect with email and LinkedIn")
print("="*60)
state1: SalesOutreachState = {
"prospect_name": "Michael Chen",
"prospect_company": "TechNova AI",
"prospect_email": "michael@technova.ai",
"prospect_linkedin_url": "https://linkedin.com/in/michaelchen",
"tone": "casual",
"requires_human_approval": False,
"human_approved": True,
"message_sent": False,
"follow_up_scheduled": False,
# Initialize all optional fields as None
"company_info": None,
"recent_news": None,
"pain_points": None,
"personalized_message": None,
"message_subject": None,
"selected_channel": None,
"channel_reasoning": None,
"send_timestamp": None,
"response_status": None,
"follow_up_date": None,
"follow_up_message": None,
"error_message": None
}
config1 = {"configurable": {"thread_id": "test-001"}}
async for event in sales_outreach_agent.astream(state1, config1):
for node_name, _ in event.items():
print(f" ➤ Executed node: {node_name}")
final1 = sales_outreach_agent.get_state(config1).values
print(f"\n Result: Channel={final1['selected_channel']}, Sent={final1['message_sent']}")
# Scenario 2: Email only, requires human approval
print("\n" + "="*60)
print("SCENARIO 2: High-value prospect, email only, requires approval")
print("="*60)
state2: SalesOutreachState = {
"prospect_name": "Sarah Johnson",
"prospect_company": "Global Financial Partners",
"prospect_email": "sjohnson@globalfin.com",
"prospect_linkedin_url": None,
"tone": "professional",
"requires_human_approval": True,
"human_approved": False,
"message_sent": False,
"follow_up_scheduled": False,
"company_info": None,
"recent_news": None,
"pain_points": None,
"personalized_message": None,
"message_subject": None,
"selected_channel": None,
"channel_reasoning": None,
"send_timestamp": None,
"response_status": None,
"follow_up_date": None,
"follow_up_message": None,
"error_message": None
}
config2 = {"configurable": {"thread_id": "test-002"}}
async for event in sales_outreach_agent.astream(state2, config2):
for node_name, _ in event.items():
print(f" ➤ Executed node: {node_name}")
final2 = sales_outreach_agent.get_state(config2).values
print(f"\n Result: Approved={final2['human_approved']}, Sent={final2['message_sent']}")
# Scenario 3: No contact methods (should error gracefully)
print("\n" + "="*60)
print("SCENARIO 3: Prospect with no contact methods")
print("="*60)
state3: SalesOutreachState = {
"prospect_name": "Unknown Prospect",
"prospect_company": "Some Startup",
"prospect_email": None,
"prospect_linkedin_url": None,
"tone": "professional",
"requires_human_approval": False,
"human_approved": True,
"message_sent": False,
"follow_up_scheduled": False,
"company_info": None,
"recent_news": None,
"pain_points": None,
"personalized_message": None,
"message_subject": None,
"selected_channel": None,
"channel_reasoning": None,
"send_timestamp": None,
"response_status": None,
"follow_up_date": None,
"follow_up_message": None,
"error_message": None
}
config3 = {"configurable": {"thread_id": "test-003"}}
async for event in sales_outreach_agent.astream(state3, config3):
for node_name, _ in event.items():
print(f" ➤ Executed node: {node_name}")
final3 = sales_outreach_agent.get_state(config3).values
print(f"\n Result: Error={final3.get('error_message', 'None')}")
print("\n" + "="*60)
print("✅ All test scenarios complete")
# Run tests
asyncio.run(test_scenarios())
Best Practices for Production Deployment
1. Implement Proper Error Handling and Retries
Network calls to research APIs, email services, and LLMs can fail. Wrap each node's external calls in retry logic with exponential backoff.
import time
from functools import wraps
def with_retry(max_retries=3, backoff_factor=2):
"""Decorator for retrying failed operations."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
wait_time = backoff_factor ** attempt
print(f" ⚠️ Attempt {attempt+1} failed, retrying in {wait_time}s...")
time.sleep(wait_time)
raise last_exception
return wrapper
return decorator
# Apply to research node's search calls
@with_retry(max_retries=3, backoff_factor=2)
def robust_search(query: str):
return tavily_search.invoke(query)
2. Use Configurable Checkpointing for State Persistence
For production, replace MemorySaver with a durable checkpoint system like Postgres or SQLite to survive server restarts.
from langgraph.checkpoint.postgres import PostgresSaver
# or
from langgraph.checkpoint.sqlite import SqliteSaver
# PostgreSQL checkpoint (production-ready)
# checkpointer = PostgresSaver.from_conn_string(
# "postgresql://user:password@host:5432/outreach_db"
# )
# SQLite checkpoint (lightweight persistence)
checkpointer = SqliteSaver.from_conn_string("sales_outreach.db")
graph = builder.compile(checkpointer=checkpointer)
3. Add Logging and Monitoring
Track every agent run with structured logging for debugging and analytics.
import logging
from datetime import datetime
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
handlers=[
logging.FileHandler('sales_agent.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
def log_state_transition(node_name: str, state: SalesOutreachState):
"""Log key state changes for monitoring."""
logger.info(
f"Node: {node_name} | "
f"Prospect: {state.get('prospect_name')} | "
f"Channel: {state.get('selected_channel', 'pending')} | "
f"Sent: {state.get('message_sent', False)}"
)
4. Validate and Clean Prospect Data Early
Add a validation node at the start of the graph to ensure prospect data meets minimum requirements before spending API credits on research.
def validation_node(state: SalesOutreachState) -> SalesOut