AI Agents in Production: Building Autonomous Multi-Agent Systems with LangGraph and CrewAI

Overview

The landscape of Artificial Intelligence is rapidly evolving, moving beyond monolithic models to sophisticated, collaborative systems. We're entering an era where AI agents, endowed with specialized skills, memory, and the ability to use tools, can communicate and cooperate to achieve complex objectives. This paradigm shift is giving rise to autonomous multi-agent systems, capable of tackling intricate problems that would overwhelm a single agent or a traditional application.

At TechNews Venture, we've been closely tracking the development and adoption of these advanced architectures. Two prominent frameworks, LangGraph and CrewAI, are emerging as frontrunners in enabling developers to build and deploy such systems in production. LangGraph, built on top of LangChain, provides a robust, stateful, and cyclic graph-based orchestration layer, allowing for highly dynamic and adaptive agent workflows. CrewAI, on the other hand, offers an intuitive and powerful way to define roles, tasks, and collaboration patterns for multiple agents, making it ideal for creating cohesive teams that work towards a common goal.

This article delves into the practical aspects of combining LangGraph and CrewAI to construct powerful, production-ready multi-agent systems. We'll explore their individual strengths, demonstrate how to integrate them effectively, and provide a detailed walkthrough of building a complex system, complete with real-world code examples, security considerations, and best practices for deployment. Our goal is to equip you, the senior technologist, with the knowledge to leverage these frameworks to unlock new levels of automation and intelligence within your applications.

Prerequisites

Before we dive into the core implementation, ensure you have the following prerequisites in place. A solid foundation here will streamline your development process.

• Python 3.9+: The frameworks are built on modern Python versions.
• Package Manager (pip): For installing Python libraries.
• API Keys for LLMs: While we'll use Ollama for local examples, production deployments typically leverage commercial LLMs like OpenAI's GPT models or Anthropic's Claude.
  • For OpenAI: Obtain an API key from the OpenAI Developer Platform.
  • For Anthropic: Obtain an API key from the Anthropic Console.
• Ollama (Optional, for local development): If you wish to run open-source models locally without external API dependencies, install Ollama from ollama.com and pull a model like Llama 3 (e.g., ollama pull llama3).
• Basic understanding of Large Language Models (LLMs): Familiarity with concepts like prompts, tokens, and model capabilities.
• Familiarity with LangChain concepts: LangGraph extends LangChain, so understanding chains, agents, and tools is beneficial.

Let's start by setting up our Python environment:

# Create a virtual environment
python3 -m venv ai_agents_env
source ai_agents_env/bin/activate # On Windows, use `ai_agents_env\Scripts\activate`

# Install necessary packages
pip install langchain langchain-openai langgraph crewai 'crewai[tools]' beautifulsoup4 duckduckgo-search

# For local LLMs with Ollama
pip install langchain-community

Remember to set your API keys as environment variables:

export OPENAI_API_KEY="your_openai_api_key_here"
export SERPER_API_KEY="your_serper_api_key_here" # For internet search tools
# For Ollama, you might set the base URL if it's not default:
export OPENAI_API_BASE="http://localhost:11434/v1" # When using OpenAI client with Ollama
export OPENAI_MODEL_NAME="llama3" # Or whatever model you pulled

We'll use langchain-openai even with Ollama by redirecting the API base URL, demonstrating flexibility. For actual OpenAI models, simply remove the `OPENAI_API_BASE` variable and ensure `OPENAI_MODEL_NAME` is set to a valid OpenAI model like `gpt-4o` or `gpt-3.5-turbo`.

Detailed Steps with commands

1. Designing a Production-Ready Multi-Agent System: Automated Content Generation and Review

To demonstrate the power of combining LangGraph and CrewAI, let's design a system for automated content generation and review. This system will research a given topic, draft an article, and then have it reviewed by another agent, with potential for revisions. This mimics a common workflow in content marketing or technical writing teams.

1.1. Defining the Problem: Automated Blog Post Generation and Review

Our goal is to automatically generate a high-quality, SEO-optimized blog post on a given technical topic (e.g., "The Future of Quantum Computing") and ensure its accuracy and readability through an automated review process. This requires:

• Research: Gathering up-to-date information from the web.
• Drafting: Structuring and writing the article based on research.
• Review: Critically evaluating the draft for accuracy, tone, grammar, and SEO compliance.
• Revision (Conditional): If the review identifies issues, the draft should be sent back for revision.

1.2. Agent Roles and Responsibilities (CrewAI Focus)

We'll define distinct roles for our agents using CrewAI, each with specific goals and backstories:

• Research Analyst:
  • Role: Senior Research Analyst
  • Goal: Conduct comprehensive, up-to-date research on a given topic, identifying key trends, statistics, and expert opinions.
  • Backstory: An expert in information retrieval and synthesis, known for uncovering critical insights from vast amounts of data.
  • Tools: Internet Search (e.g., DuckDuckGoSearch, SerperDevTool).
• Content Creator:
  • Role: Professional Content Writer
  • Goal: Draft engaging, informative, and SEO-optimized articles based on research findings.
  • Backstory: A seasoned writer with a knack for transforming complex information into accessible and compelling narratives.
  • Tools: None directly, relies on research output.
• Technical Editor:
  • Role: Senior Technical Editor
  • Goal: Review drafted articles for technical accuracy, clarity, grammar, style, and SEO best practices. Provide constructive feedback for revisions.
  • Backstory: A meticulous editor with a strong technical background, ensuring all content meets the highest standards of quality and precision.
  • Tools: None directly, relies on drafted content.

1.3. Tool Integration

For our agents to perform their tasks effectively, they need tools. We'll primarily use an internet search tool for the Research Analyst.

from crewai_tools import SerperDevTool, tool
from langchain_community.tools import DuckDuckGoSearchRun

# Initialize the SerperDevTool for detailed search
# Ensure SERPER_API_KEY is set in your environment
search_tool = SerperDevTool()

# Alternatively, use DuckDuckGoSearch for a free option
duckduckgo_search = DuckDuckGoSearchRun()

# Custom tool for writing/appending to a file
@tool("File Write Tool")
def write_to_file(filename: str, content: str):
    """Writes the given content to a specified file.
    Useful for saving research notes or final articles."""
    try:
        with open(filename, "w") as f:
            f.write(content)
        return f"Content successfully written to {filename}"
    except Exception as e:
        return f"Error writing to file {filename}: {e}"

@tool("File Append Tool")
def append_to_file(filename: str, content: str):
    """Appends the given content to a specified file.
    Useful for adding incremental notes or sections."""
    try:
        with open(filename, "a") as f:
            f.write(content + "\n")
        return f"Content successfully appended to {filename}"
    except Exception as e:
        return f"Error appending to file {filename}: {e}"

1.4. Implementing with CrewAI (Core Workflow)

Now, let's define our agents, tasks, and the crew itself. We'll use OpenAI models for this, but as mentioned, you can configure to use Ollama.

import os
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

# Set up the LLM (using OpenAI client, configured for Ollama or OpenAI API)
llm = ChatOpenAI(
    model=os.getenv("OPENAI_MODEL_NAME", "gpt-4o"), # Use gpt-4o for production
    base_url=os.getenv("OPENAI_API_BASE", None), # None for actual OpenAI API
    temperature=0.7
)

# 1. Define Agents
research_analyst = Agent(
    role='Senior Research Analyst',
    goal='Conduct comprehensive, up-to-date research on the given topic, identifying key trends, statistics, and expert opinions.',
    backstory='An expert in information retrieval and synthesis, known for uncovering critical insights from vast amounts of data.',
    verbose=True,
    allow_delegation=False,
    tools=[search_tool], # Using SerperDevTool for robust search
    llm=llm
)

content_creator = Agent(
    role='Professional Content Writer',
    goal='Draft engaging, informative, and SEO-optimized articles based on research findings provided by the research analyst.',
    backstory='A seasoned writer with a knack for transforming complex information into accessible and compelling narratives.',
    verbose=True,
    allow_delegation=False,
    llm=llm
)

technical_editor = Agent(
    role='Senior Technical Editor',
    goal='Review drafted articles for technical accuracy, clarity, grammar, style, and SEO best practices. Provide constructive feedback for revisions.',
    backstory='A meticulous editor with a strong technical background, ensuring all content meets the highest standards of quality and precision.',
    verbose=True,
    allow_delegation=False,
    llm=llm
)

# 2. Define Tasks
research_task = Task(
    description=(
        "Conduct a comprehensive search on the topic: '{topic}'. "
        "Identify key concepts, recent advancements, challenges, and future outlook. "
        "Summarize findings in a detailed report, including sources."
    ),
    expected_output='A detailed research report, including key findings, statistics, expert quotes, and URLs to sources.',
    agent=research_analyst,
    output_file='research_report.md' # Save output to a file
)

drafting_task = Task(
    description=(
        "Using the research report on '{topic}', draft a 1000-1500 word blog post. "
        "The article should be engaging, informative, technically accurate, and SEO-optimized. "
        "Include an introduction, several body paragraphs with subheadings, and a conclusion. "
        "Format it as a markdown file."
    ),
    expected_output='A complete blog post draft in markdown format, ready for review.',
    agent=content_creator,
    context=[research_task], # Content creator relies on research output
    output_file='blog_post_draft.md'
)

review_task = Task(
    description=(
        "Review the blog post draft on '{topic}' for technical accuracy, clarity, grammar, style, and SEO optimization. "
        "Provide detailed feedback. If revisions are needed, clearly state what needs to be changed. "
        "If the article is satisfactory, approve it with a 'APPROVED' statement."
    ),
    expected_output='A review report with detailed feedback. If approved, state "APPROVED".',
    agent=technical_editor,
    context=[drafting_task] # Editor reviews the draft
)

# 3. Form the Crew
content_crew = Crew(
    agents=[research_analyst, content_creator, technical_editor],
    tasks=[research_task, drafting_task, review_task],
    process=Process.sequential, # Tasks run in a predefined order
    verbose=2 # Shows a lot of detail during execution
)

# 4. Kick off the Crew
# If running this directly, uncomment the following block
# topic_to_generate = "The Rise of Edge AI and its Impact on IoT"
# print("### Kicking off the Content Generation Crew ###")
# result = content_crew.kickoff(inputs={'topic': topic_to_generate})
# print("\n\n### Crew Work Completed ###")
# print(result)

1.5. Integrating LangGraph for Advanced Workflow Control (Conditional Revisions)

While CrewAI excels at defining collaborative tasks, LangGraph provides superior control over complex, dynamic workflows, state management, and conditional routing. We can use LangGraph to wrap our CrewAI process, specifically to handle the revision loop if the editor requests changes.

Here's how we'll integrate them:

1. A LangGraph node will initiate the CrewAI process (research, draft, initial review).
2. Another LangGraph node will analyze the output of the `review_task` from CrewAI.
3. Based on the review, LangGraph will conditionally route:
   • If "APPROVED", the workflow ends.
   • If "REVISIONS NEEDED", it will loop back to the `content_creator` (or a dedicated revision agent within CrewAI, or even a new CrewAI run with specific revision tasks). For simplicity, we'll simulate sending it back to the `content_creator` with explicit revision instructions.

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator

# Define the state for our LangGraph
class ArticleState(TypedDict):
    topic: str
    research_report: str
    drafted_article: str
    review_feedback: str
    iterations: int
    max_iterations: int

# Define the nodes (functions) for our LangGraph
def run_crew(state: ArticleState):
    """
    Runs the CrewAI process for research, drafting, and initial review.
    """
    print(f"--- Running CrewAI for topic: {state['topic']} (Iteration: {state['iterations']}) ---")
    topic = state['topic']

    # Re-initialize the crew for each run if state needs to be fresh,
    # or pass specific tasks/inputs for revision.
    # For this example, we'll re-run the whole sequence,
    # but in a real scenario, you'd have more granular CrewAI tasks for revisions.

    # Ensure tasks are always fresh or parameterized for specific runs
    current_research_task = Task(
        description=research_task.description.format(topic=topic),
        expected_output=research_task.expected_output,
        agent=research_analyst,
        output_file=f'research_report_{state["iterations"]}.md'
    )

    current_drafting_task = Task(
        description=drafting_task.description.format(topic=topic),
        expected_output=drafting_task.expected_output,
        agent=content_creator,
        context=[current_research_task],
        output_file=f'blog_post_draft_{state["iterations"]}.md'
    )

    current_review_task = Task(
        description=(
            review_task.description.format(topic=topic) +
            (f"\nPrevious feedback: {state['review_feedback']}" if state['review_feedback'] else "")
        ),
        expected_output=review_task.expected_output,
        agent=technical_editor,
        context=[current_drafting_task]
    )

    # If this is a revision, we might want to modify the drafting task
    if state['review_feedback']:
        current_drafting_task.description = (
            f"REVISE the blog post draft on '{topic}' based on the following feedback: "
            f"{state['review_feedback']}. "
            f"The original draft is available. Focus on addressing the issues raised. "
            f"The revised article should be engaging, informative, technically accurate, and SEO-optimized. "
            f"Include an introduction, several body paragraphs with subheadings, and a conclusion. "
            f"Format it as a markdown file."
        )

    content_crew_iteration = Crew(
        agents=[research_analyst, content_creator, technical_editor],
        tasks=[current_research_task, current_drafting_task, current_review_task],
        process=Process.sequential,
        verbose=0 # Keep crew verbose low within LangGraph for cleaner output
    )

    result = content_crew_iteration.kickoff(inputs={'topic': topic})

    # Read the outputs from files if they were saved
    final_research_report = ""
    if os.path.exists(current_research_task.output_file):
        with open(current_research_task.output_file, 'r') as f:
            final_research_report = f.read()

    final_draft = ""
    if os.path.exists(current_drafting_task.output_file):
        with open(current_drafting_task.output_file, 'r') as f:
            final_draft = f.read()

    # The review feedback is usually the direct output of the review task
    # We need to parse 'result' to get the specific review feedback
    review_output = ""
    # In a real scenario, parse the result string carefully to extract the editor's feedback.
    # For now, we'll assume the last task's output in 'result' is the review.
    # A more robust way would be to have the editor explicitly return a structured JSON.
    review_output_lines = result.split('\n')
    for line in reversed(review_output_lines):
        if "REVIEW REPORT:" in line or "APPROVED" in line or "REVISIONS NEEDED" in line:
            review_output = line # Simplified extraction
            break
    if not review_output and len(review_output_lines) > 0:
        review_output = review_output_lines[-1] # Fallback to last line

    return {
        "research_report": final_research_report,
        "drafted_article": final_draft,
        "review_feedback": review_output,
        "iterations": state['iterations'] + 1
    }

def decide_to_revise(state: ArticleState):
    """
    Analyzes the review feedback to decide if revisions are needed.
    """
    print(f"--- Deciding on revision (Iteration: {state['iterations']}) ---")
    feedback = state['review_feedback'].upper()
    if "APPROVED" in feedback:
        print("Article APPROVED. Ending workflow.")
        return "end"
    elif state['iterations'] >= state['max_iterations']:
        print(f"Max iterations ({state['max_iterations']}) reached. Ending workflow with pending revisions.")
        return "end"
    else:
        print("Revisions needed. Looping back for revision.")
        return "revise"

# Build the LangGraph
workflow = StateGraph(ArticleState)

workflow.add_node("run_crew_node", run_crew)
workflow.add_conditional_edges(
    "run_crew_node",
    decide_to_revise,
    {
        "revise": "run_crew_node", # Loop back to run_crew for revision
        "end": END
    }
)

workflow.set_entry_point("run_crew_node")

app = workflow.compile()

# Initial state
initial_state = {
    "topic": "The Impact of Quantum Machine Learning on Drug Discovery",
    "research_report": "",
    "drafted_article": "",
    "review_feedback": "",
    "iterations": 0,
    "max_iterations": 2 # Allow up to 2 revisions
}

print(f"\n### Starting LangGraph workflow for: {initial_state['topic']} ###")
final_state = None
for s in app.stream(initial_state):
    print(s)
    final_state = s

print("\n### LangGraph Workflow Complete ###")
if final_state and "run_crew_node" in final_state:
    print(f"Final Article Status: {'APPROVED' if 'APPROVED' in final_state['run_crew_node']['review_feedback'].upper() else 'NEEDS FURTHER REVISIONS'}")
    print("\n--- Final Draft ---")
    print(final_state['run_crew_node']['drafted_article'])
    print("\n--- Final Review Feedback ---")
    print(final_state['run_crew_node']['review_feedback'])
else:
    print("Workflow did not produce a final state as expected.")

This LangGraph setup demonstrates how to create a resilient, self-correcting workflow. The run_crew node executes our entire CrewAI pipeline. The decide_to_revise node then acts as a router, sending the process back to run_crew if the editor's feedback indicates revisions are needed, up to a maximum number of iterations. This is a powerful pattern for real-world production systems where iterative refinement is crucial.

Security Considerations for Production AI Agents

Deploying AI agents, especially multi-agent systems, introduces a unique set of security challenges that must be addressed rigorously. Neglecting these can lead to data breaches, system compromises, or intellectual property leakage.

1. Prompt Injection and Indirect Prompt Injection

Agents are susceptible to malicious inputs designed to override their programmed instructions or extract sensitive information. This can come directly from user input (prompt injection) or indirectly from data retrieved by tools (indirect prompt injection, e.g., an agent searching a malicious website).

• Mitigation:
  • Input Validation and Sanitization: Filter and sanitize all user inputs before they reach the LLM.
  • Privileged Access Separation: Design agents with specific, limited permissions. An agent accessing external tools should not have direct access to sensitive internal systems.
  • Sandboxing Tools: Isolate tools in sandboxed environments (e.g., Docker containers) to limit the blast radius if a tool is exploited.
  • LLM Guardrails: Implement tools like NeMo Guardrails or custom rule-based systems to detect and block malicious prompts.
  • Human-in-the-Loop: For critical actions or outputs, introduce human review points.

2. Data Privacy and Confidentiality

Agents often process sensitive user data or proprietary information, making data privacy paramount.

• Mitigation:
  • Data Minimization: Only feed agents the absolute minimum data required for their task.
  • PII Redaction/Anonymization: Implement techniques to detect and redact Personally Identifiable Information (PII) before it reaches the LLM or is stored.
  • Secure Data Storage and Transmission: Encrypt data at rest and in transit (TLS/SSL). Follow industry best practices for data storage (e.g., AWS S3 with KMS encryption, Azure Blob Storage with customer-managed keys).
  • Access Control (RBAC): Implement strict Role-Based Access Control for who can interact with or retrieve data from the agent system.
  • Data Retention Policies: Define and enforce clear data retention and deletion policies.

3. Tool Access and Least Privilege

AI agents utilizing external tools (search engines, databases, APIs) introduce potential attack vectors if not properly managed.

• Mitigation:
  • Least Privilege Principle: Grant agents and their associated tools only the permissions absolutely necessary to perform their functions. For instance, a search agent needs read-only access to the internet, not write access to internal databases.
  • API Key Management: Store API keys securely (e.g., AWS Secrets Manager, Azure Key Vault, HashiCorp Vault) and rotate them regularly. Never hardcode API keys.
  • Rate Limiting and Monitoring: Implement rate limiting on tool usage to prevent abuse and monitor tool invocation for suspicious patterns.
  • Tool Auditing: Log all tool calls, inputs, and outputs for auditing and incident response.

4. Model Security and Bias

The underlying LLMs can have vulnerabilities or exhibit biases that impact the agent's behavior.

• Mitigation:
  • Model Evaluation: Continuously evaluate models for safety, bias, and performance degradation.
  • Red Teaming: Proactively test agents with adversarial inputs to identify vulnerabilities.
  • Bias Detection and Mitigation: Implement techniques to detect and mitigate algorithmic bias in agent outputs.
  • Supply Chain Security: Be aware of the provenance of your models and dependencies.

5. Observability and Auditing

Visibility into agent behavior is crucial for detecting and responding to security incidents.

• Mitigation:
  • Comprehensive Logging: Log agent decisions, tool calls, LLM inputs/outputs, and state transitions. Ensure logs are immutable and stored securely.
  • Monitoring and Alerting: Set up real-time monitoring for unusual activities, failed tool calls, excessive resource consumption, or unexpected agent behavior. Integrate with SIEM systems.
  • Traceability: Use tools like LangSmith or Langfuse to trace agent execution paths, understand decision-making, and debug issues.

For example, using LangSmith for tracing:

export LANGCHAIN_TRACING_V2="true"
export LANGCHAIN_API_KEY="your_langsmith_api_key"
export LANGCHAIN_PROJECT="AI Agents in Production"

This will automatically send traces of your LangGraph and LangChain (used by CrewAI) executions to the LangSmith platform, providing invaluable debugging and auditing capabilities.

Best Practices for Deploying Multi-Agent Systems

Moving from a proof-of-concept to a production-grade multi-agent system requires careful planning and adherence to best practices. Here are key considerations:

1. Modularity and Abstraction

• Principle: Design agents, tools, and tasks as modular, reusable components.
• Implementation:
  • Isolate agent definitions, tool implementations, and task descriptions into separate files or modules.
  • Use clear naming conventions.
  • This simplifies testing, maintenance, and allows for swapping out components (e.g., changing a search tool from Serper to Google Custom Search without affecting agent logic).

2. Robust Error Handling and Resilience

• Principle: Anticipate failures and design the system to recover gracefully.
• Implementation:
  • Retry Mechanisms: Implement exponential backoff and retry logic for external API calls (LLMs, tools).
  • Timeout Management: Set strict timeouts for LLM calls and tool executions to prevent agents from hanging indefinitely.
  • Fallback Strategies: Define fallback mechanisms when primary tools or LLMs fail (e.g., switch to a simpler LLM, use cached data).
  • Circuit Breakers: Implement circuit breakers to prevent cascading failures in case of persistent issues with an external service.
  • State Persistence: For long-running workflows (like our LangGraph example), ensure that the graph's state can be checkpointed and restored. LangGraph supports this with various

    Share this article

    WhatsAppX (Twitter)LinkedInFacebookTelegramCopy Link