Agent Orchestration with LangChain and CrewAI: From Concept to Production

AI2YOU — AI-FIRST TECHNICAL SERIES

For AI Engineers, Tech Leads, and CTOs making architecture decisions in production.

1. You've Already Built an Agent. Now You Need to Build an Orchestra.

A ReAct agent that queries an API and formats a response is a solved problem. The tutorials cover that ground well. What the official documentation rarely addresses is what happens when you have eight of those agents that need to collaborate, share state, recover from each other's failures, and produce auditable outputs in a system processing 400 requests per hour.

That is a categorically different problem.

The transition from agent to multi-agent system (MAS) is not a matter of scaling what already works. It is a complete re-architecture of the mental model. You stop thinking about "which prompt produces the best output" and start thinking about communication protocols, distributed state management, decision hierarchies, and failure recovery strategies.

The empirical evidence is stark: 73% of MAS projects fail at the integration phase — not at the proof of concept, not at the model level, but at the moment independent agents need to function as a coherent system in production (illustrative figure, consistent with distributed software engineering literature). The most common failure point is not technical in the sense of "the model hallucinated." It is architectural: state corrupted between executions, absence of deterministic retry logic, lack of observability when something goes wrong at 3 AM.

This article is a contract: by the end, you will have a practical framework for making architecture decisions between LangChain/LangGraph and CrewAI, with production-commented code, fault tolerance patterns, and a decision matrix that works for real teams. No "hello world" examples. No ROI promises without a technical basis.

2. Orchestration Fundamentals

2.1 Operational Definition

Orchestration is not chained prompt coordination. A classic LangChain chain — prompt | llm | parser — is sequential function composition. Useful, but brittle and deterministic: any failing step brings down the entire pipeline, there is no notion of shared state between calls, and no mechanism for one component to "ask for help" from another.

Orchestration is the layer that manages:

• Who executes each sub-task
• When execution occurs (dependencies, parallelism)
• What is passed between agents (interface contract)
• What to do when anything fails

The conductor analogy is precise for a specific reason: the conductor plays no instrument. They ensure the oboe enters on the correct beat, that the double bass does not drown the violin solo, and that when the trumpeter misses a note, the piece continues. In system terms: low coordination latency, high individual fault tolerance, global output coherence.

The 4 non-negotiable pillars of any MAS in production:

2.2 LangChain vs. CrewAI — Correct Positioning

The wrong question is "which is better." The right question is "which one solves the specific problem of this architecture."

LangChain/LangGraph is a low-level framework. You explicitly define every graph node, every conditional edge, every state transition. LangGraph compiles your graph into a deterministic state machine. You have total control — and total responsibility for every detail.

CrewAI is a declarative abstraction. You define business roles (Researcher, Analyst, Strategist), tasks, and a collaboration process. The framework manages the execution flow. You trade granular control for development speed and code readability.

Decision matrix:

3. Architecture with LangChain/LangGraph

3.1 Base Structure with LangGraph

The LangGraph mental model: a StateGraph is a directed graph where each node is a Python function that receives the current state and returns a state update. Edges define the flow. Conditional edges allow dynamic routing based on state.

The example below implements a document analysis system with three specialized agents:

pythoncopiar
1# langchain==0.3.x | langgraph==0.2.x | langchain-openai==0.2.x
2
3import logging
4import uuid
5from typing import TypedDict, Annotated, Literal
6from operator import add
7
8from langchain_openai import ChatOpenAI
9from langchain_core.messages import HumanMessage, SystemMessage
10from langgraph.graph import StateGraph, END
11from langgraph.checkpoint.sqlite import SqliteSaver
12
13# Structured logging — never print() in production
14logging.basicConfig(
15    level=logging.INFO,
16    format='{"time": "%(asctime)s", "level": "%(levelname)s", "msg": "%(message)s"}'
17)
18logger = logging.getLogger(__name__)
19
20
21class DocumentState(TypedDict):
22    """Shared state across all agents in the pipeline."""
23    correlation_id: str           # Unique execution ID for tracing
24    raw_content: str              # Input document
25    extracted_data: dict          # Extractor agent output
26    analysis: str                 # Analyzer agent output
27    final_report: str             # Writer agent output
28    errors: Annotated[list, add]  # Error accumulator — does not overwrite
29    retry_count: int              # Per-node retry counter
30    status: Literal["running", "completed", "failed"]
31
32
33llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
34
35
36def extractor_node(state: DocumentState) -> dict:
37    """
38    Extracts structured entities from the raw document.
39
40    Output contract: dict with keys 'entities', 'dates', 'amounts'.
41    Failures are signaled via the 'errors' field — never raise exceptions.
42    """
43    cid = state["correlation_id"]
44    logger.info(f"extractor_start correlation_id={cid}")
45
46    try:
47        response = llm.invoke([
48            SystemMessage(content=(
49                "Extract from the document: named entities, dates, and monetary values. "
50                "Return JSON with keys: entities (list), dates (list), amounts (list)."
51            )),
52            HumanMessage(content=state["raw_content"])
53        ])
54
55        import json
56        extracted = json.loads(response.content)
57        logger.info(f"extractor_done correlation_id={cid} entities={len(extracted.get('entities', []))}")
58        return {"extracted_data": extracted}
59
60    except Exception as e:
61        logger.error(f"extractor_error correlation_id={cid} error={str(e)}")
62        return {
63            "extracted_data": {},
64            "errors": [{"node": "extractor", "error": str(e), "cid": cid}]
65        }
66
67
68def analyzer_node(state: DocumentState) -> dict:
69    """
70    Analyzes extracted data and produces structured insights.
71
72    Depends on non-empty extracted_data. If empty, returns error
73    without calling the LLM — avoids unnecessary cost.
74    """
75    cid = state["correlation_id"]
76
77    if not state["extracted_data"]:
78        logger.warning(f"analyzer_skip correlation_id={cid} reason=empty_extracted_data")
79        return {
80            "analysis": "",
81            "errors": [{"node": "analyzer", "error": "extracted_data is empty", "cid": cid}]
82        }
83
84    logger.info(f"analyzer_start correlation_id={cid}")
85
86    response = llm.invoke([
87        SystemMessage(content=(
88            "Based on the extracted data, identify: "
89            "1) Relevant temporal patterns, "
90            "2) Anomalies in monetary values, "
91            "3) Relationships between entities. "
92            "Be concise and technical."
93        )),
94        HumanMessage(content=str(state["extracted_data"]))
95    ])
96
97    logger.info(f"analyzer_done correlation_id={cid}")
98    return {"analysis": response.content}
99
100
101def writer_node(state: DocumentState) -> dict:
102    """
103    Consolidates extraction and analysis into a structured executive report.
104
105    Includes a limitations section when errors have accumulated in state.
106    """
107    cid = state["correlation_id"]
108    has_errors = len(state.get("errors", [])) > 0
109
110    logger.info(f"writer_start correlation_id={cid} has_errors={has_errors}")
111
112    error_context = ""
113    if has_errors:
114        error_context = f"\n\nNOTE: {len(state['errors'])} error(s) occurred during processing. "
115        error_context += "Include a 'Limitations' section in the report."
116
117    response = llm.invoke([
118        SystemMessage(content=(
119            "Generate a structured executive report with: "
120            "Executive Summary, Key Findings, Risk Analysis, Recommendations."
121            + error_context
122        )),
123        HumanMessage(content=(
124            f"EXTRACTED DATA:\n{state['extracted_data']}\n\n"
125            f"ANALYSIS:\n{state['analysis']}"
126        ))
127    ])
128
129    logger.info(f"writer_done correlation_id={cid}")
130    return {
131        "final_report": response.content,
132        "status": "completed"
133    }
134
135
136def should_continue(state: DocumentState) -> Literal["analyzer", "writer", END]:
137    """
138    Conditional edge: decides the next node based on current state.
139
140    Logic: if extraction failed completely, skip analysis and go
141    directly to writer to generate a failure report.
142    """
143    if not state["extracted_data"] and len(state.get("errors", [])) > 0:
144        # Critical extraction failure — skip analysis, generate error report
145        return "writer"
146    return "analyzer"
147
148
149def build_document_pipeline() -> StateGraph:
150    """Compiles and returns the document processing graph."""
151    graph = StateGraph(DocumentState)
152
153    # Register nodes
154    graph.add_node("extractor", extractor_node)
155    graph.add_node("analyzer", analyzer_node)
156    graph.add_node("writer", writer_node)
157
158    # Set entry point
159    graph.set_entry_point("extractor")
160
161    # Conditional edge after extraction
162    graph.add_conditional_edges(
163        "extractor",
164        should_continue,
165        {
166            "analyzer": "analyzer",
167            "writer": "writer",
168        }
169    )
170
171    # Deterministic edges
172    graph.add_edge("analyzer", "writer")
173    graph.add_edge("writer", END)
174
175    return graph
176
177
178# Usage with checkpointer for state persistence
179def run_pipeline(document: str) -> DocumentState:
180    """
181    Executes the pipeline with state persistence via SQLite.
182
183    The thread_id allows resuming interrupted executions.
184    """
185    checkpointer = SqliteSaver.from_conn_string(":memory:")  # use real path in production
186    pipeline = build_document_pipeline().compile(checkpointer=checkpointer)
187
188    initial_state: DocumentState = {
189        "correlation_id": str(uuid.uuid4()),
190        "raw_content": document,
191        "extracted_data": {},
192        "analysis": "",
193        "final_report": "",
194        "errors": [],
195        "retry_count": 0,
196        "status": "running",
197    }
198
199    config = {"configurable": {"thread_id": initial_state["correlation_id"]}}
200    result = pipeline.invoke(initial_state, config=config)
201    return result

3.2 Orchestration Patterns with Trade-offs

Sequential — linear pipeline, each node receives the previous node's output.

pythoncopiar
1# langchain==0.3.x | langgraph==0.2.x
2# Suitable for: processes with strict ordering dependencies
3# Limitation: total latency = sum of individual latencies
4
5graph.set_entry_point("node_a")
6graph.add_edge("node_a", "node_b")
7graph.add_edge("node_b", "node_c")
8graph.add_edge("node_c", END)

Parallel (fan-out/fan-in) — multiple Workers executing simultaneously with result merging.

pythoncopiar
1# Reduces latency to: max(slowest_worker_latency)
2# Complexity: merge logic can be non-deterministic
3
4from langgraph.graph import Send
5
6def fan_out_node(state: dict) -> list[Send]:
7    """Distributes sub-tasks to parallel Workers."""
8    tasks = state["tasks"]
9    return [Send("worker_node", {"task": task, "parent_id": state["id"]})
10            for task in tasks]
11
12def merge_node(state: dict) -> dict:
13    """Consolidates results — watch for race conditions in state."""
14    return {"merged_results": state["partial_results"]}

Hierarchical — supervisor agent decides which Worker to invoke based on context.

pythoncopiar
1# Suitable for: domains where routing cannot be pre-determined
2# Limitation: the supervisor is a single point of failure and cost
3
4def supervisor_node(state: dict) -> dict:
5    """
6    Supervisor decides the next agent. Uses structured output
7    to ensure the decision is deterministically parseable.
8    """
9    from pydantic import BaseModel
10
11    class RoutingDecision(BaseModel):
12        next_agent: Literal["research_worker", "analysis_worker", "writer_worker", "FINISH"]
13        reasoning: str
14
15    structured_llm = llm.with_structured_output(RoutingDecision)
16    decision = structured_llm.invoke(state["messages"])
17    return {"next": decision.next_agent, "routing_log": decision.reasoning}

3.3 State Management in Detail

SqliteSaver is adequate for development and low loads. In production with concurrency:

pythoncopiar
1# langchain==0.3.x | langgraph==0.2.x | redis==5.x
2
3from langgraph.checkpoint.redis import RedisSaver
4
5# Production: Redis with TTL to prevent orphaned state accumulation
6checkpointer = RedisSaver.from_conn_string(
7    "redis://localhost:6379",
8    ttl={"default": 86400}  # 24h — adjust per process type
9)
10
11# Handoff pattern: explicit state indicating "ready for next agent"
12class HandoffState(TypedDict):
13    phase: Literal["extraction", "analysis", "writing", "done"]
14    phase_output: dict        # Current phase output
15    phase_metadata: dict      # Latency, tokens, model used
16    handoff_validated: bool   # Critic validated before handoff

4. Architecture with CrewAI

4.1 Declarative Role Model

CrewAI inverts the paradigm: instead of defining a technical graph, you define business responsibilities. An Agent is a role with a role (title), goal (objective), and backstory (context that shapes LLM behavior).

The example below implements a market intelligence Crew:

pythoncopiar
1# crewai==0.80.x | langchain-openai==0.2.x
2
3import logging
4from typing import Optional
5from pydantic import BaseModel
6
7from crewai import Agent, Task, Crew, Process
8from crewai.tools import BaseTool
9from langchain_openai import ChatOpenAI
10
11logger = logging.getLogger(__name__)
12
13
14# --- Custom tool ---
15
16class WebSearchTool(BaseTool):
17    """
18    Web search wrapper for use by agents.
19
20    In production, replace with a real integration (Tavily, Serper, etc).
21    """
22    name: str = "web_search"
23    description: str = "Searches the web for up-to-date information on a topic."
24
25    def _run(self, query: str) -> str:
26        # Real integration goes here
27        logger.info(f"web_search query={query}")
28        return f"[Simulated results for: {query}]"
29
30
31# --- Structured output schema ---
32
33class MarketIntelligenceReport(BaseModel):
34    """Pydantic schema for structured Crew output."""
35    executive_summary: str
36    key_competitors: list[str]
37    market_size_estimate: str
38    strategic_recommendations: list[str]
39    confidence_score: float  # 0.0 - 1.0
40
41
42# --- Agent definitions ---
43
44llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
45
46researcher = Agent(
47    role="Senior Market Research Specialist",
48    goal=(
49        "Collect factual, up-to-date data on market dynamics, competitors, and trends. "
50        "Prioritize primary sources. Flag when data points are estimates."
51    ),
52    backstory=(
53        "You are a competitive intelligence analyst with 10 years of experience "
54        "in B2B technology markets. You are skeptical, rigorous, and never fabricate data."
55    ),
56    tools=[WebSearchTool()],
57    llm=llm,
58    max_iter=5,          # Iteration limit — cost control
59    verbose=True,
60    allow_delegation=False  # Researcher does not delegate — executes directly
61)
62
63analyst = Agent(
64    role="Strategic Intelligence Analyst",
65    goal=(
66        "Transform raw market data into actionable insights. "
67        "Identify non-obvious patterns, anomalies, and opportunities."
68    ),
69    backstory=(
70        "You are a senior analyst specialized in synthesizing complex data. "
71        "You think in systems, not isolated data points."
72    ),
73    llm=llm,
74    max_iter=3,
75    verbose=True,
76    allow_delegation=False
77)
78
79strategist = Agent(
80    role="Go-to-Market Strategist",
81    goal=(
82        "Convert market insights into concrete strategic recommendations "
83        "with explicit prioritization criteria."
84    ),
85    backstory=(
86        "You are an execution-focused strategist. Your recommendations always "
87        "include: what to do, why, in what order, and how to measure success."
88    ),
89    llm=llm,
90    max_iter=3,
91    verbose=True,
92    allow_delegation=True  # Strategist can delegate reviews to Analyst
93)
94
95
96# --- Task definitions ---
97
98research_task = Task(
99    description=(
100        "Research the {market_segment} market focusing on: "
101        "1) Key players and estimated market share, "
102        "2) Growth trends over the past 18 months, "
103        "3) Recent M&A or funding activity. "
104        "Document each source used."
105    ),
106    expected_output=(
107        "Research report with raw data organized by category. "
108        "Include confidence level (high/medium/low) for each data point."
109    ),
110    agent=researcher
111)
112
113analysis_task = Task(
114    description=(
115        "Based on the research report, produce: "
116        "1) Positioning analysis of the top 3 competitors, "
117        "2) Identification of unaddressed market gaps, "
118        "3) Threats and opportunities assessment (matrix format)."
119    ),
120    expected_output=(
121        "Structured analysis with distinct sections for each deliverable. "
122        "Each insight must be supported by data from the research report."
123    ),
124    agent=analyst,
125    context=[research_task]  # Explicit dependency
126)
127
128strategy_task = Task(
129    description=(
130        "Based on the market analysis, develop strategic recommendations "
131        "prioritized by impact and 90-day execution feasibility."
132    ),
133    expected_output=(
134        "Executive report in MarketIntelligenceReport format with: "
135        "executive summary, key competitors, market size estimate, "
136        "prioritized recommendations, and overall confidence score."
137    ),
138    agent=strategist,
139    context=[research_task, analysis_task],
140    output_pydantic=MarketIntelligenceReport  # Structured, parseable output
141)
142
143
144# --- Crew assembly ---
145
146market_intel_crew = Crew(
147    agents=[researcher, analyst, strategist],
148    tasks=[research_task, analysis_task, strategy_task],
149    process=Process.sequential,  # Guaranteed order: research → analysis → strategy
150    verbose=True,
151    memory=True,                 # Enables memory between tasks
152    max_rpm=10,                  # Rate limiting — prevents API throttling
153)
154
155
156def run_market_intelligence(market_segment: str) -> MarketIntelligenceReport:
157    """
158    Runs the market intelligence Crew for a specific segment.
159
160    Returns:
161        MarketIntelligenceReport with structured, Pydantic-validated output.
162    """
163    logger.info(f"crew_start segment={market_segment}")
164    result = market_intel_crew.kickoff(inputs={"market_segment": market_segment})
165    logger.info(f"crew_done segment={market_segment}")
166    return result.pydantic

4.2 Collaboration Processes

Hierarchical with Manager LLM — CrewAI automatically instantiates a manager agent that decides task order and delegation:

pythoncopiar
1# crewai==0.80.x
2from crewai import LLM
3
4manager_llm = LLM(model="gpt-4o", temperature=0)  # Manager requires high precision
5
6hierarchical_crew = Crew(
7    agents=[researcher, analyst, strategist],
8    tasks=[research_task, analysis_task, strategy_task],
9    process=Process.hierarchical,
10    manager_llm=manager_llm,
11    verbose=True
12)

Memory Configuration in Detail:

pythoncopiar
1# crewai==0.80.x — Memory requires explicit embeddings configuration
2
3from langchain_openai import OpenAIEmbeddings
4
5crew_with_memory = Crew(
6    agents=[researcher, analyst, strategist],
7    tasks=[research_task, analysis_task, strategy_task],
8    process=Process.sequential,
9    memory=True,
10    # Short-term: current execution context (in-memory)
11    # Long-term: RAG over past executions (ChromaDB by default)
12    # Entity: graph of mentioned entities
13    embedder={
14        "provider": "openai",
15        "config": {"model": "text-embedding-3-small"}
16    },
17    verbose=True
18)

4.3 Advanced Production Configuration

Human-in-the-loop for high-risk decisions:

pythoncopiar
1# crewai==0.80.x
2# Human input pauses execution and waits for stdin input
3# In production: integrate with webhook or approval system
4
5approval_task = Task(
6    description="Validate whether the proposed strategy aligns with business objectives.",
7    expected_output="Approval or list of required adjustments.",
8    agent=strategist,
9    human_input=True  # Pauses execution for human review
10)

5. Real Production Challenges

5.1 Failure Management — Do Not Ignore This

The most common MAS failure pattern is not the agent returning garbage — it is the agent returning nothing due to timeout, rate limiting, or network error. Deterministic retry logic is non-negotiable:

pythoncopiar
1# langchain==0.3.x | tenacity==8.x
2
3import logging
4from functools import wraps
5from tenacity import (
6    retry,
7    stop_after_attempt,
8    wait_exponential,
9    retry_if_exception_type,
10    before_sleep_log
11)
12from openai import RateLimitError, APITimeoutError, APIConnectionError
13
14logger = logging.getLogger(__name__)
15
16RETRYABLE_EXCEPTIONS = (RateLimitError, APITimeoutError, APIConnectionError)
17
18
19def with_agent_retry(max_attempts: int = 3, min_wait: float = 1.0, max_wait: float = 30.0):
20    """
21    Retry decorator with exponential backoff for agent nodes.
22
23    Strategy: random jitter in wait avoids thundering herd
24    when multiple agents fail simultaneously.
25    """
26    def decorator(func):
27        @retry(
28            stop=stop_after_attempt(max_attempts),
29            wait=wait_exponential(multiplier=min_wait, max=max_wait),
30            retry=retry_if_exception_type(RETRYABLE_EXCEPTIONS),
31            before_sleep=before_sleep_log(logger, logging.WARNING),
32            reraise=True
33        )
34        @wraps(func)
35        def wrapper(*args, **kwargs):
36            return func(*args, **kwargs)
37        return wrapper
38    return decorator
39
40
41class CircuitBreaker:
42    """
43    Circuit breaker for calls to external APIs.
44
45    States: CLOSED (normal) → OPEN (consecutive failures) → HALF_OPEN (testing)
46    Prevents failure cascades when a downstream API is degraded.
47    """
48    def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 60.0):
49        self.failure_count = 0
50        self.failure_threshold = failure_threshold
51        self.recovery_timeout = recovery_timeout
52        self.state = "CLOSED"
53        self.last_failure_time: float = 0
54
55    def call(self, func, *args, **kwargs):
56        import time
57
58        if self.state == "OPEN":
59            if time.time() - self.last_failure_time > self.recovery_timeout:
60                self.state = "HALF_OPEN"
61                logger.info("circuit_breaker state=HALF_OPEN")
62            else:
63                raise RuntimeError("Circuit breaker OPEN — awaiting recovery")
64
65        try:
66            result = func(*args, **kwargs)
67            if self.state == "HALF_OPEN":
68                self.state = "CLOSED"
69                self.failure_count = 0
70                logger.info("circuit_breaker state=CLOSED")
71            return result
72        except Exception as e:
73            self.failure_count += 1
74            self.last_failure_time = time.time()
75            if self.failure_count >= self.failure_threshold:
76                self.state = "OPEN"
77                logger.error(f"circuit_breaker state=OPEN failures={self.failure_count}")
78            raise
79
80
81# Agent with fallback: if primary model fails, uses smaller model
82@with_agent_retry(max_attempts=3)
83def resilient_agent_node(state: dict) -> dict:
84    """Agent node with automatic retry and model fallback."""
85    try:
86        primary_llm = ChatOpenAI(model="gpt-4o", temperature=0)
87        return _execute_agent_logic(primary_llm, state)
88    except Exception as e:
89        logger.warning(f"primary_model_failed error={str(e)} falling_back=gpt-4o-mini")
90        fallback_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
91        return _execute_agent_logic(fallback_llm, state)

5.2 Observability Is Not Optional

pythoncopiar
1# langchain==0.3.x | python-json-logger==2.x
2
3import json
4import time
5import uuid
6from contextlib import contextmanager
7from pythonjsonlogger import jsonlogger
8
9# Structured JSON logging configuration
10handler = logging.StreamHandler()
11handler.setFormatter(jsonlogger.JsonFormatter(
12    fmt="%(asctime)s %(levelname)s %(name)s %(message)s"
13))
14logger = logging.getLogger("mas.agent")
15logger.addHandler(handler)
16
17
18@contextmanager
19def agent_trace(agent_name: str, correlation_id: str):
20    """
21    Context manager for agent execution tracing.
22
23    Captures: latency, status, estimated cost, tokens used.
24    Compatible with LangSmith via callbacks when configured.
25    """
26    span_id = str(uuid.uuid4())[:8]
27    start_time = time.perf_counter()
28
29    logger.info("agent_start", extra={
30        "agent": agent_name,
31        "correlation_id": correlation_id,
32        "span_id": span_id,
33        "event": "span_start"
34    })
35
36    try:
37        yield span_id
38        elapsed_ms = (time.perf_counter() - start_time) * 1000
39        logger.info("agent_success", extra={
40            "agent": agent_name,
41            "correlation_id": correlation_id,
42            "span_id": span_id,
43            "latency_ms": round(elapsed_ms, 2),
44            "event": "span_end",
45            "status": "success"
46        })
47    except Exception as e:
48        elapsed_ms = (time.perf_counter() - start_time) * 1000
49        logger.error("agent_error", extra={
50            "agent": agent_name,
51            "correlation_id": correlation_id,
52            "span_id": span_id,
53            "latency_ms": round(elapsed_ms, 2),
54            "event": "span_end",
55            "status": "error",
56            "error_type": type(e).__name__,
57            "error_msg": str(e)
58        })
59        raise

Audit log format — each entry is an independent JSON object, parseable by any log aggregation system (Datadog, CloudWatch, Loki):

jsoncopiar
1{
2  "asctime": "2026-03-05T14:32:01.234Z",
3  "levelname": "INFO",
4  "agent": "analyzer_node",
5  "correlation_id": "a3f7c2d1-8b4e-4f9a-b2c1-d5e8f0a1b3c4",
6  "span_id": "7f2c3a1b",
7  "latency_ms": 1243.7,
8  "event": "span_end",
9  "status": "success",
10  "tokens_used": 847,
11  "model": "gpt-4o-mini",
12  "estimated_cost_usd": 0.000423
13}

5.3 Cost and Rate Limiting

The cost of a MAS pipeline is not the sum of individual costs — it is amplified by retries, redundant context between agents, and unnecessary calls when state already satisfies the exit condition.

pythoncopiar
1# langchain==0.3.x | gptcache==0.1.x
2
3from gptcache import cache
4from gptcache.adapter import openai as cached_openai
5from gptcache.embedding import Onnx
6
7# Semantic caching: semantically similar requests
8# reuse previous responses — 30-60% cost reduction
9# in workflows with repetitive queries (illustrative figure)
10onnx = Onnx()
11cache.init(embedding_func=onnx.to_embeddings)
12cache.set_openai_key()
13
14# Cost estimate per orchestration pattern
15# (based on gpt-4o-mini at $0.15/1M input tokens — verify current pricing)
16COST_ESTIMATES = {
17    "sequential_5_agents": "~$0.002-0.008 per execution",
18    "parallel_5_agents": "~$0.002-0.008 per execution (same calls, lower latency)",
19    "hierarchical_supervisor": "~$0.005-0.020 per execution (+supervisor cost)",
20    "crew_sequential_3_agents": "~$0.003-0.012 per execution"
21}

6. Architecture Decision — Full Comparative Table

Decision rule in 3 lines:

1. Use LangGraph when the execution graph has complex conditional logic, strict regulatory audit requirements, or when the team has senior engineers available to maintain the infrastructure.
2. Use CrewAI when the business domain maps cleanly to roles, the team is small, the deadline is tight, and granular graph control is not a requirement.
3. Use hybrid when the business Crew (CrewAI) needs reliable technical sub-graphs for critical tasks — CrewAI orchestrates the business flow, LangGraph executes the steps that require determinism and full observability.

7. Conclusion

Three insights not found in the official documentation — they only emerge in production:

1. Shared state is the most important contract in the system. Before writing any agent code, define the complete state schema. Late changes to the TypedDict or Pydantic schema break persisted checkpoints and require migrations. Treat state the way you would treat a database schema.

2. The Critic (validator) reduces total cost, it does not increase it. The intuition that "one more agent = more cost" is incorrect when the Critic eliminates reprocessing caused by invalid outputs reaching downstream steps. In pipelines with more than 4 agents, a well-calibrated Critic reduces total cost by 15-35% (illustrative figure).

3. CrewAI and LangGraph do not compete — they stratify. The most robust pattern observed in production uses CrewAI to define "what to do" (business orchestration) and LangGraph to define "how to do it with guarantees" (critical execution sub-graphs). The separation of concerns is clean, and the resulting code is more readable than monoliths in either framework alone.

Concrete next steps:

1. Implement DocumentState with a real low-risk process — do not try to design the "perfect" state upfront. It will evolve.
2. Configure LangSmith or an equivalent before going to production — debugging MAS without tracing is orders of magnitude more costly than with it.
3. Write unit tests for each agent node with fixed input states — nodes are pure functions and are fully testable.

Keywords: Agent Orchestration LangChain CrewAI, LangGraph production, CrewAI advanced tutorial, Multi-Agent Systems Python, AI agent architecture, LLM orchestration framework.

Published by AI2You — AI-First Technical Series | ai2you.online/en/blog