Multi-Agent Workflows with CrewAI and LangChain Integration
Multi-agent systems are moving from research demos to production applications. The key challenge now is orchestrating them reliably for real work. CrewAI answers this with an intuitive metaphor: agents are crew members with roles, they execute tasks with specific goals, and they collaborate through defined processes. When you combine this with LangChain's extensive tool ecosystem, you get a practical framework for building complex workflows.
We will build a multi-agent content research and publishing pipeline that demonstrates sequential and hierarchical processes, custom tools, and LangChain integration.
Understanding CrewAI's mental model
CrewAI organizes work around three core concepts:
- Agents have a role, a goal, and a backstory that shapes their behavior
- Tasks define specific work items with expected outputs
- Crews bring agents and tasks together with a defined process
This maps naturally to how teams work in the real world. A research analyst gathers data, a writer produces content, an editor reviews it. Each person knows their role and what success looks like.
Setting up
pip install crewai crewai-tools langchain-openai langchain-community
Defining agents
Each agent gets a clear role, goal, and backstory. The backstory is not just flavor text; it significantly affects how the LLM approaches problems.
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
research_analyst = Agent(
role="Senior Research Analyst",
goal="Conduct thorough research on the given topic and produce "
"detailed, accurate findings with supporting evidence",
backstory=(
"You are a veteran research analyst with 15 years of experience "
"in technology research. You are meticulous about fact-checking, "
"always look for primary sources, and excel at synthesizing "
"complex information into clear findings. You never speculate "
"without clearly marking it as such."
),
llm=llm,
verbose=True,
allow_delegation=False,
)
content_strategist = Agent(
role="Content Strategist",
goal="Transform research findings into a compelling content plan "
"with clear structure, key messages, and target audience alignment",
backstory=(
"You are a content strategist who has worked with major tech "
"publications. You understand how to structure technical content "
"for maximum impact, balance depth with accessibility, and "
"create narratives that engage both beginners and experts."
),
llm=llm,
verbose=True,
allow_delegation=False,
)
technical_writer = Agent(
role="Senior Technical Writer",
goal="Produce polished, publication-ready technical articles "
"that are accurate, engaging, and well-structured",
backstory=(
"You are an award-winning technical writer specializing in "
"AI and software engineering. Your writing is known for clear "
"explanations, practical code examples, and a conversational "
"tone that makes complex topics accessible without dumbing "
"them down."
),
llm=llm,
verbose=True,
allow_delegation=False,
)
Setting allow_delegation=False prevents agents from passing their work to other agents. This gives you explicit control over the workflow. In a hierarchical process, you would set this to True for the manager agent.
Integrating LangChain tools
CrewAI integrates easily with LangChain tools. You can use LangChain's built-in tools or create custom ones:
from crewai_tools import SerperDevTool, WebsiteSearchTool
from crewai.tools import tool
# Built-in search tool
search_tool = SerperDevTool()
# Website-specific search
docs_search = WebsiteSearchTool(
website="https://docs.python.org/3/"
)
# Custom tool using the @tool decorator
@tool("Content Analyzer")
def analyze_content_quality(text: str) -> str:
"""Analyze content quality and provide a structured assessment
including readability, technical depth, and engagement metrics.
Args:
text: The content to analyze.
"""
words = text.split()
word_count = len(words)
avg_word_length = sum(len(w) for w in words) / max(len(words), 1)
sentence_count = text.count(".") + text.count("!") + text.count("?")
avg_sentence_length = word_count / max(sentence_count, 1)
code_blocks = text.count("```")
has_headings = "##" in text or "**" in text
analysis = f"""Content Quality Analysis:
- Word count: {word_count}
- Average word length: {avg_word_length:.1f} characters
- Sentences: {sentence_count}
- Average sentence length: {avg_sentence_length:.1f} words
- Code blocks: {code_blocks // 2}
- Has structure (headings): {has_headings}
- Readability: {'Good' if avg_sentence_length < 25 else 'Needs improvement'}
- Technical depth: {'High' if code_blocks > 2 else 'Medium' if code_blocks > 0 else 'Low'}
"""
return analysis
# Assign tools to agents
research_analyst.tools = [search_tool, docs_search]
technical_writer.tools = [analyze_content_quality]
Defining tasks
Tasks are the work items that agents execute. Each task has a description, an expected output format, and is assigned to a specific agent.
research_task = Task(
description=(
"Research the topic: '{topic}'. Your research should cover:\n"
"1. Current state and recent developments\n"
"2. Key players and their contributions\n"
"3. Technical architecture and implementation patterns\n"
"4. Advantages and limitations\n"
"5. Real-world use cases and case studies\n\n"
"Use the search tools to find up-to-date information. "
"Focus on technical accuracy and cite your sources."
),
expected_output=(
"A comprehensive research report with sections for each "
"area of investigation, including specific data points, "
"technical details, and source references."
),
agent=research_analyst,
)
strategy_task = Task(
description=(
"Based on the research findings, create a content strategy:\n"
"1. Identify the target audience (beginner/intermediate/advanced)\n"
"2. Define 3-5 key messages to convey\n"
"3. Outline the article structure with section descriptions\n"
"4. Suggest code examples that would illustrate key concepts\n"
"5. Recommend a tone and style approach"
),
expected_output=(
"A detailed content plan document with target audience, "
"key messages, article outline, code example suggestions, "
"and style guidelines."
),
agent=content_strategist,
context=[research_task],
)
writing_task = Task(
description=(
"Write a complete technical article based on the research "
"and content strategy. The article should:\n"
"1. Follow the outlined structure exactly\n"
"2. Include practical code examples with explanations\n"
"3. Be between 1500-2000 words\n"
"4. Use a conversational yet authoritative tone\n"
"5. Include a compelling introduction and actionable conclusion\n\n"
"After writing, use the Content Analyzer tool to assess quality "
"and revise if needed."
),
expected_output=(
"A publication-ready technical article in Markdown format "
"with proper headings, code blocks, and a natural flow "
"from introduction to conclusion."
),
agent=technical_writer,
context=[research_task, strategy_task],
)
The context parameter matters here. It tells CrewAI to pass the output of specified tasks as input context to the current task. This is how information flows through the pipeline.
Sequential process
The simplest orchestration pattern runs tasks in order:
sequential_crew = Crew(
agents=[research_analyst, content_strategist, technical_writer],
tasks=[research_task, strategy_task, writing_task],
process=Process.sequential,
verbose=True,
)
result = sequential_crew.kickoff(
inputs={"topic": "Building event-driven architectures with Python and Apache Kafka"}
)
print(result.raw)
In sequential mode, each task runs after the previous one completes. The output of each task is available to subsequent tasks through the context mechanism.
Hierarchical process
For more complex workflows, hierarchical process adds a manager agent that coordinates the team:
manager = Agent(
role="Editorial Director",
goal="Coordinate the research, strategy, and writing team to "
"produce the highest quality technical content possible",
backstory=(
"You are an editorial director at a leading tech publication. "
"You excel at coordinating specialists, providing clear "
"direction, and ensuring the final output meets publication "
"standards. You know when to push back and when to approve."
),
llm=ChatOpenAI(model="gpt-4o", temperature=0.3),
allow_delegation=True,
)
hierarchical_crew = Crew(
agents=[research_analyst, content_strategist, technical_writer],
tasks=[research_task, strategy_task, writing_task],
process=Process.hierarchical,
manager_agent=manager,
verbose=True,
)
result = hierarchical_crew.kickoff(
inputs={"topic": "Serverless machine learning inference at scale"}
)
In hierarchical mode, the manager agent decides which agent should work on which task, can request revisions, and coordinates the overall workflow. This is more flexible but uses more tokens since the manager adds an orchestration layer.
Handling structured output data
CrewAI supports structured output through Pydantic models, which is essential for integrating agent outputs into larger systems:
from pydantic import BaseModel, Field
class ArticleOutput(BaseModel):
title: str = Field(description="The article title")
summary: str = Field(description="A 2-3 sentence summary")
content: str = Field(description="The full article in Markdown")
tags: list[str] = Field(description="Relevant tags for the article")
target_audience: str = Field(description="Primary target audience level")
writing_task_structured = Task(
description=(
"Write a complete technical article based on the research "
"and content strategy provided."
),
expected_output="A structured article with title, summary, content, tags, and audience.",
agent=technical_writer,
context=[research_task, strategy_task],
output_pydantic=ArticleOutput,
)
When output_pydantic is set, CrewAI ensures the agent's output conforms to the schema. You can then access typed fields directly:
result = sequential_crew.kickoff(inputs={"topic": "Edge computing with Python"})
article = result.pydantic
print(f"Title: {article.title}")
print(f"Tags: {', '.join(article.tags)}")
Error handling and guardrails
Production workflows need guardrails. CrewAI provides several mechanisms:
from crewai import Crew
crew = Crew(
agents=[research_analyst, content_strategist, technical_writer],
tasks=[research_task, strategy_task, writing_task],
process=Process.sequential,
max_rpm=30, # Rate limit API calls
max_iter=15, # Maximum iterations per task
verbose=True,
memory=True, # Enable memory across tasks
)
try:
result = crew.kickoff(
inputs={"topic": "Zero-trust security in microservices"}
)
if result.raw:
print("Pipeline completed successfully")
print(f"Token usage: {result.token_usage}")
else:
print("Pipeline produced no output")
except Exception as e:
print(f"Pipeline failed: {e}")
The max_rpm parameter prevents you from hitting API rate limits, max_iter prevents infinite loops when an agent gets stuck, and memory=True enables agents to recall context from earlier in the workflow.
Key takeaways
CrewAI hits a good balance between simplicity and power. Its crew metaphor maps naturally to how teams work, which makes agent workflows intuitive to design. The LangChain integration gives you access to a large tool ecosystem without reinventing the wheel.
For straightforward pipelines, use sequential processes. For complex workflows where the optimal execution order is not known in advance, use hierarchical processes with a manager agent. And always define structured outputs when your agent pipeline feeds into a larger system.
CrewAI's orchestration paired with LangChain's tools gives you a solid base for multi-agent automation.