A senior engineer's field-tested breakdown of LangGraph, AutoGen, CrewAI, Microsoft Agent Framework, and Haystack — from reviewing real production systems across teams.
Everyone is building AI agents right now.
LangGraph. AutoGen. CrewAI. Semantic Kernel. Microsoft Agent Framework.
But most production AI systems don't actually need an agent framework.
Across multiple teams and production codebases I've reviewed, the same two failure modes appear constantly — over-engineering and under-engineering. In one case, replacing a complex agent framework with ~200 lines of plain tool-calling code made the system 3× faster, easier to debug, and easier to maintain. In another, the absence of a framework caused a codebase to collapse under its own complexity.
Both failures had the same root cause:
The problem isn't choosing the wrong framework. It's choosing a framework before understanding the workflow.
This guide is the decision framework I now use before touching any agent tooling.
TL;DR — Pick Your Approach in 30 Seconds
| Workflow Shape | Recommended Approach |
|---|---|
| Single request → tools → response | Plain tool calling (no framework) |
| Self-correcting loops, retries, re-evaluation | LangGraph |
| Parallel specialist agents | AutoGen 0.7.5 |
| Enterprise persistent agents (Azure) | Microsoft Agent Framework RC |
| Sequential role-based task delegation | CrewAI 1.10.1 |
| Document analysis and extraction pipelines | Haystack 2.x |
Workflow Shapes at a Glance
Two Real Failure Modes I've Seen Across Teams
The over-engineering case.
During a cross-team architecture review, I found a financial data analysis pipeline — pull market metrics, cross-reference filings, produce a risk score — built with three microservices, a LangGraph orchestrator with twelve nodes, Redis for inter-agent memory, and a separate evaluation loop. Six weeks of engineering. It worked.
What it actually needed: 180 lines of FastAPI with direct OpenAI tool calling. Same output. 3× faster inference. Any engineer on the team could debug it in minutes.
After the review, the team simplified it together. The framework had been adopted before anyone mapped the workflow shape — a straight line — and LangGraph's graph model added cost with no return.
The under-engineering case.
In a separate review, I found the opposite problem. An infrastructure incident response system — Azure alerts, metric retrieval, remediation decisions, rollback logic, retry conditions, escalation thresholds — built with hand-rolled state machines, custom retry logic, and a bespoke tool orchestration layer. Three weeks in, the codebase was a maze. Every new remediation path required rewriting core routing logic.
A proper agent framework would have provided state management, conditional branching, retry handling, and checkpointing for free. Instead the team was reinventing those primitives by hand — the most expensive kind of technical debt.
The pattern is the same in both directions: the architecture was chosen before the workflow was understood.
The Core Question: Does Your Workflow Resist Simplicity?
Before touching any framework, draw the workflow on paper. Then answer these:
- Does step N's output determine whether to redo step N-1? → You have a loop
- Do multiple specialized agents need to run simultaneously? → You have parallelism
- Does the workflow run for minutes or hours, surviving restarts? → You need persistent state
- Does the agent need to decide its next action from intermediate results? → Dynamic planning
- Do independent agents need to hand off context to each other? → Multi-agent delegation
None of these? Use plain tool calling. Here's what that looks like at production quality.
The Baseline: Plain Tool Calling (No Framework Needed)
Use case: Real-time ESG (Environmental, Social, Governance) risk scoring. Given a ticker, pull sustainability metrics, cross-reference regulatory filings, produce a risk-adjusted score, and persist an audit trail — one clean, observable service.
# requirements: fastapi, openai, psycopg2-binary, httpx, pydantic
from fastapi import FastAPI
from openai import OpenAI
from pydantic import BaseModel
import json, httpx, psycopg2, os
app = FastAPI()
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
class ESGRequest(BaseModel):
ticker: str
portfolio_id: str
def get_esg_metrics(ticker: str) -> dict:
resp = httpx.get(
f"https://api.sustainalytics.com/v1/esg/{ticker}",
headers={"Authorization": f"Bearer {os.environ['ESG_API_KEY']}"},
timeout=10
)
return resp.json()
def get_regulatory_flags(ticker: str) -> dict:
with psycopg2.connect(os.environ["DATABASE_URL"]) as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT flag_type, severity, filing_date, description
FROM regulatory_flags WHERE ticker = %s
AND filing_date > NOW() - INTERVAL '2 years'
ORDER BY severity DESC LIMIT 10
""", (ticker,))
rows = cur.fetchall()
return {"flags": [{"type": r[0], "severity": r[1], "date": str(r[2]), "detail": r[3]} for r in rows]}
TOOLS = [
{"type": "function", "function": {
"name": "get_esg_metrics",
"description": "Fetch ESG sustainability scores for a stock ticker",
"parameters": {"type": "object", "properties": {"ticker": {"type": "string"}}, "required": ["ticker"]}
}},
{"type": "function", "function": {
"name": "get_regulatory_flags",
"description": "Retrieve regulatory violations and compliance flags",
"parameters": {"type": "object", "properties": {"ticker": {"type": "string"}}, "required": ["ticker"]}
}}
]
TOOL_MAP = {
"get_esg_metrics": get_esg_metrics,
"get_regulatory_flags": get_regulatory_flags
}
@app.post("/assess-esg")
async def assess_esg(req: ESGRequest):
messages = [{"role": "user", "content": (
f"Run a full ESG risk assessment for {req.ticker} in portfolio {req.portfolio_id}. "
f"Produce a risk-adjusted score with recommendation: HOLD, REDUCE, or DIVEST."
)}]
while True:
response = client.chat.completions.create(
model="gpt-4o", messages=messages, tools=TOOLS, tool_choice="auto"
)
msg = response.choices[0].message
if not msg.tool_calls:
return {"ticker": req.ticker, "assessment": msg.content}
messages.append(msg)
for tc in msg.tool_calls:
result = TOOL_MAP[tc.function.name](**json.loads(tc.function.arguments))
messages.append({"role": "tool", "tool_call_id": tc.id, "content": json.dumps(result)})
Multi-step tool calling, audit trail, fully debuggable by any engineer. If this solves the problem — stop here. No framework needed.
1. LangGraph — Stateful Cyclic Workflows
Version: 1.1.2 | pip install langgraph langgraph-checkpoint-redis
Use when: Your workflow has genuine loops — the result of one step conditionally re-runs a previous step. Redis checkpointing lets state survive service restarts mid-workflow.
Avoid when: The workflow is strictly sequential with no conditional branching. The graph model adds measurable overhead for zero architectural return.
Use case: Automated cloud cost optimization — scan Azure VMs for underutilization, simulate right-sizing savings, apply low-risk changes, re-scan. The loop continues until no optimization exceeds the savings threshold.
# requirements: langgraph, langgraph-checkpoint-redis, openai
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.redis import RedisSaver
from typing import TypedDict, Annotated
import operator, json, os
from openai import OpenAI
oai = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
class CostState(TypedDict):
subscription_id: str
resources: list[dict]
candidates: Annotated[list, operator.add]
applied: Annotated[list, operator.add]
total_savings: float
iteration: int
def scan_resources(state: CostState) -> dict:
resources = azure_monitor_client.list_resources(state["subscription_id"])
underutilized = [r for r in resources if r["avg_cpu_7d"] < 15 and r["avg_memory_7d"] < 30]
return {"resources": underutilized, "iteration": state["iteration"] + 1}
def analyze_savings(state: CostState) -> dict:
resp = oai.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": f"""
Analyze these underutilized Azure resources and recommend right-sizing.
For each: resource_id, current_sku, recommended_sku, monthly_savings_usd, risk_level (LOW/MEDIUM/HIGH).
Only include LOW or MEDIUM risk recommendations.
Resources: {json.dumps(state["resources"])}
Respond: {{"recommendations": [...]}}
"""}],
response_format={"type": "json_object"}
)
candidates = json.loads(resp.choices[0].message.content)["recommendations"]
return {"candidates": candidates, "total_savings": sum(c["monthly_savings_usd"] for c in candidates)}
def apply_changes(state: CostState) -> dict:
applied = []
for c in state["candidates"]:
if c["risk_level"] == "LOW":
azure_compute.resize_vm(c["resource_id"], c["recommended_sku"])
applied.append(c)
return {"applied": applied}
def check_threshold(state: CostState) -> str:
return "loop" if state["total_savings"] > 500 and state["iteration"] < 5 else "done"
graph = StateGraph(CostState)
graph.add_node("scan", scan_resources)
graph.add_node("analyze", analyze_savings)
graph.add_node("apply", apply_changes)
graph.set_entry_point("scan")
graph.add_edge("scan", "analyze")
graph.add_edge("analyze", "apply")
graph.add_conditional_edges("apply", check_threshold, {"loop": "scan", "done": END})
optimizer = graph.compile(checkpointer=RedisSaver.from_conn_string(os.environ["REDIS_URL"]))
result = optimizer.invoke({
"subscription_id": os.environ["AZURE_SUBSCRIPTION_ID"],
"candidates": [], "applied": [], "total_savings": 0.0, "iteration": 0
})
Why this justifies LangGraph: The self-correcting re-scan is genuinely hard to express cleanly in plain tool calling without writing a custom state machine — which is exactly the under-engineering trap described earlier.
2. AutoGen 0.7.5 — Parallel Multi-Agent Collaboration
Version: 0.7.5 | pip install "autogen-agentchat>=0.7.5" "autogen-ext[openai,redis]"
Key additions in 0.7.5: linear memory via RedisMemory, fixed GraphFlow cycle detection, Anthropic thinking mode support, reasoning_effort parameter for GPT-5 models, improved Azure AI client streaming.
Use when: Multiple specialized agents run independent analysis simultaneously. Async, event-driven — agents can be deployed on separate containers with zero blocking between them.
Avoid when: The task is sequential and single-agent. Multi-agent coordination overhead only pays off when genuine parallelism exists.
Use case: Automated M&A due diligence — Legal, Financial, and Tech Audit agents work in parallel, then a Synthesis agent consolidates findings into an investment decision. Sequential execution would add hours of unnecessary latency to a time-sensitive deal process.
# requirements: autogen-agentchat>=0.7.5, autogen-ext[openai,redis]
import asyncio, os
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.memory.redis import RedisMemory
model_client = OpenAIChatCompletionClient(model="gpt-4o", api_key=os.environ["OPENAI_API_KEY"])
legal_agent = AssistantAgent(
name="LegalAgent",
model_client=model_client,
memory=[RedisMemory(redis_url=os.environ["REDIS_URL"], session_id="ma-legal")],
system_message="""M&A legal counsel. Identify: IP gaps, change-of-control clauses,
litigation exposure, employment liabilities.
Output JSON: {risk_items, severity: BLOCKER|HIGH|MEDIUM|LOW, deal_impact}""",
tools=[fetch_contract_repository, search_litigation_database]
)
financial_agent = AssistantAgent(
name="FinancialAgent",
model_client=model_client,
memory=[RedisMemory(redis_url=os.environ["REDIS_URL"], session_id="ma-financial")],
system_message="""M&A financial analyst. Identify: revenue quality, off-balance-sheet
liabilities, working capital needs post-acquisition, EBITDA normalization.
Output JSON: {financial_flags, normalized_ebitda, recommended_valuation_range}""",
tools=[fetch_financial_statements, compute_dcf_model]
)
tech_audit_agent = AssistantAgent(
name="TechAuditAgent",
model_client=model_client,
system_message="""Technical due diligence expert. Assess: tech debt score (1–10),
security vulnerabilities, scalability ceiling, bus-factor risk.
Output JSON: {tech_risks, estimated_remediation_cost_usd, integration_complexity}""",
tools=[clone_and_analyze_repo, run_dependency_scan]
)
synthesis_agent = AssistantAgent(
name="SynthesisAgent",
model_client=model_client,
system_message="""M&A deal lead. Wait for all domain agents to complete.
Synthesize: PROCEED / PROCEED_WITH_CONDITIONS / ABORT.
Include: top 5 risks, price adjustment recommendation, 90-day priorities.
End your message with: ANALYSIS_COMPLETE""",
tools=[generate_pdf_memo, notify_deal_team_slack]
)
async def run_ma_due_diligence(target: str):
team = SelectorGroupChat(
participants=[legal_agent, financial_agent, tech_audit_agent, synthesis_agent],
model_client=model_client,
termination_condition=TextMentionTermination("ANALYSIS_COMPLETE"),
selector_prompt="""Run LegalAgent, FinancialAgent, TechAuditAgent first (order flexible,
can run in parallel). Only select SynthesisAgent after all three have reported."""
)
async for msg in team.run_stream(task=f"Full M&A due diligence for: {target}"):
print(f"[{msg.source}] {str(msg.content)[:120]}...")
asyncio.run(run_ma_due_diligence("TargetCorp Inc."))
3. Microsoft Agent Framework RC — Enterprise Production Agents
Version: Release Candidate, Feb 19, 2026 | pip install agent-framework --pre
The most significant shift in the Microsoft AI ecosystem right now. This framework unifies Semantic Kernel and AutoGen into a single SDK — both are entering maintenance mode as of this writing, and all new Microsoft investment flows into Agent Framework first.
RC signals a frozen, stable API surface with GA targeting Q1 2026. Core capabilities: persistent threads (Cosmos DB), Service Bus integration, MCP + A2A protocol support, multi-agent orchestration with handoff and group chat patterns, streaming checkpointing for long-running agents, full .NET and Python support.
Use when: Azure-first enterprise teams, production 24/7 background agents, regulated systems requiring full audit trails, or any team currently building on Semantic Kernel or AutoGen.
Avoid when: Greenfield Python-only stacks with no Azure dependency, or where GA stability is a hard requirement before adoption.
Use case: Autonomous DevOps deployment agent — monitors CI/CD completion events, validates health via Application Insights, promotes through dev → staging → prod automatically, pages on-call only when a gate fails. Runs continuously as a persistent, resumable agent.
# requirements: agent-framework --pre, azure-identity, azure-monitor-query, kubernetes
import os
from agent_framework import AgentClient
from azure.identity import DefaultAzureCredential
credential = DefaultAzureCredential()
agent_client = AgentClient(
endpoint=os.environ["AZURE_AI_FOUNDRY_ENDPOINT"],
credential=credential
)
def check_app_insights(resource_id: str, environment: str, window_minutes: int = 5) -> dict:
from azure.monitor.query import MetricsQueryClient
from datetime import timedelta
monitor = MetricsQueryClient(credential)
result = monitor.query_resource(
resource_id,
metrics=["requests/failed", "requests/duration"],
timespan=timedelta(minutes=window_minutes)
)
return {
"environment": environment,
"error_rate_percent": _calculate_error_rate(result),
"p99_latency_ms": _calculate_p99(result),
"gate_passed": _calculate_error_rate(result) < 1.0 and _calculate_p99(result) < 500
}
def promote_deployment(service: str, image_tag: str, environment: str) -> dict:
from kubernetes import client as k8s, config
config.load_incluster_config()
k8s.AppsV1Api().patch_namespaced_deployment(
name=service, namespace=environment,
body={"spec": {"template": {"spec": {"containers": [{
"name": service,
"image": f"{os.environ['ACR_REGISTRY']}/{service}:{image_tag}"
}]}}}}
)
return {"status": "promoted", "service": service, "tag": image_tag, "environment": environment}
# Persistent thread — Cosmos DB preserves state across restarts
agent = agent_client.create_agent(
model="gpt-4o",
name="DeploymentOrchestrator",
instructions="""Autonomous DevOps deployment agent.
1. Validate health: error rate < 1%, p99 < 500ms, zero pod restarts
2. Run smoke tests on the deployed environment
3. All gates pass → promote to next environment
4. Any gate fails → halt, capture full diagnostics, page on-call
5. After prod deploy: monitor 15 minutes, auto-rollback if error rate > 2%
Log every action: timestamp, metric values, decision, outcome.""",
tools=[check_app_insights, promote_deployment, run_smoke_tests,
rollback_deployment, page_oncall, log_deployment_event]
)
thread = agent_client.create_thread()
async def handle_pipeline_completion(event: dict):
"""Azure Event Grid webhook — fires on every pipeline completion"""
agent_client.create_message(
thread_id=thread.id, role="user",
content=(
f"Deployment complete — Service: {event['service_name']}, "
f"Tag: {event['image_tag']}, Environment: {event['environment']}. "
f"Begin validation and promotion workflow."
)
)
return agent_client.create_and_process_run(thread_id=thread.id, agent_id=agent.id)
4. CrewAI 1.10.1 — Role-Based Sequential Pipelines
Version: 1.10.1 (March 3, 2026) | pip install crewai==1.10.1
⚠️ Note: v1.10.0 was yanked from PyPI due to an AMP runtime issue. Always pin to 1.10.1 directly. This release lazy loading in Memory module and resolves a concurrent multi-process LockException in production flows.
Use when: Clearly defined specialist roles, sequential task handoff, and fastest path from idea to working prototype. CrewAI is the most opinionated framework and the fastest to build with.
Avoid when: Fine-grained conditional edge control is needed, or throughput-sensitive production paths where abstraction overhead is measurable.
Use case: Automated system architecture pipeline — an Architect designs from requirements, a Reviewer stress-tests for production failures, a Documentation Lead produces the Architecture Decision Record for the engineering wiki.
# requirements: crewai==1.10.1
from crewai import Agent, Task, Crew, Process
architect = Agent(
role="Principal Solutions Architect",
goal="Design a scalable, cloud-native system architecture from requirements",
backstory="10+ years on Azure and AWS. Expert in event-driven architecture, CQRS, and zero-trust.",
verbose=True
)
reviewer = Agent(
role="Staff Engineer — Technical Reviewer",
goal="Identify scalability bottlenecks, security vulnerabilities, and operational risks",
backstory="Former SRE. Has seen every production failure mode. Only approves designs that hold at 10x load.",
verbose=True
)
doc_lead = Agent(
role="Technical Documentation Lead",
goal="Produce a complete Architecture Decision Record capturing every design trade-off",
backstory="Undocumented architecture is a liability. Writes for the engineer joining 2 years later.",
verbose=True
)
design_task = Task(
description="""Design a complete architecture for: {requirements}
Output JSON: components (name + responsibility), data_flows (source → destination),
tech_choices (with justification), scaling_strategy per component, security_controls.""",
expected_output="JSON architecture specification",
agent=architect
)
review_task = Task(
description="""Stress-test the architecture:
1. 10x traffic spike — what breaks first?
2. Single region failure — what is the blast radius?
3. Compromised service account — what can an attacker reach?
Return: {approved: YES/NO, blocking_issues: [], recommended_changes: []}""",
expected_output="Review JSON with approval and findings",
agent=reviewer,
context=[design_task]
)
adr_task = Task(
description="Write a complete ADR: Context, Decision, Consequences, Alternatives Considered, Risk Register. Format: Markdown.",
expected_output="Complete ADR in Markdown",
agent=doc_lead,
context=[design_task, review_task]
)
crew = Crew(
agents=[architect, reviewer, doc_lead],
tasks=[design_task, review_task, adr_task],
process=Process.sequential,
verbose=True
)
result = crew.kickoff(inputs={
"requirements": "Real-time fraud detection — 50K TPS, sub-100ms decisions, 99.99% uptime, multi-region active-active"
})
print(result.raw)
5. Haystack 2.x — Document Intelligence Pipelines
Version: 2.x | pip install haystack-ai
Use when: Document processing, extraction, or compliance analysis at scale is the core product — not general agentic behavior. Purpose-built for this and measurably outperforms general frameworks on document-centric workloads.
Avoid when: The problem is general agentic orchestration, multi-agent coordination, or any real-time interactive system.
Use case: Automated SOC 2 evidence collection — ingests all internal policy documents, maps clauses against Trust Service Criteria, produces a compliance gap report showing which controls are missing or non-conformant.
The pipeline maps extracted content against all six SOC 2 Trust Service Criteria (CC6, CC7, CC8, CC9, A1, C1), flags NOT_COVERED gaps with descriptions, and returns a structured compliance report with overall coverage percentage — no custom parsing layer required.
The Practical Decision Rule
If your AI system needs:
Retries or self-correction → LangGraph
Multiple parallel specialists → AutoGen
Enterprise Azure persistence → Microsoft Agent Framework
Sequential role-based tasks → CrewAI
Document extraction at scale → Haystack
None of the above → No framework. Plain tool calling.
Framework Reference
| Framework | Version | Primary Strength | Use When | Avoid When |
|---|---|---|---|---|
| Plain tool calling | — | Speed, simplicity, debuggability | Straight-line workflow | Never — if no loops exist |
| LangGraph | 1.1.2 | Cyclic graphs, checkpointing | Self-correcting loops, retries | No conditional branching |
| AutoGen | 0.7.5 | Parallel async agents, RedisMemory | Multiple specialists in parallel | Sequential single-agent tasks |
| Microsoft Agent Framework | RC Feb 2026 | Enterprise persistence, SK + AutoGen unified | Azure production 24/7 agents | Python-only, no Azure stack |
| CrewAI | 1.10.1 | Role-based prototyping, fast iteration | Sequential delegation, fast prototyping | Fine-grained production control |
| Haystack | 2.x | Document extraction pipelines | Document processing as core product | General agentic tasks |
Production Lessons From Real Systems
After reviewing AI systems across multiple teams, a few patterns appear consistently:
1. Most workflows are simpler than they look.
The instinct when building with LLMs is to reach for orchestration layers. That instinct is usually wrong. Start with the simplest thing that could work, measure it, then add complexity only where the problem resists simplicity.
2. Agent frameworks pay off only when the workflow has the right shape.
Loops, parallel specialists, or long-running persistent state. If none of these exist, the framework is cost with no architectural return.
3. Debuggability matters more than clever architecture.
A system the team can debug at 2AM is worth more than an elegant multi-agent pipeline nobody fully understands. Production incidents don't wait for framework comprehension.
4. Hand-rolling framework primitives is the most expensive mistake.
Custom state machines, retry logic, and checkpointing written to avoid a framework dependency consistently cost more engineering time than learning the framework properly. Both failure modes described earlier confirm this.
5. Decide the architecture before writing the first line.
Draw the workflow. Identify the shape. Pick the tool. That decision should take 30 minutes, not six weeks of refactoring.
The Real Rule
If you can draw your workflow as a straight line — plain tool calling is your production architecture.
If that line needs to loop, branch, or coordinate parallel specialists — match the tool to the shape.
The best production AI systems are architecturally boring. One clean service, typed tool schemas, structured output, observable logs. No framework overhead unless the problem demands it.
Add complexity only when the problem resists simplicity. That's the whole framework.
What's Next in This Series
This post focused on orchestration — how to structure and run AI workflows in production.
But once orchestration is solved, most teams run into a different problem:
Their RAG system doesn’t actually work.
Not in demos — those look fine.
In production — it breaks.
Wrong answers. Missing context. Hallucinations with high confidence.
And in most cases, the root cause is not the vector database or the embedding model.
It’s the architecture around it.
The next article breaks down:
- Why most RAG systems fail after launch
- The common design mistakes teams repeat
- And the production architecture that actually fixes it
If you’re building anything on top of retrieval, this is where things either scale — or quietly fail.
Are you running an agent framework in production — or did you strip one out and go back to basics?
What made that decision clear? Drop your stack and the reason in the comments. The real production stories are always more useful than the official docs.
Tags: #ai #llm #machinelearning #softwareengineering #devops
Version data sourced from official release channels: AutoGen 0.7.5 , Microsoft Agent Framework RC, CrewAI 1.10.1, LangGraph, Haystack
Top comments (0)