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 techniques can do all three, with velocity and consistency that no human crew can match. 

This information will present you precisely find out how 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 accomplished by gross sales reps or SDRs. A multi-agent system assigns every job to a devoted agent as a substitute of a single mannequin doing all of the work. One agent appears into. Another qualifies. A 3rd information the consequence. Collectively, they make for a sturdy, repeatable pipeline. 

Why Gross sales Groups Want AI Automation At this time

At this time’s B2B gross sales cycles are information wealthy. A rep might be spending 30-40% of their time on non-selling actions akin to researching LinkedIn profiles, scoring leads towards ICP standards, or updating Salesforce. That point has a direct price. 

AI automation addresses this in 3 ways: 

• Pace: Prospecting takes seconds, not minutes. 
• Consistency: Consistency Every lead is judged by the identical requirements, eliminating the potential of human bias. 
• Scale: The pipeline that handles 10 prospects handles 10,000 with no extra 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 complete gross sales analysis to CRM pipeline. Every job akin to analysis, qualification, and information entry has completely different logic, completely different failure modes, and completely different information necessities. 

Multi-agent techniques remedy 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 Excellent Buyer Profile (ICP). 
• Agent 3: will do the formatting and writes the consequence 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 purposes on prime of LangChain. It fashions your workflow as a directed graph, the place the nodes are brokers or capabilities, and the perimeters describe how the state flows between them. 

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

Why LangGraph Is Excellent for Gross sales Workflows 

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

• Conditional routing:  If the lead shouldn’t be 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 velocity. 

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

Key LangGraph Ideas You Must Know 

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

• State: A TypedDict describing all the info your pipeline retains observe of. Consider it as a communal scratchpad that every 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 at all times 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 closing object which can 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 could 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 assemble 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 unhealthy, or at the very least irrelevant, information downstream 

System Structure Overview 

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

Setting Up Your Setting

Getting the dependencies in place the appropriate manner is the very first step. This strategy leans on LangGraph, LangChain, and the OpenAI consumer, which is smart I suppose. Additionally you’ll usher in 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 must be 3.9 or greater. For the correct execution of LangGraph, TypeDict, and Trace help. After which run the next command in your terminal: 

pip set up langgraph langchain-openai langchain python-dotenv 

Venture Construction 

You’ll be able to organise your working listing like proven beneath for correct readability and maintainability. 

Setting Configuration 

Create a .env file in your challenge root: 

OPENAI_API_KEY=your_openai_api_key_here 

Observe: Please ensure to not laborious code the api keys within the supply information. All the time load them from the surroundings variables. 

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

Now we’ll begin to construct every element, type of, and maintain 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 complete system is constructed on. Each agent reads from and writes to this object, no exceptions actually. 

graph/state.py 

from typing import (
    TypedDict,
    Elective,
    Record,
    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: Elective[str]

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

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

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

    crm_updated: Elective[bool]

    # --- Observability ---
    pipeline_messages: Record[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 online 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 may drop in actual instrument 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 alerts,
    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 professional B2B gross sales researcher.

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

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

    Return this actual construction:

     series_b 
    """

    user_message = f"""
    Analysis this prospect:

    Title:
    {state['prospect_name']}

    Firm:
    {state['company']}

    Function:
    {state['role']}

    E mail:
    {state['email']}

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

    Return structured JSON analysis information.
    """

    strive:

        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"Function 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 Excellent Buyer Profile (ICP). It makes use of the structured analysis information from Agent 1, then apply a scoring rubric kind 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

- Function shopping for energy: medium or excessive
- Function 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:
  Sturdy ICP match — prioritize instantly

- 60–79:
  Good match — value 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 obtainable. "
            "Marking as unqualified."
        )

        return {
            **state,
            "qualification_score": 0,
            "qualification_reason": (
                "Analysis information was unavailable. "
                "Can not 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 actual construction:

    {{
        "rating": ,

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

        "score_breakdown": {{
            "company_size_fit": ,
            "industry_fit": ,
            "role_seniority_fit": ,
            "buying_power_fit": ,
            "pain_point_alignment": 
        }},

        "qualification_reason":
            "",

        "recommended_action":
            ""
    }}
    """

    user_message = f"""
    Qualify this prospect:

    Prospect:
    {state['prospect_name']}

    Firm:
    {state['company']}

    Function:
    {state['role']}

    Analysis Knowledge:

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

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

    strive:

        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 because 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; kind of a drop in substitute sample so it’s simple 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 along 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:

            strive:
                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 customized 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 customized 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": {
            "title": "",
            "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:

    Title: {state['prospect_name']}
    Firm: {state['company']}
    Function: {state['role']}
    E mail: {state['email']}
    LinkedIn: {state.get('linkedin_url', 'N/A')}

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

    Qualification Motive:
    {state['qualification_reason']}

    Analysis Knowledge:

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

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

    strive:

        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] Document 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 you then 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
        }
    )

    # Last 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 any case these elements are prepared, you wire all the things up inside predominant.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 Important Entry Level 

predominant.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:
        Last 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 Document ID: "
            f"{final_state['crm_record'].get('id')}"
        )

    return final_state


def run_batch_pipeline(prospects: checklist) -> checklist:
    """
    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)}..."
        )

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

    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"
    }

    consequence = 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 predominant.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 value noting on this output. Sarah Chen scores 88/100 and will get a completely personalised e mail draft that references her Sequence 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 completely 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 fully, no CRM write in any respect, no wasted API name, no unhealthy 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 whole lead particulars, the qualification reasoning, and the outreach draft so a rep can take motion straight away. 

Ideas For Manufacturing Concerns

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

Including Actual Instrument Calls to the Analysis Agent 

Proper now the analysis agent makes use of the LLM skilled data as a type of substitute to simulate analysis. For manufacturing, it’s best to give it precise devices, not simply realized heuristics. With LangChain this turns into fairly direct, for instance through the use of the @instrument decorator: 

from langchain_core.instruments import instrument
import requests


@instrument
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 instrument to your analysis agent’s LLM with bind_tools([search_company_apollo]) and modify 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 in-built help 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):
                strive:
                    return func(*args, **kwargs)

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

                    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 prime-volume pipelines the place you’re shifting by way of a whole bunch of prospects, run them concurrently, utilizing asyncio, it kind of helps maintain 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):
                strive:
                    return func(*args, **kwargs)

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

                    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 along with your actual CRM consumer, 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:

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

    contact = consumer.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 surroundings 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 useful 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 unhealthy 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 reside integrations for manufacturing use. With minimal adjustments, this workflow can scale to a whole bunch of prospects, permitting gross sales groups to spend extra time on conversations as a substitute of prospect analysis and admin work. 

Hiya! I am Vipin, a passionate information science and machine studying fanatic with a powerful 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 aim is to use data-driven insights to create sensible options that drive outcomes. I am desirous to contribute my expertise in a collaborative surroundings whereas persevering with to study and develop within the fields of Knowledge Science, Machine Studying, and NLP.