Construct a Gross sales AI Workflow: Automate Analysis with LangGraph

Gross sales groups spend hours day by day on duties that ought to by no means see a human. Analysis a prospect, rating them towards their match, and put all of it right into a CRM. These are repeatable, rule based mostly processes AI workflows pushed by multi-agent methods can do all three, with pace and consistency that no human group can match. 

This information will present you precisely methods to construct that system. You’ll use LangGraph to orchestrate a number of AI brokers, powered by OpenAI’s state-of-the-art LLMs, to construct a gross sales pipeline that takes uncooked prospect information and turns it into a completely up to date CRM entry, with zero guide effort. Let’s do it!

What Are AI Workflows for Gross sales Groups?

An AI workflow is an automatic collection of AI-enabled steps that performs jobs historically achieved by gross sales reps or SDRs. A multi-agent system assigns every job to a devoted agent as an alternative of a single mannequin doing all of the work. One agent seems into. Another qualifies. A 3rd information the outcome. Collectively, they make for a sturdy, repeatable pipeline. 

Why Gross sales Groups Want AI Automation Right this moment

Right this moment’s B2B gross sales cycles are information wealthy. A rep could possibly be spending 30-40% of their time on non-selling actions reminiscent of researching LinkedIn profiles, scoring leads towards ICP standards, or updating Salesforce. That point has a direct value. 

AI automation addresses this in 3 ways: 

• Velocity: Prospecting takes seconds, not minutes. 
• Consistency: Consistency Every lead is judged by the identical requirements, eliminating the potential for human bias. 
• Scale: The pipeline that handles 10 prospects handles 10,000 with no further headcount. 

Automating gross sales workflows doesn’t imply eliminating salespeople. It’s about returning their time to do what solely people can do, construct relationships and shut offers. 

The Multi-Agent Method to Gross sales Automation 

A single LLM immediate isn’t strong sufficient to cowl your entire gross sales analysis to CRM pipeline. Every job reminiscent of analysis, qualification, and information entry has totally different logic, totally different failure modes, and totally different information necessities. 

Multi-agent methods clear up this by breaking the issue into centered sub-tasks: 

• Agent 1: Agent 1 is solely involved with gathering structured analysis information on a prospect. 
• Agent 2: will care about scoring that information towards your Ultimate Buyer Profile (ICP). 
• Agent 3: will do the formatting and writes the outcome again to your CRM. 

LangGraph handles the connections between these brokers, how each runs, and the way information flows between them. The structure is extra dependable, simpler to debug and far easier to increase than a single monolithic immediate. 

Understanding LangGraph for Gross sales Automation

LangGraph is a framework for constructing stateful multi-step AI functions on high of LangChain. It fashions your workflow as a directed graph, the place the nodes are brokers or capabilities, and the sides describe how the state flows between them. 

This design makes LangGraph particularly highly effective for gross sales workflows the place the subsequent step usually is dependent upon the results of the earlier one, e.g., solely working the CRM replace if the lead really qualifies. 

Why LangGraph Is Ultimate for Gross sales Workflows 

Different orchestration frameworks mannequin pipelines as linear chains. What you get from LangGraph: 

• Conditional routing:  If the lead is just not certified, skip the CRM replace. 
• Shared state:  Prospect information, scores, messages, full context for every agent. 
• Checkpointing:  Resume a failed pipeline proper the place it stopped. 
• Parallel execution: Run impartial brokers in parallel for pace. 

That management of the movement of execution is vital in a gross sales automation system, the place choices like “is that this lead price pursuing?” have a direct impact on downstream steps. 

Key LangGraph Ideas You Have to Know 

Earlier than writing any code, one ought to perceive the under 4 core ideas: 

• State: A TypedDict describing all the info your pipeline retains monitor of. Consider it as a communal scratchpad that each one brokers can learn and write from. 
• Nodes: Nodes are principally the python capabilities that take the present state and return a brand new state. 
• Edges: Edges are the hyperlinks between the nodes. They are often mounted like all the time go to Node B after visiting Node A or conditional like go to Node B or Node C relying on a worth within the state. 
• Graph: Graph would be the ultimate object which is able to get executed by you. It hooks up all of the nodes and edges and exposes a invoke() perform. 

Architecting the Multi-Agent Gross sales System

This technique can have 3agents working so as with some conditional branching, kinda like if this then that however not precisely…   

If the analysis agent finally ends up failing to collect information, the pipeline simply bails out early. If the qualification agent then marks a lead as disqualified, the CRM replace doesn’t occur in any respect, so no notes or fields get written. This conditional logic retains issues from going messy later, so that you by no means find yourself with dangerous, or no less than irrelevant, information downstream 

System Structure Overview 

The shared state object is basically the spine of this entire setup. Every agent reads from that very same object, after which writes again to it too. That is what your entire state seems like throughout the lifecycle of 1 single prospect: 

Setting Up Your Setting

Getting the dependencies in place the appropriate method is the very first step. This strategy leans on LangGraph, LangChain, and the OpenAI shopper, which is smart I assume. Additionally you’ll herald python-dotenv, simply to assist handle your API key safely and with out leaking it someplace. 

Putting in Dependencies

Earlier than putting in the dependencies please be sure that your present python model needs to be 3.9 or increased. For the right execution of LangGraph, TypeDict, and Trace assist. After which run the next command in your terminal: 

pip set up langgraph langchain-openai langchain python-dotenv 

Venture Construction 

You may organise your working listing like proven under for correct readability and maintainability. 

Setting Configuration 

Create a .env file in your challenge root: 

OPENAI_API_KEY=your_openai_api_key_here 

Be aware: Please be certain to not onerous code the api keys within the supply information. All the time load them from the atmosphere variables. 

Constructing the Multi-Agent System: Step-by-Step

Now we’ll begin to construct every part, sort of, and hold it constant. Begin with the state schema, after which construct each agent, afterwards you wire them collectively in a LangGraph workflow. 

Step 1: Outline the Shared State Schema

The state schema is the contract your entire system is constructed on. Each agent reads from and writes to this object, no exceptions actually. 

graph/state.py 

from typing import (
    TypedDict,
    Non-compulsory,
    Checklist,
    Dict,
    Any
)

class SalesState(TypedDict):
    """
    Shared state throughout all brokers
    within the gross sales workflow.

    Each agent reads this in full
    and returns an up to date model.
    """

    # --- Enter fields ---
    # Set at pipeline begin
    prospect_name: str
    firm: str
    function: str
    e mail: str
    linkedin_url: Non-compulsory[str]

    # --- Analysis Agent outputs ---
    research_data: Non-compulsory[
        Dict[str, Any]
    ]

    # --- Qualification Agent outputs ---
    qualification_score: Non-compulsory[int]
    qualification_reason: Non-compulsory[str]
    is_qualified: Non-compulsory[bool]

    # --- CRM Agent outputs ---
    crm_record: Non-compulsory[
        Dict[str, Any]
    ]

    crm_updated: Non-compulsory[bool]

    # --- Observability ---
    pipeline_messages: Checklist[str]

Step 2: Construct the Prospect Analysis Agent 

The analysis agent takes uncooked prospect data after which enriches it. In a manufacturing setting, this agent would name exterior instruments like Apollo.io, Clearbit, or an internet search API. Right here, you’ll use gpt-4.1-mini to simulate “clever analysis” based mostly on the inputs you present. The entire structure is designed so you possibly can drop in actual software calls later with minimal adjustments, and also you don’t have to transform all the things. 

brokers/research_agent.py 

import json

from langchain_openai import ChatOpenAI
from langchain_core.messages import (
    SystemMessage,
    HumanMessage
)

from graph.state import SalesState


llm = ChatOpenAI(
    mannequin="gpt-4.1-mini",
    temperature=0
)


def prospect_research_agent(
    state: SalesState
) -> SalesState:
    """
    Agent 1: Prospect Analysis

    Takes fundamental prospect data and
    returns structured analysis information
    together with firm context,
    function indicators,
    and shopping for intent indicators.
    """

    print(
        f"n[Research Agent] Beginning analysis on "
        f"{state['prospect_name']} "
        f"at {state['company']}"
    )

    system_prompt = """
    You're an skilled B2B gross sales researcher.

    Given a prospect's identify,
    firm,
    and function,
    you generate structured analysis information
    {that a} gross sales rep would want
    to qualify and personalize outreach.

    You could return ONLY legitimate JSON.
    No markdown,
    no rationalization,
    simply the JSON object.

    Return this precise construction:

     medium 
    """

    user_message = f"""
    Analysis this prospect:

    Identify:
    {state['prospect_name']}

    Firm:
    {state['company']}

    Position:
    {state['role']}

    Electronic mail:
    {state['email']}

    LinkedIn:
    {state.get('linkedin_url', 'Not offered')}

    Return structured JSON analysis information.
    """

    attempt:

        response = llm.invoke([
            SystemMessage(content=system_prompt),
            HumanMessage(content=user_message)
        ])

        raw_content = response.content material.strip()

        # Strip markdown code fences
        # if the mannequin wraps output

        if raw_content.startswith("```"):

            raw_content = raw_content.break up("```")[1]

            if raw_content.startswith("json"):
                raw_content = raw_content[4:]

        research_data = json.hundreds(raw_content)

        print(
            f"[Research Agent] Analysis full. "
            f"Trade: {research_data.get('business')}, "
            f"Seniority: "
            f"{research_data.get('role_seniority')}"
        )

        return {
            **state,
            "research_data": research_data,
            "pipeline_messages": (
                state["pipeline_messages"] + [
                    f"Research Agent: Successfully researched "
                    f"{state['prospect_name']} "
                    f"at {state['company']}. "
                    f"Trade: "
                    f"{research_data.get('business')}, "
                    f"Position shopping for energy: "
                    f"{research_data.get('role_buying_power')}"
                ]
            )
        }

    besides (json.JSONDecodeError, Exception) as e:

        print(f"[Research Agent] Error: {e}")

        return {
            **state,
            "research_data": None,
            "pipeline_messages": (
                state["pipeline_messages"] + [
                    f"Research Agent: Failed to gather "
                    f"research data. Error: {str(e)}"
                ]
            )
        }

Step 3: Construct the Lead Qualification Agent 

The qualification agent scores every prospect towards your Ultimate Buyer Profile (ICP). It makes use of the structured analysis information from Agent 1, then apply a scoring rubric form of factor, and units a boolean is_qualified flag that controls if the pipeline retains going into the CRM step. 

brokers/qualification_agent.py 

import json

from langchain_openai import ChatOpenAI
from langchain_core.messages import (
    SystemMessage,
    HumanMessage
)

from graph.state import SalesState


llm = ChatOpenAI(
    mannequin="gpt-4.1-mini",
    temperature=0
)


# Outline your ICP right here
# Regulate this to match your precise goal buyer profile
ICP_CRITERIA = """

IDEAL CUSTOMER PROFILE (ICP):

- Firm measurement: 20–500 workers
- Funding stage: seed, series_a, series_b, or series_c
- Industries:
  SaaS, B2B Tech, Advertising Businesses,
  Skilled Companies

- Position shopping for energy: medium or excessive
- Position seniority:
  supervisor, director, vp, or c_suite

- Key ache factors:
  guide workflows,
  operational inefficiency,
  scaling operations,
  lack of automation,
  excessive headcount prices

SCORING GUIDE (0–100):

- 80–100:
  Robust ICP match — prioritize instantly

- 60–79:
  Good match — price pursuing
  with personalised outreach

- 40–59:
  Partial match — add to nurture sequence

- Under 40:
  Poor match — disqualify

"""


def lead_qualification_agent(
    state: SalesState
) -> SalesState:
    """
    Agent 2: Lead Qualification

    Scores the prospect towards ICP standards
    utilizing structured analysis information.

    Units is_qualified flag to regulate
    downstream CRM replace step.
    """

    print(
        f"n[Qualification Agent] "
        f"Scoring {state['prospect_name']}..."
    )

    research_data = state.get("research_data")

    if not research_data:

        print(
            "[Qualification Agent] "
            "No analysis information out there. "
            "Marking as unqualified."
        )

        return {
            **state,
            "qualification_score": 0,
            "qualification_reason": (
                "Analysis information was unavailable. "
                "Can't qualify with out information."
            ),
            "is_qualified": False,
            "pipeline_messages": (
                state["pipeline_messages"] + [
                    "Qualification Agent: "
                    "Skipped — no research data available."
                ]
            )
        }

    system_prompt = f"""
    You're a senior gross sales qualification specialist.

    You consider prospects towards a strict ICP
    and return a qualification rating with reasoning.

    {ICP_CRITERIA}

    You could return ONLY legitimate JSON
    with this precise construction:

    {{
        "rating": ,

        "is_qualified":
            = 60,
            false in any other case>,

        "score_breakdown": {{
            "company_size_fit": <0-20>,
            "industry_fit": <0-20>,
            "role_seniority_fit": <0-20>,
            "buying_power_fit": <0-20>,
            "pain_point_alignment": <0-20>
        }},

        "qualification_reason":
            "<2-3 sentence rationalization>",

        "recommended_action":
            ""
    }}
    """

    user_message = f"""
    Qualify this prospect:

    Prospect:
    {state['prospect_name']}

    Firm:
    {state['company']}

    Position:
    {state['role']}

    Analysis Information:

    {json.dumps(research_data, indent=2)}

    Rating them towards the ICP and
    return the qualification JSON.
    """

    attempt:

        response = llm.invoke([
            SystemMessage(content=system_prompt),
            HumanMessage(content=user_message)
        ])

        raw_content = response.content material.strip()

        if raw_content.startswith("```"):

            raw_content = raw_content.break up("```")[1]

            if raw_content.startswith("json"):
                raw_content = raw_content[4:]

        qualification_data = json.hundreds(raw_content)

        rating = qualification_data.get("rating", 0)

        is_qualified = qualification_data.get(
            "is_qualified",
            False
        )

        purpose = qualification_data.get(
            "qualification_reason",
            "No purpose offered."
        )

        print(
            f"[Qualification Agent] "
            f"Rating: {rating}/100 | "
            f"Certified: {is_qualified}"
        )

        return {
            **state,
            "qualification_score": rating,
            "qualification_reason": purpose,
            "is_qualified": is_qualified,
            "pipeline_messages": (
                state["pipeline_messages"] + [
                    f"Qualification Agent: "
                    f"{state['prospect_name']} "
                    f"scored {rating}/100. "
                    f"Certified: {is_qualified}. "
                    f"Motion: "
                    f"{qualification_data.get('recommended_action')}. "
                    f"Motive: {purpose}"
                ]
            )
        }

    besides (json.JSONDecodeError, Exception) as e:

        print(f"[Qualification Agent] Error: {e}")

        return {
            **state,
            "qualification_score": 0,
            "qualification_reason": (
                f"Qualification failed on account of error: "
                f"{str(e)}"
            ),
            "is_qualified": False,
            "pipeline_messages": (
                state["pipeline_messages"] + [
                    f"Qualification Agent: "
                    f"Failed with error — {str(e)}"
                ]
            )
        }

Step 4: Construct the Mock CRM Utility

In manufacturing, this module calls your precise CRM’s API. For this information, you implement a mock that writes information into an area JSON file i.e; form of a drop in substitute sample so it’s straightforward to swap it with an actual API name later, with out an excessive amount of fuss. 

utils/crm_mock.py 

import json
import os
from datetime import datetime

CRM_FILE = "crm_records.json"

def write_to_crm(file: dict) -> dict:
    """
    Mock CRM write perform.

    In manufacturing, change this together with your CRM SDK name.

    HubSpot instance:
        hubspot_client.crm.contacts.basic_api.create(
            SimplePublicObjectInput(properties=file)
        )

    Salesforce instance:
        sf.Contact.create(file)
    """

    # Load present information
    information = []

    if os.path.exists(CRM_FILE):

        with open(CRM_FILE, "r") as f:

            attempt:
                information = json.load(f)

            besides json.JSONDecodeError:
                information = []

    # Add new file with metadata
    crm_record = {
        "id": f"LEAD-{len(information) + 1:04d}",
        "created_at": datetime.utcnow().isoformat(),
        "standing": "certified",
        **file
    }

    information.append(crm_record)

    # Write again
    with open(CRM_FILE, "w") as f:
        json.dump(information, f, indent=2)

    print(
        f"[CRM] Written file with ID: "
        f"{crm_record['id']}"
    )

    return crm_record

Step 5: Construct the CRM Replace Agent

The CRM agent solely runs when a lead is certified. It takes the complete prospect data and turns it right into a tidy CRM entry, then it writes it utilizing the utility above. Additionally it produces a personalised first contact e mail draft as a part of that very same file. 

brokers/crm_agent.py 

import json

from langchain_openai import ChatOpenAI
from langchain_core.messages import (
    SystemMessage,
    HumanMessage
)

from graph.state import SalesState
from utils.crm_mock import write_to_crm


llm = ChatOpenAI(
    mannequin="gpt-4.1-mini",
    temperature=0.3
)


def crm_update_agent(state: SalesState) -> SalesState:
    """
    Agent 3: CRM Replace

    Codecs certified prospect information right into a structured CRM file.
    Additionally generates a personalised first-touch e mail draft.

    Solely runs when is_qualified is True.
    """

    print(
        f"n[CRM Agent] Creating CRM file "
        f"for {state['prospect_name']}..."
    )

    system_prompt = """
    You're a CRM specialist.

    Given prospect analysis and qualification information,
    you create a clear, structured CRM file and a
    personalised first-touch e mail.

    Return ONLY legitimate JSON with this construction:

    {
        "contact": {
            "identify": "",
            "e mail": "",
            "firm": "",
            "function": "",
            "linkedin_url": ""
        },
        "lead_details": {
            "business": "",
            "company_size": "",
            "funding_stage": "",
            "qualification_score": ,
            "qualification_reason": "",
            "pain_points": ["", ""],
            "tech_stack": ["", ""]
        },
        "outreach": {
            "first_touch_email_subject": "",
            "first_touch_email_body":
                "",
            "personalization_notes":
                ""
        },
        "next_steps": ""
    }
    """

    user_message = f"""
    Create a CRM file for this certified lead:

    Identify: {state['prospect_name']}
    Firm: {state['company']}
    Position: {state['role']}
    Electronic mail: {state['email']}
    LinkedIn: {state.get('linkedin_url', 'N/A')}

    Qualification Rating:
    {state['qualification_score']}/100

    Qualification Motive:
    {state['qualification_reason']}

    Analysis Information:

    {json.dumps(state.get('research_data', {}), indent=2)}

    Generate a structured CRM file with a
    personalised outreach e mail.
    """

    attempt:

        response = llm.invoke([
            SystemMessage(content=system_prompt),
            HumanMessage(content=user_message)
        ])

        raw_content = response.content material.strip()

        if raw_content.startswith("```"):
            raw_content = raw_content.break up("```")[1]

            if raw_content.startswith("json"):
                raw_content = raw_content[4:]

        crm_data = json.hundreds(raw_content)

        # Write to CRM
        crm_record = write_to_crm(crm_data)

        print(
            f"[CRM Agent] Report created efficiently. "
            f"ID: {crm_record.get('id')}"
        )

        return {
            **state,
            "crm_record": crm_record,
            "crm_updated": True,
            "pipeline_messages": (
                state["pipeline_messages"] + [
                    f"CRM Agent: Successfully created "
                    f"CRM record {crm_record.get('id')} "
                    f"for {state['prospect_name']}. "
                    f"First-touch e mail draft generated."
                ]
            )
        }

    besides (json.JSONDecodeError, Exception) as e:

        print(f"[CRM Agent] Error: {e}")

        return {
            **state,
            "crm_record": None,
            "crm_updated": False,
            "pipeline_messages": (
                state["pipeline_messages"] + [
                    f"CRM Agent: Failed to create CRM "
                    f"record. Error: {str(e)}"
                ]
            )
        }

Step 6: Wire the Graph Collectively

That is kinda the place LangGraph is available in. You outline the nodes, set the conditional edges, and then you definitely compile the graph. The routing logic reads the state values to determine which node runs subsequent, after which, it continues. 

graph/workflow.py 

from langgraph.graph import StateGraph, END
from graph.state import SalesState
from brokers.research_agent import prospect_research_agent
from brokers.qualification_agent import lead_qualification_agent
from brokers.crm_agent import crm_update_agent


def route_after_research(state: SalesState) -> str:
    """
    Path to qualification if analysis succeeded.
    Exit early if analysis failed.
    """

    if state.get("research_data"):
        return "qualify"

    print("[Router] Analysis failed. Exiting pipeline.")
    return END

def route_after_qualification(state: SalesState) -> str:
    """
    Path to CRM replace provided that lead is certified.
    Exit if lead is disqualified.
    """

    if state.get("is_qualified"):
        return "update_crm"

    print(
        f"[Router] Lead disqualified "
        f"(rating: {state.get('qualification_score')}). "
        f"Skipping CRM replace."
    )

    return END

def build_sales_workflow() -> StateGraph:
    """
    Construct and compile the multi-agent gross sales workflow graph.
    """

    workflow = StateGraph(SalesState)

    # Register agent nodes
    workflow.add_node(
        "analysis",
        prospect_research_agent
    )

    workflow.add_node(
        "qualify",
        lead_qualification_agent
    )

    workflow.add_node(
        "update_crm",
        crm_update_agent
    )

    # Set entry level
    workflow.set_entry_point("analysis")

    # Add conditional edges
    workflow.add_conditional_edges(
        "analysis",
        route_after_research,
        {
            "qualify": "qualify",
            END: END
        }
    )

    workflow.add_conditional_edges(
        "qualify",
        route_after_qualification,
        {
            "update_crm": "update_crm",
            END: END
        }
    )

    # Closing edge
    workflow.add_edge("update_crm", END)
    return workflow.compile()

# Export compiled graph
sales_pipeline = build_sales_workflow()

Working the Pipeline Finish-to-Finish

In spite of everything these parts are prepared, you wire all the things up inside important.py. It’s principally the beginning level the place the pipeline kicks in, you absorb prospect enter after which it runs, like finish to finish with out a lot delay. 

The Predominant Entry Level 

important.py 

import os
import json

from dotenv import load_dotenv
from graph.workflow import sales_pipeline

load_dotenv()


def run_sales_pipeline(prospect: dict) -> dict:
    """
    Run the complete gross sales AI workflow for a single prospect.

    Args:
        prospect: Dict with keys:
            prospect_name,
            firm,
            function,
            e mail,
            linkedin_url (non-compulsory)

    Returns:
        Closing pipeline state with all agent outputs.
    """

    # Initialize state with empty pipeline messages
    initial_state = {
        "prospect_name": prospect["prospect_name"],
        "firm": prospect["company"],
        "function": prospect["role"],
        "e mail": prospect["email"],
        "linkedin_url": prospect.get("linkedin_url"),
        "research_data": None,
        "qualification_score": None,
        "qualification_reason": None,
        "is_qualified": None,
        "crm_record": None,
        "crm_updated": None,
        "pipeline_messages": []
    }

    print(f"n{'=' * 60}")
    print(
        f"RUNNING SALES PIPELINE: "
        f"{prospect['prospect_name']} @ {prospect['company']}"
    )
    print(f"{'=' * 60}")

    # Run the graph
    final_state = sales_pipeline.invoke(initial_state)

    # Print pipeline abstract
    print(f"n{'=' * 60}")
    print("PIPELINE SUMMARY")
    print(f"{'=' * 60}")

    for msg in final_state["pipeline_messages"]:
        print(f"  • {msg}")

    print(f"nFinal Outcome:")
    print(f"  Certified: {final_state.get('is_qualified')}")
    print(f"  Rating: {final_state.get('qualification_score')}/100")
    print(f"  CRM Up to date: {final_state.get('crm_updated')}")

    if final_state.get("crm_record"):
        print(
            f"  CRM Report ID: "
            f"{final_state['crm_record'].get('id')}"
        )

    return final_state


def run_batch_pipeline(prospects: listing) -> listing:
    """
    Run the pipeline for a number of prospects sequentially.

    For parallel execution, use asyncio or a thread pool.
    """

    outcomes = []

    for i, prospect in enumerate(prospects):
        print(
            f"nProcessing prospect "
            f"{i + 1}/{len(prospects)}..."
        )

        outcome = run_sales_pipeline(prospect)
        outcomes.append(outcome)

    return outcomes


if __name__ == "__main__":

    # --- Single prospect instance ---
    sample_prospect = {
        "prospect_name": "Sarah Chen",
        "firm": "Growthly",
        "function": "VP of Operations",
        "e mail": "[email protected]",
        "linkedin_url": "https://linkedin.com/in/sarahchen"
    }

    outcome = run_sales_pipeline(sample_prospect)

    # --- Batch instance ---
    batch_prospects = [
        {
            "prospect_name": "Marcus Rivera",
            "company": "DataLayer",
            "role": "Head of Sales",
            "email": "[email protected]",
            "linkedin_url": None
        },
        {
            "prospect_name": "Priya Nair",
            "firm": "Loopform",
            "function": "Founder & CEO",
            "e mail": "[email protected]",
            "linkedin_url": "https://linkedin.com/in/priyanair"
        }
    ]

    batch_results = run_batch_pipeline(batch_prospects)

    print(
        f"nBatch full. "
        f"Processed {len(batch_results)} prospects."
    )

    certified = [
        r for r in batch_results
        if r.get("is_qualified")
    ]

    print(
        f"Certified: "
        f"{len(certified)}/{len(batch_results)}"
    )

Full Pipeline Output

Under is the precise terminal output from working python important.py towards three prospects. Two that qualify and one that doesn’t, Concentrate how the conditional router fires for the third prospect, and it exits cleanly earlier than the CRM step, with no errors in any respect. 

A couple of issues price noting on this output. Sarah Chen scores 88/100 and will get a completely personalised e mail draft that references her Collection B announcement, and the prevailing Zapier utilization context the analysis agent surfaced, kinda clear. Marcus Rivera scores 74/100 and qualifies, however the outreach takes a barely totally different angle: extra about constructing gross sales infrastructure from scratch and fewer about different baggage. Priya Nair scores 52/100 and the LangGraph router simply skips Agent 3 solely, no CRM write in any respect, no wasted API name, no dangerous information shifting downstream. 

The crm_records.json file on disk ought to find yourself with two structured information (LEAD-0001 and LEAD-0002), each containing the entire lead particulars, the qualification reasoning, and the outreach draft so a rep can take motion immediately. 

Suggestions For Manufacturing Concerns

A working prototype is one factor. A system you possibly can belief with actual gross sales information is one other. Earlier than deploying this pipeline in manufacturing, tackle these areas. 

Including Actual Instrument Calls to the Analysis Agent 

Proper now the analysis agent makes use of the LLM educated information as a sort of substitute to simulate analysis. For manufacturing, it’s best to give it precise devices, not simply discovered heuristics. With LangChain this turns into fairly direct, for instance by utilizing the @software decorator: 

from langchain_core.instruments import software
import requests


@software
def search_company_apollo(company_name: str) -> dict:
    """Search Apollo.io for firm information."""

    response = requests.get(
        "https://api.apollo.io/v1/organizations/search",
        headers={
            "x-api-key": os.getenv("APOLLO_API_KEY")
        },
        json={
            "q_organization_name": company_name
        }
    )

    return response.json()

Bind this software to your analysis agent’s LLM with bind_tools([search_company_apollo]) and regulate the agent so it will possibly deal with tool_call responses getting back from the mannequin. 

Error Dealing with and Retry Logic 

Then add retry logic for transient API hiccups, utilizing LangGraph’s inbuilt assist or a easy wrapper decorator: 

import time
from functools import wraps

def retry(max_attempts=3, delay=2):
    def decorator(func):

        @wraps(func)
        def wrapper(*args, **kwargs):
            for try in vary(max_attempts):
                attempt:
                    return func(*args, **kwargs)

                besides Exception as e:
                    if try == max_attempts - 1:
                        increase

                    print(
                        f"Try {try + 1} failed: {e}. "
                        f"Retrying in {delay}s..."
                    )

                    time.sleep(delay)

        return wrapper

    return decorator


@retry(max_attempts=3, delay=2)
def prospect_research_agent(state: SalesState) -> SalesState:
    
    # ... agent code

Parallel Batch Processing 

For top-volume pipelines the place you’re shifting by means of tons of of prospects, run them concurrently, utilizing asyncio, it form of helps hold issues flowing. 

import time
from functools import wraps

def retry(max_attempts=3, delay=2):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for try in vary(max_attempts):
                attempt:
                    return func(*args, **kwargs)

                besides Exception as e:
                    if try == max_attempts - 1:
                        increase

                    print(
                        f"Try {try + 1} failed: {e}. "
                        f"Retrying in {delay}s..."
                    )

                    time.sleep(delay)

        return wrapper

    return decorator


@retry(max_attempts=3, delay=2)
def prospect_research_agent(state: SalesState) -> SalesState:
    # ... agent code

Connecting to a Actual CRM 

Swap out the mock write_to_crm perform together with your actual CRM shopper, that’s the half you’d wire to your precise API or service. 

# HubSpot
from hubspot import HubSpot
from hubspot.crm.contacts import SimplePublicObjectInput

def write_to_crm_hubspot(file: dict) -> dict:

    shopper = HubSpot(api_key=os.getenv("HUBSPOT_API_KEY"))

    contact = shopper.crm.contacts.basic_api.create(
        SimplePublicObjectInput(
            properties={
                "firstname": file["contact"]["name"].break up()[0],
                "lastname": file["contact"]["name"].break up()[-1],
                "e mail": file["contact"]["email"],
                "firm": file["contact"]["company"],
                "jobtitle": file["contact"]["role"],
            }
        )
    )

    return {
        "id": contact.id,
        **file
    }

Including LangSmith Observability 

Additionally set these atmosphere variables so each agent run will get traced inside LangSmith: 

LANGCHAIN_TRACING_V2=true 
LANGCHAIN_API_KEY=your_langsmith_api_key 
LANGCHAIN_PROJECT=sales-ai-workflow

Every time something fires, you’ll see it in your LangSmith dashboard, with full enter/output traces for each node, which finally ends up being, genuinely invaluable for debugging and tuning prompts. 

Conclusion

Gross sales groups nonetheless doing guide analysis and lead scoring are losing worthwhile time. The multi-agent system on this information automates the workflow from uncooked prospect information to enriched CRM information and personalised outreach drafts. The structure is deliberately modular, with every agent dealing with a single duty, whereas conditional routing prevents dangerous or incomplete information from shifting additional down the pipeline.

Begin with the mock setup, refine your ICP scoring utilizing actual pipeline information, after which join stay integrations for manufacturing use. With minimal adjustments, this workflow can scale to tons of of prospects, permitting gross sales groups to spend extra time on conversations as an alternative of prospect analysis and admin work. 

Whats up! I am Vipin, a passionate information science and machine studying fanatic with a robust basis in information evaluation, machine studying algorithms, and programming. I’ve hands-on expertise in constructing fashions, managing messy information, and fixing real-world issues. My purpose is to use data-driven insights to create sensible options that drive outcomes. I am desperate to contribute my expertise in a collaborative atmosphere whereas persevering with to be taught and develop within the fields of Information Science, Machine Studying, and NLP.