Introduction: What is a Sales Outreach Agent?
A Sales Outreach Agent is an AI-powered autonomous system that researches prospects, crafts personalized messages, and manages multi-channel outreach campaigns—all with minimal human intervention. Built on LangGraph, a framework for constructing stateful, multi-actor agentic workflows, this agent can orchestrate complex sequences of actions: from scraping a lead's LinkedIn profile, to generating a tailored cold email, to scheduling follow-ups, and even adapting its strategy based on whether a prospect replies.
LangGraph extends LangChain's capabilities by introducing graph-based state machines that enable branching logic, persistent memory, and human-in-the-loop checkpoints. Unlike a simple linear chain, a LangGraph agent can loop back to refine its research, fork into different outreach channels, and recover gracefully from failures—making it ideal for the nuanced, multi-step nature of sales prospecting.
Why a LangGraph Sales Agent Matters
Traditional sales outreach automation suffers from three critical limitations:
- Rigid sequences — Most tools follow fixed if/then rules that cannot adapt when a research step returns unexpected data.
- No state persistence — If a script crashes mid-campaign, you lose all context and must restart from scratch.
- Generic personalization — Template-based "personalization" often amounts to {{first_name}} placeholders, which prospects increasingly ignore.
A LangGraph-based agent addresses all three. It maintains a rich, evolving state object across every node in its decision graph. It can pause execution to ask for human approval before sending a critical email. It can dynamically choose between different research tools based on what data is available. The result is outreach that feels genuinely human-crafted, operates reliably at scale, and continuously improves through feedback loops.
Core Concepts: LangGraph Fundamentals
The Graph Structure
In LangGraph, your agent is defined as a StateGraph—a directed graph where nodes represent actions (like "research lead" or "draft email") and edges represent transitions between actions. The graph carries a shared state dictionary that every node can read and modify. This state persists across the entire workflow, enabling complex, context-aware behavior.
Key Components
- State: A TypedDict or Pydantic model that defines what data flows through the graph (lead info, email drafts, conversation history, etc.).
- Nodes: Python functions that receive the current state and return updates. These are your atomic actions—calling an API, generating text, logging a result.
- Edges: Connections between nodes. Can be static (always go from A to B) or conditional (choose the next node based on a function's output).
- Conditional Edges: Functions that inspect the state and return the name of the next node to execute, enabling branching and looping.
- Checkpoints: LangGraph's built-in persistence layer that saves state after each node, enabling pause/resume and human-in-the-loop workflows.
Step 1: Setting Up Your Environment
Start by installing the required dependencies. You'll need LangGraph, LangChain (for LLM interactions), and a few utility libraries for web research and email sending.
# Create a new project directory and virtual environment
mkdir sales-outreach-agent
cd sales-outreach-agent
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Install core dependencies
pip install langgraph langchain langchain-openai langchain-community
pip install tavily-python # For web research
pip install exa-py # Alternative research tool (optional)
pip install sendgrid # For email delivery (or your preferred ESP)
pip install python-dotenv # For environment variable management
# Create a .env file for your API keys
echo 'OPENAI_API_KEY=your-key-here' > .env
echo 'TAVILY_API_KEY=your-key-here' >> .env
echo 'SENDGRID_API_KEY=your-key-here' >> .envStep 2: Defining the Agent State
The state is the backbone of your agent. Every node will operate on this shared object. Define it carefully to capture everything needed throughout the sales outreach lifecycle.
# state.py
from typing import TypedDict, Optional, List, Dict
from typing_extensions import Annotated
import operator
class SalesOutreachState(TypedDict):
"""State shared across all nodes in the sales outreach graph."""
# Input fields (provided at invocation)
prospect_name: str
company_name: str
prospect_linkedin_url: Optional[str]
prospect_email: Optional[str]
product_description: str
outreach_channel: str # "email" | "linkedin" | "both"
# Research fields (populated by research nodes)
research_notes: Annotated[str, operator.add] # Accumulated research findings
company_summary: Optional[str]
prospect_role: Optional[str]
prospect_interests: Optional[List[str]]
pain_points: Optional[List[str]]
recent_news: Optional[str]
# Drafting fields
email_draft: Optional[str]
linkedin_message_draft: Optional[str]
personalization_hooks: Optional[List[str]]
# Execution fields
email_sent: bool
linkedin_sent: bool
send_status: Optional[str]
# Feedback / iteration fields
human_approved: bool
revision_feedback: Optional[str]
iteration_count: int # Track how many revision loops we've done
# Final output
final_outcome: Optional[str]Note the Annotated[str, operator.add] for research_notes. This tells LangGraph to append new research findings rather than overwrite them, which is critical when multiple research nodes contribute information.
Step 3: Building the Research Nodes
Research is the foundation of effective personalization. You'll create nodes that gather intelligence from different sources and merge the findings into the shared state.
# research_nodes.py
import os
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from dotenv import load_dotenv
load_dotenv()
# Initialize LLM and research tool
llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
tavily_search = TavilySearchResults(
max_results=5,
tavily_api_key=os.getenv("TAVILY_API_KEY")
)
def research_company(state: SalesOutreachState) -> dict:
"""Research the prospect's company using web search."""
company = state["company_name"]
prospect = state["prospect_name"]
# Search for company overview and recent news
query = f"{company} company overview recent news 2024 business model target market"
search_results = tavily_search.invoke(query)
# Use LLM to synthesize findings
synthesis_prompt = f"""
Based on these search results about {company}, extract:
1. A 2-3 sentence company summary (what they do, who they serve)
2. Any recent news, funding rounds, product launches, or leadership changes
3. Potential pain points or challenges the company might be facing
Search results:
{search_results}
Format your response as:
COMPANY SUMMARY: [your summary]
RECENT NEWS: [bullet points of recent developments]
POTENTIAL PAIN POINTS: [bullet points]
"""
synthesis = llm.invoke(synthesis_prompt).content
return {
"company_summary": synthesis,
"research_notes": f"\n--- Company Research for {company} ---\n{synthesis}\n",
}
def research_prospect(state: SalesOutreachState) -> dict:
"""Research the individual prospect—role, interests, and personal hooks."""
prospect = state["prospect_name"]
company = state["company_name"]
linkedin_url = state.get("prospect_linkedin_url", "")
# Search for the prospect's professional background
query = f"{prospect} {company} role professional background interests achievements"
if linkedin_url:
query += f" linkedin"
search_results = tavily_search.invoke(query)
synthesis_prompt = f"""
Based on these search results about {prospect} at {company}, extract:
1. Their likely role and responsibilities
2. Professional interests, topics they engage with, or content they share
3. Recent professional achievements (talks, publications, awards)
4. 2-3 specific personalization hooks that could be used in outreach
Search results:
{search_results}
Format as:
ROLE: [role description]
INTERESTS: [bullet points]
ACHIEVEMENTS: [bullet points]
PERSONALIZATION HOOKS: [numbered list of specific, genuine hooks]
"""
synthesis = llm.invoke(synthesis_prompt).content
return {
"prospect_role": synthesis,
"research_notes": f"\n--- Prospect Research for {prospect} ---\n{synthesis}\n",
}
def synthesize_research(state: SalesOutreachState) -> dict:
"""Combine all research into a structured personalization brief."""
synthesis_prompt = f"""
You are preparing a personalization brief for a sales outreach message.
Product being sold: {state["product_description"]}
Research gathered:
{state["research_notes"]}
Create a concise personalization brief that includes:
1. The single most compelling reason this prospect would care about our product
2. Three specific personalization hooks ranked by strength
3. A recommended tone (e.g., formal, casual, technical, value-focused)
4. Any topics or angles to AVOID (e.g., recent negative news)
BRIEF:
"""
brief = llm.invoke(synthesis_prompt).content
# Extract personalization hooks as a list
hooks_prompt = f"Extract only the personalization hooks from this brief as a Python list of strings:\n{brief}"
hooks_response = llm.invoke(hooks_prompt).content
# Parse the hooks (handle both JSON and plain text formats)
try:
import json
hooks = json.loads(hooks_response)
except:
hooks = [line.strip("123456789. ") for line in hooks_response.split("\n")
if line.strip() and not line.startswith("PERSONALIZATION")]
return {
"personalization_hooks": hooks[:3],
"research_notes": f"\n--- PERSONALIZATION BRIEF ---\n{brief}\n",
}Step 4: Building the Drafting Nodes
With research complete, the agent crafts outreach messages. Separating email and LinkedIn message drafting into distinct nodes allows each to be optimized for its channel's conventions.
# drafting_nodes.py
from langchain_openai import ChatOpenAI
from datetime import datetime
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
def draft_email(state: SalesOutreachState) -> dict:
"""Draft a highly personalized cold email based on research."""
# Only draft email if the channel includes email
if state["outreach_channel"] not in ["email", "both"]:
return {"email_draft": None}
draft_prompt = f"""
You are an expert sales development representative writing a cold outreach email.
CONTEXT:
- Prospect: {state["prospect_name"]} at {state["company_name"]}
- Product: {state["product_description"]}
- Personalization hooks: {state.get("personalization_hooks", [])}
- Research: {state.get("research_notes", "")}
RULES FOR THE EMAIL:
1. Start with a GENUINE, specific compliment or observation about the prospect (use the hooks)
2. Keep total length under 120 words
3. Make ONE clear connection between the prospect's world and your product
4. End with a low-friction call to action (e.g., "Would you be open to a 15-minute chat?")
5. Avoid: generic flattery, AI-sounding phrases, excessive enthusiasm
6. Subject line should be lowercase, curiosity-driven, under 8 words
Format:
SUBJECT: [subject line]
[email body]
SIGNATURE: [your name / company]
"""
email = llm.invoke(draft_prompt).content
return {
"email_draft": email,
"iteration_count": state.get("iteration_count", 0) + 1,
}
def draft_linkedin_message(state: SalesOutreachState) -> dict:
"""Draft a LinkedIn connection request or InMail based on research."""
if state["outreach_channel"] not in ["linkedin", "both"]:
return {"linkedin_message_draft": None}
draft_prompt = f"""
You are crafting a LinkedIn outreach message for a sales context.
CONTEXT:
- Prospect: {state["prospect_name"]} at {state["company_name"]}
- Product: {state["product_description"]}
- Personalization hooks: {state.get("personalization_hooks", [])}
RULES FOR LINKEDIN MESSAGE:
1. Maximum 300 characters (LinkedIn connection note limit)
2. Mention something specific about their work or content they've shared
3. State your reason for connecting authentically (not "I'm expanding my network")
4. Do NOT pitch the product directly—this is about starting a relationship
5. If it's an InMail, you can include a soft value proposition
Write the message directly (no labels or formatting):
"""
linkedin_msg = llm.invoke(draft_prompt).content
return {
"linkedin_message_draft": linkedin_msg.strip(),
}Step 5: Adding the Human-in-the-Loop Review Node
One of LangGraph's most powerful features is the ability to interrupt execution and wait for human input. This is perfect for sales outreach, where a human should review and approve messages before they're sent.
# review_node.py
def human_review(state: SalesOutreachState) -> dict:
"""
This node acts as a checkpoint for human review.
The graph will PAUSE here and wait for external input
before proceeding to the send nodes.
In a production system, this would trigger a notification
(Slack, email, UI) and await approval with optional feedback.
"""
# This node doesn't modify state directly.
# Instead, LangGraph's interrupt mechanism pauses execution.
# The human provides updated state fields via the graph's
# update_state() method.
# We just return an empty dict—the actual review happens
# through LangGraph's external update mechanism.
return {}
def apply_revision_feedback(state: SalesOutreachState) -> dict:
"""Apply human revision feedback to re-draft messages."""
feedback = state.get("revision_feedback", "")
if not feedback:
return {}
revision_prompt = f"""
Revise the following outreach drafts based on this feedback:
FEEDBACK: {feedback}
ORIGINAL EMAIL DRAFT:
{state.get('email_draft', 'N/A')}
ORIGINAL LINKEDIN DRAFT:
{state.get('linkedin_message_draft', 'N/A')}
Return the revised versions in the same format as the originals.
Keep all the original personalization elements but incorporate the changes requested.
REVISED EMAIL:
[revised email]
REVISED LINKEDIN:
[revised linkedin message]
"""
revised = llm.invoke(revision_prompt).content
# Parse the revised drafts
email_draft = None
linkedin_draft = None
if "REVISED EMAIL:" in revised and "REVISED LINKEDIN:" in revised:
parts = revised.split("REVISED LINKEDIN:")
email_part = parts[0].replace("REVISED EMAIL:", "").strip()
linkedin_part = parts[1].strip()
email_draft = email_part
linkedin_draft = linkedin_part
elif "REVISED EMAIL:" in revised:
email_draft = revised.split("REVISED EMAIL:")[1].strip()
return {
"email_draft": email_draft or state.get("email_draft"),
"linkedin_message_draft": linkedin_draft or state.get("linkedin_message_draft"),
"revision_feedback": None, # Clear feedback after applying
"human_approved": True, # Mark as approved after revision
}Step 6: Building the Execution Nodes
These nodes handle the actual sending of emails and LinkedIn messages. In a production system, you'd integrate with SendGrid, a LinkedIn automation tool, or your CRM.
# execution_nodes.py
import os
from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail
def send_email(state: SalesOutreachState) -> dict:
"""Send the approved email via SendGrid."""
if state.get("outreach_channel") not in ["email", "both"]:
return {"send_status": "skipped_email"}
if not state.get("human_approved", False):
return {"send_status": "not_approved_for_sending"}
email_draft = state.get("email_draft", "")
prospect_email = state.get("prospect_email")
if not prospect_email:
return {"send_status": "no_email_address"}
# Parse subject and body from draft
subject = "Connecting"
body = email_draft
if "SUBJECT:" in email_draft:
lines = email_draft.split("\n")
subject = lines[0].replace("SUBJECT:", "").strip()
body = "\n".join(lines[1:]).strip()
try:
sg = SendGridAPIClient(api_key=os.getenv("SENDGRID_API_KEY"))
message = Mail(
from_email='sales@yourcompany.com',
to_emails=prospect_email,
subject=subject,
plain_text_content=body
)
response = sg.send(message)
if response.status_code in [200, 201, 202]:
status = f"email_sent_successfully_{response.status_code}"
else:
status = f"email_send_failed_{response.status_code}"
except Exception as e:
status = f"email_send_error_{str(e)}"
return {
"email_sent": True if "successfully" in status else False,
"send_status": status,
}
def send_linkedin_message(state: SalesOutreachState) -> dict:
"""Send LinkedIn message via LinkedIn API or a connected tool."""
if state.get("outreach_channel") not in ["linkedin", "both"]:
return {}
if not state.get("human_approved", False):
return {}
linkedin_draft = state.get("linkedin_message_draft", "")
# In production, integrate with LinkedIn API, Expandi, or similar
# For this tutorial, we simulate the send and log the message
print(f"[LINKEDIN SIMULATION] Would send to {state['prospect_name']}:")
print(linkedin_draft)
# Simulate success
return {
"linkedin_sent": True,
}Step 7: Assembling the LangGraph Graph
Now you wire everything together into a StateGraph with conditional edges. This is where LangGraph's power shines: you define the flow logic declaratively.
# graph.py
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from typing import Literal
# Import all nodes
from state import SalesOutreachState
from research_nodes import research_company, research_prospect, synthesize_research
from drafting_nodes import draft_email, draft_linkedin_message
from review_node import human_review, apply_revision_feedback
from execution_nodes import send_email, send_linkedin_message
# Initialize the LLM for routing decisions
from langchain_openai import ChatOpenAI
router_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
def build_sales_outreach_graph():
"""Construct the full sales outreach agent graph."""
# Create the graph with our state type
workflow = StateGraph(SalesOutreachState)
# Add all nodes
workflow.add_node("research_company", research_company)
workflow.add_node("research_prospect", research_prospect)
workflow.add_node("synthesize_research", synthesize_research)
workflow.add_node("draft_email", draft_email)
workflow.add_node("draft_linkedin", draft_linkedin_message)
workflow.add_node("human_review", human_review)
workflow.add_node("apply_revision", apply_revision_feedback)
workflow.add_node("send_email", send_email)
workflow.add_node("send_linkedin", send_linkedin_message)
# Define edges
# Start: research phase (both run in parallel conceptually—but we sequence them)
workflow.add_edge(START, "research_company")
workflow.add_edge("research_company", "research_prospect")
workflow.add_edge("research_prospect", "synthesize_research")
# After research synthesis, draft messages
workflow.add_edge("synthesize_research", "draft_email")
workflow.add_edge("synthesize_research", "draft_linkedin")
# After drafting, go to human review
workflow.add_edge("draft_email", "human_review")
workflow.add_edge("draft_linkedin", "human_review")
# Conditional edge: after human review, either revise or proceed
workflow.add_conditional_edges(
"human_review",
route_after_review,
{
"revise": "apply_revision",
"approved": "send_email", # Proceed to sending
"rejected": END, # Stop if fully rejected
}
)
# After revision, loop back to drafting
workflow.add_edge("apply_revision", "draft_email")
# After sending email, optionally send LinkedIn
workflow.add_conditional_edges(
"send_email",
route_after_email_send,
{
"send_linkedin": "send_linkedin",
"skip": END,
}
)
workflow.add_edge("send_linkedin", END)
# Compile with checkpointing for human-in-the-loop
memory = MemorySaver()
compiled_graph = workflow.compile(
checkpointer=memory,
interrupt_before=["human_review"] # Pause before human review
)
return compiled_graph
def route_after_review(state: SalesOutreachState) -> Literal["revise", "approved", "rejected"]:
"""Determine next step after human review node."""
if state.get("human_approved", False):
if state.get("revision_feedback"):
return "revise"
return "approved"
return "rejected"
def route_after_email_send(state: SalesOutreachState) -> Literal["send_linkedin", "skip"]:
"""Decide whether to also send LinkedIn message."""
if state.get("outreach_channel") == "both" and not state.get("linkedin_sent"):
return "send_linkedin"
return "skip"
# Build the graph (call this in your main application)
sales_agent_graph = build_sales_outreach_graph()Step 8: Running the Agent with Human-in-the-Loop
Here's how you invoke the agent, handle the interrupt for human review, and resume execution with approval.
# main.py
import asyncio
from graph import sales_agent_graph
from state import SalesOutreachState
async def run_sales_agent():
"""Run a complete sales outreach workflow with human review."""
# Initial input configuration
initial_config = {
"prospect_name": "Sarah Chen",
"company_name": "NovaTech Solutions",
"prospect_linkedin_url": "https://linkedin.com/in/sarahchen",
"prospect_email": "sarah@novatech.com",
"product_description": "AI-powered customer support platform that reduces ticket handling time by 40%",
"outreach_channel": "email", # Start with email only
"human_approved": False,
"iteration_count": 0,
}
# Create a thread_id for checkpointing
thread_config = {"configurable": {"thread_id": "outreach-sarah-chen-001"}}
# --- FIRST INVOCATION: Runs until human_review ---
print("🚀 Starting agent execution...")
print("⏸️ Agent will pause at human_review node\n")
# This will run until it hits the interrupt_before=["human_review"] checkpoint
async for event in sales_agent_graph.astream(
initial_config,
thread_config,
stream_mode="values"
):
node_name = event.get("__metadata__", {}).get("langgraph_node", "unknown")
state_snapshot = event.get("__data__", event)
print(f"📍 Node: {node_name}")
if "email_draft" in state_snapshot and state_snapshot.get("email_draft"):
print("\n📧 DRAFT EMAIL:")
print("=" * 40)
print(state_snapshot["email_draft"])
print("=" * 40)
# --- CHECK CURRENT STATE ---
current_state = sales_agent_graph.get_state(thread_config)
print(f"\n⏸️ Agent paused at: {current_state.next}")
# --- SIMULATE HUMAN REVIEW ---
# In production, you'd send this draft to a UI/Slack/email for review
print("\n👤 HUMAN REVIEW REQUIRED")
print("Review the draft above and choose an action:")
print("1. Approve as-is")
print("2. Approve with revisions")
print("3. Reject entirely")
# Simulate human choosing option 1 (approve)
human_decision = "approve" # Change this to test different paths
if human_decision == "approve":
# Update state with human approval and resume
sales_agent_graph.update_state(
thread_config,
{"human_approved": True}
)
print("\n✅ Approved! Resuming execution...")
# Resume from the interrupt point
async for event in sales_agent_graph.astream(
None, # No new input—just resume with updated state
thread_config,
stream_mode="values"
):
node_name = event.get("__metadata__", {}).get("langgraph_node", "unknown")
state_snapshot = event.get("__data__", event)
print(f"📍 Node: {node_name}")
if state_snapshot.get("send_status"):
print(f"📨 Send status: {state_snapshot['send_status']}")
elif human_decision == "revise":
# Provide feedback and resume
feedback = "Make the tone more casual and mention their recent podcast appearance"
sales_agent_graph.update_state(
thread_config,
{
"human_approved": True,
"revision_feedback": feedback
}
)
# Resume—will loop through revision and re-draft
# ... (similar async event loop as above)
elif human_decision == "reject":
sales_agent_graph.update_state(
thread_config,
{"human_approved": False, "final_outcome": "rejected_by_human"}
)
print("❌ Outreach rejected. Ending workflow.")
# --- FINAL STATE ---
final_state = sales_agent_graph.get_state(thread_config)
print("\n📊 FINAL OUTCOME:")
print(f"Email sent: {final_state.values.get('email_sent', False)}")
print(f"LinkedIn sent: {final_state.values.get('linkedin_sent', False)}")
print(f"Final status: {final_state.values.get('send_status', 'unknown')}")
print(f"Research notes length: {len(final_state.values.get('research_notes', ''))} chars")
# Run the agent
if __name__ == "__main__":
asyncio.run(run_sales_agent())Step 9: Adding Multi-Channel Conditional Logic
One of the most valuable patterns in LangGraph is dynamic branching based on state. Here's an enhanced router that decides which research nodes to activate based on available data.
# enhanced_router.py
from typing import Literal
from state import SalesOutreachState
def route_research_depth(state: SalesOutreachState) -> Literal["deep_research", "quick_research", "skip_research"]:
"""
Decide research depth based on what we already know.
- If we have LinkedIn URL AND company name → deep research
- If we only have company name → quick research
- If we have neither → skip research, use provided info
"""
has_linkedin = bool(state.get("prospect_linkedin_url"))
has_company = bool(state.get("company_name"))
if has_linkedin and has_company:
return "deep_research"
elif has_company:
return "quick_research"
else:
return "skip_research"
def route_outreach_channel(state: SalesOutreachState) -> str:
"""
Dynamically choose the best outreach channel based on research findings.
"""
channel = state.get("outreach_channel", "email")
# If research reveals the prospect is very active on LinkedIn, prefer that
research = state.get("research_notes", "").lower()
if "linkedin" in research and "active" in research and channel == "email":
# Override to both channels
return "both_channels"
return channel
def route_after_send(state: SalesOutreachState) -> Literal["send_linkedin", "log_to_crm", "schedule_followup", "end"]:
"""
After sending the primary message, decide on next actions:
- Send LinkedIn if channel is 'both'
- Log to CRM always
- Schedule follow-up if no reply buffer exists
"""
steps = []
if state.get("outreach_channel") == "both" and not state.get("linkedin_sent"):
steps.append("send_linkedin")
if not state.get("crm_logged", False):
steps.append("log_to_crm")
if steps:
return steps[0] # Return first pending step
return "end"Step 10: Building a Follow-Up Scheduler Node
A complete sales agent doesn't stop at the first message. It schedules intelligent follow-ups based on whether the prospect engaged.
# followup_node.py
from datetime import datetime, timedelta
def schedule_followup(state: SalesOutreachState) -> dict:
"""
Schedule a follow-up message based on engagement signals.
Logic:
- If no response after 3 days → gentle follow-up
- If prospect opened email (tracked via pixel) but didn't reply → reference their interest
- If they replied → hand off to human
"""
send_status = state.get("send_status", "")
email_sent = state.get("email_sent", False)
if not email_sent or "failed" in send_status.lower():
return {"final_outcome": "no_followup_needed_send_failed"}
followup_prompt = f"""
Draft a brief follow-up email for this prospect.
Original email sent:
{state.get('email_draft', '')}
Research context:
{state.get('research_notes', '')[:1000]}
Rules for follow-up:
1. Reference the original email briefly ("Following up on my note last week...")
2. Add ONE new piece of value (a relevant article, insight, or data point)
3. Keep it under 80 words
4. Make it easy to say yes or no
5. Do NOT apologize for "following up" or "taking time"
Schedule this for { (datetime.now() + timedelta(days=3)).strftime('%A, %B %d') }
FOLLOW-UP EMAIL:
"""
followup_draft = llm.invoke(followup_prompt).content
# In production: actually schedule via your calendar/CRM API
scheduled_date = (datetime.now() + timedelta(days=3)).isoformat()
return {
"final_outcome": f"followup_scheduled_for_{scheduled_date}",
"followup_draft": followup_draft,
"followup_scheduled_date": scheduled_date,
}Complete Graph with All Advanced Features
Here's the final graph assembly incorporating all the advanced patterns: dynamic routing, revision loops, multi-channel execution, and follow-up scheduling.
# complete_graph.py
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from state import SalesOutreachState
from enhanced_router import route_research_depth, route_after_send
def build_complete_sales_graph():
"""Build the production-ready sales outreach agent graph."""
workflow = StateGraph(SalesOutreachState)
# --- NODE REGISTRATION ---
workflow.add_node("deep_research_company", research_company)
workflow.add_node("deep_research_prospect", research_prospect)
workflow.add_node("quick_research", quick_company_research) # Simplified version
workflow.add_node("synthesize_research", synthesize_research)
workflow.add_node("draft_email", draft_email)
workflow.add_node("draft_linkedin", draft_linkedin_message)
workflow.add_node("human_review", human_review)
workflow.add_node("apply_revision", apply_revision_feedback)
workflow.add_node("send_email", send_email)
workflow.add_node("send_linkedin", send_linkedin_message)
workflow.add_node("log_to_crm", log_outreach_to_crm)
workflow.add_node("schedule_followup", schedule_followup)
# --- EDGE DEFINITION ---
# Start with conditional research routing
workflow.add_conditional_edges(
START,
route_research_depth,
{
"deep_research": "deep_research_company",
"quick_research": "quick_research",
"skip_research": "draft_email", # Skip directly to drafting
}
)
# Deep research path
workflow.add_edge("deep_research_company", "deep_research_prospect")
workflow.add_edge("deep_research_prospect", "synthesize_research")
# Quick research path