# CrewAI Documentation Source: https://docs.crewai.com/index Build collaborative AI agents, crews, and flows — production ready from day one.
CrewAI

Ship multi‑agent systems with confidence

Design agents, orchestrate crews, and automate flows with guardrails, memory, knowledge, and observability baked in.

Get started Coding-agent guide API Reference
## Get started Overview of CrewAI concepts, architecture, and what you can build with agents, crews, and flows. Install via `uv`, configure API keys, and set up the CLI for local development. Spin up your first crew in minutes. Learn the core runtime, project layout, and dev loop. ## Build the basics Compose agents with tools, memory, knowledge, and structured outputs using Pydantic. Includes templates and best practices. Orchestrate start/listen/router steps, manage state, persist execution, and resume long-running workflows. Define sequential, hierarchical, or hybrid processes with guardrails, callbacks, and human-in-the-loop triggers. ## Enterprise journey Manage environments, redeploy safely, and monitor live runs directly from the Enterprise console. Connect Gmail, Slack, Salesforce, and more. Pass trigger payloads into crews and flows automatically. Invite teammates, configure RBAC, and control access to production automations. ## What’s new Unified overview for Gmail, Drive, Outlook, Teams, OneDrive, HubSpot, and more — now with sample payloads and crews. Call existing CrewAI automations or Amazon Bedrock Agents directly from your crews using the updated integration toolkit. Browse the examples and cookbooks for end-to-end reference implementations across agents, flows, and enterprise automations. ## Stay connected If CrewAI helps you ship faster, give us a star and share your builds with the community. Ask questions, showcase workflows, and request features alongside other builders. # CrewAI Documentation Source: https://docs.crewai.com/index Build collaborative AI agents, crews, and flows — production ready from day one.
CrewAI

Ship multi‑agent systems with confidence

Design agents, orchestrate crews, and automate flows with guardrails, memory, knowledge, and observability baked in.

Get started Coding-agent guide API Reference
## Get started Overview of CrewAI concepts, architecture, and what you can build with agents, crews, and flows. Install via `uv`, configure API keys, and set up the CLI for local development. Spin up your first crew in minutes. Learn the core runtime, project layout, and dev loop. ## Build the basics Compose agents with tools, memory, knowledge, and structured outputs using Pydantic. Includes templates and best practices. Orchestrate start/listen/router steps, manage state, persist execution, and resume long-running workflows. Define sequential, hierarchical, or hybrid processes with guardrails, callbacks, and human-in-the-loop triggers. ## Enterprise journey Manage environments, redeploy safely, and monitor live runs directly from the Enterprise console. Connect Gmail, Slack, Salesforce, and more. Pass trigger payloads into crews and flows automatically. Invite teammates, configure RBAC, and control access to production automations. ## What’s new Unified overview for Gmail, Drive, Outlook, Teams, OneDrive, HubSpot, and more — now with sample payloads and crews. Call existing CrewAI automations or Amazon Bedrock Agents directly from your crews using the updated integration toolkit. Browse the examples and cookbooks for end-to-end reference implementations across agents, flows, and enterprise automations. ## Stay connected If CrewAI helps you ship faster, give us a star and share your builds with the community. Ask questions, showcase workflows, and request features alongside other builders. # Crafting Effective Agents Source: https://docs.crewai.com/v1.15.13/en/guides/agents/crafting-effective-agents Learn best practices for designing powerful, specialized AI agents that collaborate effectively to solve complex problems. ## The Art and Science of Agent Design At the heart of CrewAI lies the agent - a specialized AI entity designed to perform specific roles within a collaborative framework. While creating basic agents is simple, crafting truly effective agents that produce exceptional results requires understanding key design principles and best practices. This guide will help you master the art of agent design, enabling you to create specialized AI personas that collaborate effectively, think critically, and produce high-quality outputs tailored to your specific needs. ### Why Agent Design Matters The way you define your agents significantly impacts: 1. **Output quality**: Well-designed agents produce more relevant, high-quality results 2. **Collaboration effectiveness**: Agents with complementary skills work together more efficiently 3. **Task performance**: Agents with clear roles and goals execute tasks more effectively 4. **System scalability**: Thoughtfully designed agents can be reused across multiple crews and contexts Let's explore best practices for creating agents that excel in these dimensions. ## The 80/20 Rule: Focus on Tasks Over Agents When building effective AI systems, remember this crucial principle: **80% of your effort should go into designing tasks, and only 20% into defining agents**. Why? Because even the most perfectly defined agent will fail with poorly designed tasks, but well-designed tasks can elevate even a simple agent. This means: * Spend most of your time writing clear task instructions * Define detailed inputs and expected outputs * Add examples and context to guide execution * Dedicate the remaining time to agent role, goal, and backstory This doesn't mean agent design isn't important - it absolutely is. But task design is where most execution failures occur, so prioritize accordingly. ## Core Principles of Effective Agent Design ### 1. The Role-Goal-Backstory Framework The most powerful agents in CrewAI are built on a strong foundation of three key elements: #### Role: The Agent's Specialized Function The role defines what the agent does and their area of expertise. When crafting roles: * **Be specific and specialized**: Instead of "Writer," use "Technical Documentation Specialist" or "Creative Storyteller" * **Align with real-world professions**: Base roles on recognizable professional archetypes * **Include domain expertise**: Specify the agent's field of knowledge (e.g., "Financial Analyst specializing in market trends") **Examples of effective roles:** ```yaml theme={null} role: "Senior UX Researcher specializing in user interview analysis" role: "Full-Stack Software Architect with expertise in distributed systems" role: "Corporate Communications Director specializing in crisis management" ``` #### Goal: The Agent's Purpose and Motivation The goal directs the agent's efforts and shapes their decision-making process. Effective goals should: * **Be clear and outcome-focused**: Define what the agent is trying to achieve * **Emphasize quality standards**: Include expectations about the quality of work * **Incorporate success criteria**: Help the agent understand what "good" looks like **Examples of effective goals:** ```yaml theme={null} goal: "Uncover actionable user insights by analyzing interview data and identifying recurring patterns, unmet needs, and improvement opportunities" goal: "Design robust, scalable system architectures that balance performance, maintainability, and cost-effectiveness" goal: "Craft clear, empathetic crisis communications that address stakeholder concerns while protecting organizational reputation" ``` #### Backstory: The Agent's Experience and Perspective The backstory gives depth to the agent, influencing how they approach problems and interact with others. Good backstories: * **Establish expertise and experience**: Explain how the agent gained their skills * **Define working style and values**: Describe how the agent approaches their work * **Create a cohesive persona**: Ensure all elements of the backstory align with the role and goal **Examples of effective backstories:** ```yaml theme={null} backstory: "You have spent 15 years conducting and analyzing user research for top tech companies. You have a talent for reading between the lines and identifying patterns that others miss. You believe that good UX is invisible and that the best insights come from listening to what users don't say as much as what they do say." backstory: "With 20+ years of experience building distributed systems at scale, you've developed a pragmatic approach to software architecture. You've seen both successful and failed systems and have learned valuable lessons from each. You balance theoretical best practices with practical constraints and always consider the maintenance and operational aspects of your designs." backstory: "As a seasoned communications professional who has guided multiple organizations through high-profile crises, you understand the importance of transparency, speed, and empathy in crisis response. You have a methodical approach to crafting messages that address concerns while maintaining organizational credibility." ``` ### 2. Specialists Over Generalists Agents perform significantly better when given specialized roles rather than general ones. A highly focused agent delivers more precise, relevant outputs: **Generic (Less Effective):** ```yaml theme={null} role: "Writer" ``` **Specialized (More Effective):** ```yaml theme={null} role: "Technical Blog Writer specializing in explaining complex AI concepts to non-technical audiences" ``` **Specialist Benefits:** * Clearer understanding of expected output * More consistent performance * Better alignment with specific tasks * Improved ability to make domain-specific judgments ### 3. Balancing Specialization and Versatility Effective agents strike the right balance between specialization (doing one thing extremely well) and versatility (being adaptable to various situations): * **Specialize in role, versatile in application**: Create agents with specialized skills that can be applied across multiple contexts * **Avoid overly narrow definitions**: Ensure agents can handle variations within their domain of expertise * **Consider the collaborative context**: Design agents whose specializations complement the other agents they'll work with ### 4. Setting Appropriate Expertise Levels The expertise level you assign to your agent shapes how they approach tasks: * **Novice agents**: Good for straightforward tasks, brainstorming, or initial drafts * **Intermediate agents**: Suitable for most standard tasks with reliable execution * **Expert agents**: Best for complex, specialized tasks requiring depth and nuance * **World-class agents**: Reserved for critical tasks where exceptional quality is needed Choose the appropriate expertise level based on task complexity and quality requirements. For most collaborative crews, a mix of expertise levels often works best, with higher expertise assigned to core specialized functions. ## Practical Examples: Before and After Let's look at some examples of agent definitions before and after applying these best practices: ### Example 1: Content Creation Agent **Before:** ```yaml theme={null} role: "Writer" goal: "Write good content" backstory: "You are a writer who creates content for websites." ``` **After:** ```yaml theme={null} role: "B2B Technology Content Strategist" goal: "Create compelling, technically accurate content that explains complex topics in accessible language while driving reader engagement and supporting business objectives" backstory: "You have spent a decade creating content for leading technology companies, specializing in translating technical concepts for business audiences. You excel at research, interviewing subject matter experts, and structuring information for maximum clarity and impact. You believe that the best B2B content educates first and sells second, building trust through genuine expertise rather than marketing hype." ``` ### Example 2: Research Agent **Before:** ```yaml theme={null} role: "Researcher" goal: "Find information" backstory: "You are good at finding information online." ``` **After:** ```yaml theme={null} role: "Academic Research Specialist in Emerging Technologies" goal: "Discover and synthesize cutting-edge research, identifying key trends, methodologies, and findings while evaluating the quality and reliability of sources" backstory: "With a background in both computer science and library science, you've mastered the art of digital research. You've worked with research teams at prestigious universities and know how to navigate academic databases, evaluate research quality, and synthesize findings across disciplines. You're methodical in your approach, always cross-referencing information and tracing claims to primary sources before drawing conclusions." ``` ## Crafting Effective Tasks for Your Agents While agent design is important, task design is critical for successful execution. Here are best practices for designing tasks that set your agents up for success: ### The Anatomy of an Effective Task A well-designed task has two key components that serve different purposes: #### Task Description: The Process The description should focus on what to do and how to do it, including: * Detailed instructions for execution * Context and background information * Scope and constraints * Process steps to follow #### Expected Output: The Deliverable The expected output should define what the final result should look like: * Format specifications (markdown, JSON, etc.) * Structure requirements * Quality criteria * Examples of good outputs (when possible) ### Task Design Best Practices #### 1. Single Purpose, Single Output Tasks perform best when focused on one clear objective: **Bad Example (Too Broad):** ```yaml theme={null} task_description: "Research market trends, analyze the data, and create a visualization." ``` **Good Example (Focused):** ```yaml theme={null} # Task 1 research_task: description: "Research the top 5 market trends in the AI industry for 2024." expected_output: "A markdown list of the 5 trends with supporting evidence." # Task 2 analysis_task: description: "Analyze the identified trends to determine potential business impacts." expected_output: "A structured analysis with impact ratings (High/Medium/Low)." # Task 3 visualization_task: description: "Create a visual representation of the analyzed trends." expected_output: "A description of a chart showing trends and their impact ratings." ``` #### 2. Be Explicit About Inputs and Outputs Always clearly specify what inputs the task will use and what the output should look like: **Example:** ```yaml theme={null} analysis_task: description: > Analyze the customer feedback data from the CSV file. Focus on identifying recurring themes related to product usability. Consider sentiment and frequency when determining importance. expected_output: > A markdown report with the following sections: 1. Executive summary (3-5 bullet points) 2. Top 3 usability issues with supporting data 3. Recommendations for improvement ``` #### 3. Include Purpose and Context Explain why the task matters and how it fits into the larger workflow: **Example:** ```yaml theme={null} competitor_analysis_task: description: > Analyze our three main competitors' pricing strategies. This analysis will inform our upcoming pricing model revision. Focus on identifying patterns in how they price premium features and how they structure their tiered offerings. ``` #### 4. Use Structured Output Tools For machine-readable outputs, specify the format clearly: **Example:** ```yaml theme={null} data_extraction_task: description: "Extract key metrics from the quarterly report." expected_output: "JSON object with the following keys: revenue, growth_rate, customer_acquisition_cost, and retention_rate." ``` ## Common Mistakes to Avoid Based on lessons learned from real-world implementations, here are the most common pitfalls in agent and task design: ### 1. Unclear Task Instructions **Problem:** Tasks lack sufficient detail, making it difficult for agents to execute effectively. **Example of Poor Design:** ```yaml theme={null} research_task: description: "Research AI trends." expected_output: "A report on AI trends." ``` **Improved Version:** ```yaml theme={null} research_task: description: > Research the top emerging AI trends for 2024 with a focus on: 1. Enterprise adoption patterns 2. Technical breakthroughs in the past 6 months 3. Regulatory developments affecting implementation For each trend, identify key companies, technologies, and potential business impacts. expected_output: > A comprehensive markdown report with: - Executive summary (5 bullet points) - 5-7 major trends with supporting evidence - For each trend: definition, examples, and business implications - References to authoritative sources ``` ### 2. "God Tasks" That Try to Do Too Much **Problem:** Tasks that combine multiple complex operations into one instruction set. **Example of Poor Design:** ```yaml theme={null} comprehensive_task: description: "Research market trends, analyze competitor strategies, create a marketing plan, and design a launch timeline." ``` **Improved Version:** Break this into sequential, focused tasks: ```yaml theme={null} # Task 1: Research market_research_task: description: "Research current market trends in the SaaS project management space." expected_output: "A markdown summary of key market trends." # Task 2: Competitive Analysis competitor_analysis_task: description: "Analyze strategies of the top 3 competitors based on the market research." expected_output: "A comparison table of competitor strategies." context: [market_research_task] # Continue with additional focused tasks... ``` ### 3. Misaligned Description and Expected Output **Problem:** The task description asks for one thing while the expected output specifies something different. **Example of Poor Design:** ```yaml theme={null} analysis_task: description: "Analyze customer feedback to find areas of improvement." expected_output: "A marketing plan for the next quarter." ``` **Improved Version:** ```yaml theme={null} analysis_task: description: "Analyze customer feedback to identify the top 3 areas for product improvement." expected_output: "A report listing the 3 priority improvement areas with supporting customer quotes and data points." ``` ### 4. Not Understanding the Process Yourself **Problem:** Asking agents to execute tasks that you yourself don't fully understand. **Solution:** 1. Try to perform the task manually first 2. Document your process, decision points, and information sources 3. Use this documentation as the basis for your task description ### 5. Premature Use of Hierarchical Structures **Problem:** Creating unnecessarily complex agent hierarchies where sequential processes would work better. **Solution:** Start with sequential processes and only move to hierarchical models when the workflow complexity truly requires it. ### 6. Vague or Generic Agent Definitions **Problem:** Generic agent definitions lead to generic outputs. **Example of Poor Design:** ```yaml theme={null} agent: role: "Business Analyst" goal: "Analyze business data" backstory: "You are good at business analysis." ``` **Improved Version:** ```yaml theme={null} agent: role: "SaaS Metrics Specialist focusing on growth-stage startups" goal: "Identify actionable insights from business data that can directly impact customer retention and revenue growth" backstory: "With 10+ years analyzing SaaS business models, you've developed a keen eye for the metrics that truly matter for sustainable growth. You've helped numerous companies identify the leverage points that turned around their business trajectory. You believe in connecting data to specific, actionable recommendations rather than general observations." ``` ## Advanced Agent Design Strategies ### Designing for Collaboration When creating agents that will work together in a crew, consider: * **Complementary skills**: Design agents with distinct but complementary abilities * **Handoff points**: Define clear interfaces for how work passes between agents * **Constructive tension**: Sometimes, creating agents with slightly different perspectives can lead to better outcomes through productive dialogue For example, a content creation crew might include: ```yaml theme={null} # Research Agent role: "Research Specialist for technical topics" goal: "Gather comprehensive, accurate information from authoritative sources" backstory: "You are a meticulous researcher with a background in library science..." # Writer Agent role: "Technical Content Writer" goal: "Transform research into engaging, clear content that educates and informs" backstory: "You are an experienced writer who excels at explaining complex concepts..." # Editor Agent role: "Content Quality Editor" goal: "Ensure content is accurate, well-structured, and polished while maintaining consistency" backstory: "With years of experience in publishing, you have a keen eye for detail..." ``` ### Creating Specialized Tool Users Some agents can be designed specifically to leverage certain tools effectively: ```yaml theme={null} role: "Data Analysis Specialist" goal: "Derive meaningful insights from complex datasets through statistical analysis" backstory: "With a background in data science, you excel at working with structured and unstructured data..." tools: [PythonREPLTool, DataVisualizationTool, CSVAnalysisTool] ``` ### Tailoring Agents to LLM Capabilities Different LLMs have different strengths. Design your agents with these capabilities in mind: ```yaml theme={null} # For complex reasoning tasks analyst: role: "Data Insights Analyst" goal: "..." backstory: "..." llm: openai/gpt-4o # For creative content writer: role: "Creative Content Writer" goal: "..." backstory: "..." llm: anthropic/claude-3-opus ``` ## Testing and Iterating on Agent Design Agent design is often an iterative process. Here's a practical approach: 1. **Start with a prototype**: Create an initial agent definition 2. **Test with sample tasks**: Evaluate performance on representative tasks 3. **Analyze outputs**: Identify strengths and weaknesses 4. **Refine the definition**: Adjust role, goal, and backstory based on observations 5. **Test in collaboration**: Evaluate how the agent performs in a crew setting ## Conclusion Crafting effective agents is both an art and a science. By carefully defining roles, goals, and backstories that align with your specific needs, and combining them with well-designed tasks, you can create specialized AI collaborators that produce exceptional results. Remember that agent and task design is an iterative process. Start with these best practices, observe your agents in action, and refine your approach based on what you learn. And always keep in mind the 80/20 rule - focus most of your effort on creating clear, focused tasks to get the best results from your agents. Congratulations! You now understand the principles and practices of effective agent design. Apply these techniques to create powerful, specialized agents that work together seamlessly to accomplish complex tasks. ## Next Steps * Experiment with different agent configurations for your specific use case * Learn about [building your first crew](/en/guides/crews/first-crew) to see how agents work together * Explore [CrewAI Flows](/en/guides/flows/first-flow) for more advanced orchestration # Build with AI Source: https://docs.crewai.com/v1.15.13/en/guides/coding-tools/build-with-ai Everything AI coding agents need to build, deploy, and scale with CrewAI — skills, machine-readable docs, deployment, and enterprise features. # Build with AI CrewAI is AI-native. This page brings together everything an AI coding agent needs to build with CrewAI — whether you're Claude Code, Codex, Cursor, Gemini CLI, or any other assistant helping a developer ship crews and flows. ### Supported Coding Agents This page is designed to be consumed by both humans and AI assistants. If you're a coding agent, start with **Skills** to get CrewAI context, then use **llms.txt** for full docs access. *** ## 1. Skills — Teach Your Agent CrewAI **Skills** are instruction packs that give coding agents deep CrewAI knowledge — how to scaffold Flows, configure Crews, use tools, and follow framework conventions. Anthropic CrewAI skills are available in the **Claude Code plugin marketplace** — the same distribution channel used by top AI-native companies: ```shell theme={null} /plugin marketplace add crewAIInc/skills /plugin install crewai-skills@crewai-plugins /reload-plugins ``` Four skills activate automatically when you ask relevant CrewAI questions: | Skill | When it runs | | ----------------- | -------------------------------------------------------------------------------------------------------------------- | | `getting-started` | Scaffolding new projects, choosing between `LLM.call()` / `Agent` / `Crew` / `Flow`, wiring `crew.jsonc` / `main.py` | | `design-agent` | Configuring agents — role, goal, backstory, tools, LLMs, memory, guardrails | | `design-task` | Writing task descriptions, dependencies, structured output (`output_pydantic`, `output_json`), human review | | `ask-docs` | Querying the live [CrewAI docs MCP server](https://docs.crewai.com/mcp) for up-to-date API details | Works with Claude Code, Codex, Cursor, Gemini CLI, or any coding agent: ```shell theme={null} npx skills add crewaiinc/skills ``` Pulls from the [skills.sh registry](https://skills.sh/crewaiinc/skills). Use either method above — the Claude Code plugin marketplace or `npx skills add`. Both install the official [crewAIInc/skills](https://github.com/crewAIInc/skills) pack. The skill pack teaches your agent: * **Flows** — stateful apps, steps, and crew kickoffs * **Crews & Agents** — JSON-first patterns (`crew.jsonc`, `agents/*.jsonc`), roles, tasks, delegation * **Tools & Integrations** — search, APIs, MCP servers, and common CrewAI tools * **Project layout** — CLI scaffolds and repo conventions * **Up-to-date patterns** — tracks current CrewAI docs and best practices Your agent can now scaffold and build CrewAI projects without you re-explaining the framework each session. How skills work in CrewAI agents — injection, activation, and patterns. Overview of the crewAIInc/skills pack and what it includes. Set up AGENTS.md for Claude Code, Codex, Cursor, and Gemini CLI. Official listing — skills, install stats, and audits. *** ## 2. llms.txt — Machine-Readable Docs CrewAI publishes an `llms.txt` file that gives AI assistants direct access to the full documentation in a machine-readable format. ``` https://docs.crewai.com/llms.txt ``` [`llms.txt`](https://llmstxt.org/) is an emerging standard for making documentation consumable by large language models. Instead of scraping HTML, your agent can fetch a single structured text file with all the content it needs. CrewAI's `llms.txt` is **already live** — your agent can use it right now. Point your coding agent at the URL when it needs CrewAI reference docs: ``` Fetch https://docs.crewai.com/llms.txt for CrewAI documentation. ``` Many coding agents (Claude Code, Cursor, etc.) can fetch URLs directly. The file contains structured documentation covering all CrewAI concepts, APIs, and guides. * **No scraping required** — clean, structured content in one request * **Always up-to-date** — served directly from docs.crewai.com * **Optimized for LLMs** — formatted for context windows, not browsers * **Complements skills** — skills teach patterns, llms.txt provides reference *** ## 3. Deploy to Enterprise Go from a local crew to production on **CrewAI AMP** (Agent Management Platform) in minutes. Scaffold and test your crew or flow: ```bash theme={null} crewai create crew my_crew cd my_crew crewai run ``` Ensure your project structure is ready: ```bash theme={null} crewai deploy --prepare ``` See the [preparation guide](https://docs-platform.crewai.com/platform/en/guides/prepare-for-deployment) for details on project structure and requirements. Push to the CrewAI AMP platform: ```bash theme={null} crewai deploy ``` You can also deploy via [GitHub integration](https://docs-platform.crewai.com/platform/en/guides/deploy-to-amp) or [Crew Studio](https://docs-platform.crewai.com/platform/en/guides/enable-crew-studio). Your deployed crew gets a REST API endpoint. Integrate it into any application: ```bash theme={null} curl -X POST https://app.crewai.com/api/v1/crews//kickoff \ -H "Authorization: Bearer $CREWAI_API_KEY" \ -H "Content-Type: application/json" \ -d '{"inputs": {"topic": "AI agents"}}' ``` Full deployment guide — CLI, GitHub, and Crew Studio methods. Platform overview — what AMP provides for production crews. *** ## 4. Enterprise Features CrewAI AMP is built for production teams. Here's what you get beyond deployment. Detailed execution traces, logs, and performance metrics for every crew run. Monitor agent decisions, tool calls, and task completion in real time. No-code/low-code interface to create, customize, and deploy crews visually — then export to code or deploy directly. Stream real-time events from crew executions to your systems. Integrate with Slack, Zapier, or any webhook consumer. SSO, RBAC, and organization-level controls. Manage who can create, deploy, and access crews across your team. Publish and share custom tools across your organization. Install community tools from the registry. Run CrewAI AMP on your own infrastructure. Full platform capabilities with data residency and compliance controls. AMP is for teams that need to move AI agent workflows from prototypes to production — with observability, access controls, and scalable infrastructure. Whether you're a startup or enterprise, AMP handles the operational complexity so you can focus on building agents. * **Cloud (app.crewai.com)** — managed by CrewAI, fastest path to production * **Factory (self-hosted)** — run on your own infrastructure for full data control * **Hybrid** — mix cloud and self-hosted based on sensitivity requirements Sign up and deploy your first crew to production. # Evaluating Use Cases for CrewAI Source: https://docs.crewai.com/v1.15.13/en/guides/concepts/evaluating-use-cases Learn how to assess your AI application needs and choose the right approach between Crews and Flows based on complexity and precision requirements. ## Understanding the Decision Framework When building AI applications with CrewAI, one of the most important decisions you'll make is choosing the right approach for your specific use case. Should you use a Crew? A Flow? A combination of both? This guide will help you evaluate your requirements and make informed architectural decisions. At the heart of this decision is understanding the relationship between **complexity** and **precision** in your application: Complexity vs. Precision Matrix This matrix helps visualize how different approaches align with varying requirements for complexity and precision. Let's explore what each quadrant means and how it guides your architectural choices. ## The Complexity-Precision Matrix Explained ### What is Complexity? In the context of CrewAI applications, **complexity** refers to: * The number of distinct steps or operations required * The diversity of tasks that need to be performed * The interdependencies between different components * The need for conditional logic and branching * The sophistication of the overall workflow ### What is Precision? **Precision** in this context refers to: * The accuracy required in the final output * The need for structured, predictable results * The importance of reproducibility * The level of control needed over each step * The tolerance for variation in outputs ### The Four Quadrants #### 1. Low Complexity, Low Precision **Characteristics:** * Simple, straightforward tasks * Tolerance for some variation in outputs * Limited number of steps * Creative or exploratory applications **Recommended Approach:** Simple Crews with minimal agents **Example Use Cases:** * Basic content generation * Idea brainstorming * Simple summarization tasks * Creative writing assistance #### 2. Low Complexity, High Precision **Characteristics:** * Simple workflows that require exact, structured outputs * Need for reproducible results * Limited steps but high accuracy requirements * Often involves data processing or transformation **Recommended Approach:** Flows with direct LLM calls or simple Crews with structured outputs **Example Use Cases:** * Data extraction and transformation * Form filling and validation * Structured content generation (JSON, XML) * Simple classification tasks #### 3. High Complexity, Low Precision **Characteristics:** * Multi-stage processes with many steps * Creative or exploratory outputs * Complex interactions between components * Tolerance for variation in final results **Recommended Approach:** Complex Crews with multiple specialized agents **Example Use Cases:** * Research and analysis * Content creation pipelines * Exploratory data analysis * Creative problem-solving #### 4. High Complexity, High Precision **Characteristics:** * Complex workflows requiring structured outputs * Multiple interdependent steps with strict accuracy requirements * Need for both sophisticated processing and precise results * Often mission-critical applications **Recommended Approach:** Flows orchestrating multiple Crews with validation steps **Example Use Cases:** * Enterprise decision support systems * Complex data processing pipelines * Multi-stage document processing * Regulated industry applications ## Choosing Between Crews and Flows ### When to Choose Crews Crews are ideal when: 1. **You need collaborative intelligence** - Multiple agents with different specializations need to work together 2. **The problem requires emergent thinking** - The solution benefits from different perspectives and approaches 3. **The task is primarily creative or analytical** - The work involves research, content creation, or analysis 4. **You value adaptability over strict structure** - The workflow can benefit from agent autonomy 5. **The output format can be somewhat flexible** - Some variation in output structure is acceptable ```python theme={null} # Example: Research Crew for market analysis from crewai import Agent, Crew, Process, Task # Create specialized agents researcher = Agent( role="Market Research Specialist", goal="Find comprehensive market data on emerging technologies", backstory="You are an expert at discovering market trends and gathering data." ) analyst = Agent( role="Market Analyst", goal="Analyze market data and identify key opportunities", backstory="You excel at interpreting market data and spotting valuable insights." ) # Define their tasks research_task = Task( description="Research the current market landscape for AI-powered healthcare solutions", expected_output="Comprehensive market data including key players, market size, and growth trends", agent=researcher ) analysis_task = Task( description="Analyze the market data and identify the top 3 investment opportunities", expected_output="Analysis report with 3 recommended investment opportunities and rationale", agent=analyst, context=[research_task] ) # Create the crew market_analysis_crew = Crew( agents=[researcher, analyst], tasks=[research_task, analysis_task], process=Process.sequential, verbose=True ) # Run the crew result = market_analysis_crew.kickoff() ``` ### When to Choose Flows Flows are ideal when: 1. **You need precise control over execution** - The workflow requires exact sequencing and state management 2. **The application has complex state requirements** - You need to maintain and transform state across multiple steps 3. **You need structured, predictable outputs** - The application requires consistent, formatted results 4. **The workflow involves conditional logic** - Different paths need to be taken based on intermediate results 5. **You need to combine AI with procedural code** - The solution requires both AI capabilities and traditional programming ```python theme={null} # Example: Customer Support Flow with structured processing from crewai.flow.flow import Flow, listen, or_, router, start from pydantic import BaseModel from typing import List, Dict # Define structured state class SupportTicketState(BaseModel): ticket_id: str = "" customer_name: str = "" issue_description: str = "" category: str = "" priority: str = "medium" resolution: str = "" satisfaction_score: int = 0 class CustomerSupportFlow(Flow[SupportTicketState]): @start() def receive_ticket(self): # In a real app, this might come from an API self.state.ticket_id = "TKT-12345" self.state.customer_name = "Alex Johnson" self.state.issue_description = "Unable to access premium features after payment" return "Ticket received" @listen(receive_ticket) def categorize_ticket(self, _): # Use a direct LLM call for categorization from crewai import LLM llm = LLM(model="openai/gpt-4o-mini") prompt = f""" Categorize the following customer support issue into one of these categories: - Billing - Account Access - Technical Issue - Feature Request - Other Issue: {self.state.issue_description} Return only the category name. """ self.state.category = llm.call(prompt).strip() return self.state.category @router(categorize_ticket) def route_by_category(self, category): # Route to different handlers based on category return category.lower().replace(" ", "_") @listen("billing") def handle_billing_issue(self): # Handle billing-specific logic self.state.priority = "high" # More billing-specific processing... return "Billing issue handled" @listen("account_access") def handle_access_issue(self): # Handle access-specific logic self.state.priority = "high" # More access-specific processing... return "Access issue handled" # Additional category handlers... @listen(or_("billing", "account_access", "technical_issue", "feature_request", "other")) def resolve_ticket(self, resolution_info): # Final resolution step self.state.resolution = f"Issue resolved: {resolution_info}" return self.state.resolution # Run the flow support_flow = CustomerSupportFlow() result = support_flow.kickoff() ``` ### When to Combine Crews and Flows The most sophisticated applications often benefit from combining Crews and Flows: 1. **Complex multi-stage processes** - Use Flows to orchestrate the overall process and Crews for complex subtasks 2. **Applications requiring both creativity and structure** - Use Crews for creative tasks and Flows for structured processing 3. **Enterprise-grade AI applications** - Use Flows to manage state and process flow while leveraging Crews for specialized work ```python theme={null} # Example: Content Production Pipeline combining Crews and Flows from crewai.flow.flow import Flow, listen, start from crewai import Agent, Crew, Process, Task from pydantic import BaseModel from typing import List, Dict class ContentState(BaseModel): topic: str = "" target_audience: str = "" content_type: str = "" outline: Dict = {} draft_content: str = "" final_content: str = "" seo_score: int = 0 class ContentProductionFlow(Flow[ContentState]): @start() def initialize_project(self): # Set initial parameters self.state.topic = "Sustainable Investing" self.state.target_audience = "Millennial Investors" self.state.content_type = "Blog Post" return "Project initialized" @listen(initialize_project) def create_outline(self, _): # Use a research crew to create an outline researcher = Agent( role="Content Researcher", goal=f"Research {self.state.topic} for {self.state.target_audience}", backstory="You are an expert researcher with deep knowledge of content creation." ) outliner = Agent( role="Content Strategist", goal=f"Create an engaging outline for a {self.state.content_type}", backstory="You excel at structuring content for maximum engagement." ) research_task = Task( description=f"Research {self.state.topic} focusing on what would interest {self.state.target_audience}", expected_output="Comprehensive research notes with key points and statistics", agent=researcher ) outline_task = Task( description=f"Create an outline for a {self.state.content_type} about {self.state.topic}", expected_output="Detailed content outline with sections and key points", agent=outliner, context=[research_task] ) outline_crew = Crew( agents=[researcher, outliner], tasks=[research_task, outline_task], process=Process.sequential, verbose=True ) # Run the crew and store the result result = outline_crew.kickoff() # Parse the outline (in a real app, you might use a more robust parsing approach) import json try: self.state.outline = json.loads(result.raw) except: # Fallback if not valid JSON self.state.outline = {"sections": result.raw} return "Outline created" @listen(create_outline) def write_content(self, _): # Use a writing crew to create the content writer = Agent( role="Content Writer", goal=f"Write engaging content for {self.state.target_audience}", backstory="You are a skilled writer who creates compelling content." ) editor = Agent( role="Content Editor", goal="Ensure content is polished, accurate, and engaging", backstory="You have a keen eye for detail and a talent for improving content." ) writing_task = Task( description=f"Write a {self.state.content_type} about {self.state.topic} following this outline: {self.state.outline}", expected_output="Complete draft content in markdown format", agent=writer ) editing_task = Task( description="Edit and improve the draft content for clarity, engagement, and accuracy", expected_output="Polished final content in markdown format", agent=editor, context=[writing_task] ) writing_crew = Crew( agents=[writer, editor], tasks=[writing_task, editing_task], process=Process.sequential, verbose=True ) # Run the crew and store the result result = writing_crew.kickoff() self.state.final_content = result.raw return "Content created" @listen(write_content) def optimize_for_seo(self, _): # Use a direct LLM call for SEO optimization from crewai import LLM llm = LLM(model="openai/gpt-4o-mini") prompt = f""" Analyze this content for SEO effectiveness for the keyword "{self.state.topic}". Rate it on a scale of 1-100 and provide 3 specific recommendations for improvement. Content: {self.state.final_content[:1000]}... (truncated for brevity) Format your response as JSON with the following structure: {{ "score": 85, "recommendations": [ "Recommendation 1", "Recommendation 2", "Recommendation 3" ] }} """ seo_analysis = llm.call(prompt) # Parse the SEO analysis import json try: analysis = json.loads(seo_analysis) self.state.seo_score = analysis.get("score", 0) return analysis except: self.state.seo_score = 50 return {"score": 50, "recommendations": ["Unable to parse SEO analysis"]} # Run the flow content_flow = ContentProductionFlow() result = content_flow.kickoff() ``` ## Practical Evaluation Framework To determine the right approach for your specific use case, follow this step-by-step evaluation framework: ### Step 1: Assess Complexity Rate your application's complexity on a scale of 1-10 by considering: 1. **Number of steps**: How many distinct operations are required? * 1-3 steps: Low complexity (1-3) * 4-7 steps: Medium complexity (4-7) * 8+ steps: High complexity (8-10) 2. **Interdependencies**: How interconnected are the different parts? * Few dependencies: Low complexity (1-3) * Some dependencies: Medium complexity (4-7) * Many complex dependencies: High complexity (8-10) 3. **Conditional logic**: How much branching and decision-making is needed? * Linear process: Low complexity (1-3) * Some branching: Medium complexity (4-7) * Complex decision trees: High complexity (8-10) 4. **Domain knowledge**: How specialized is the knowledge required? * General knowledge: Low complexity (1-3) * Some specialized knowledge: Medium complexity (4-7) * Deep expertise in multiple domains: High complexity (8-10) Calculate your average score to determine overall complexity. ### Step 2: Assess Precision Requirements Rate your precision requirements on a scale of 1-10 by considering: 1. **Output structure**: How structured must the output be? * Free-form text: Low precision (1-3) * Semi-structured: Medium precision (4-7) * Strictly formatted (JSON, XML): High precision (8-10) 2. **Accuracy needs**: How important is factual accuracy? * Creative content: Low precision (1-3) * Informational content: Medium precision (4-7) * Critical information: High precision (8-10) 3. **Reproducibility**: How consistent must results be across runs? * Variation acceptable: Low precision (1-3) * Some consistency needed: Medium precision (4-7) * Exact reproducibility required: High precision (8-10) 4. **Error tolerance**: What is the impact of errors? * Low impact: Low precision (1-3) * Moderate impact: Medium precision (4-7) * High impact: High precision (8-10) Calculate your average score to determine overall precision requirements. ### Step 3: Map to the Matrix Plot your complexity and precision scores on the matrix: * **Low Complexity (1-4), Low Precision (1-4)**: Simple Crews * **Low Complexity (1-4), High Precision (5-10)**: Flows with direct LLM calls * **High Complexity (5-10), Low Precision (1-4)**: Complex Crews * **High Complexity (5-10), High Precision (5-10)**: Flows orchestrating Crews ### Step 4: Consider Additional Factors Beyond complexity and precision, consider: 1. **Development time**: Crews are often faster to prototype 2. **Maintenance needs**: Flows provide better long-term maintainability 3. **Team expertise**: Consider your team's familiarity with different approaches 4. **Scalability requirements**: Flows typically scale better for complex applications 5. **Integration needs**: Consider how the solution will integrate with existing systems ## Conclusion Choosing between Crews and Flows—or combining them—is a critical architectural decision that impacts the effectiveness, maintainability, and scalability of your CrewAI application. By evaluating your use case along the dimensions of complexity and precision, you can make informed decisions that align with your specific requirements. Remember that the best approach often evolves as your application matures. Start with the simplest solution that meets your needs, and be prepared to refine your architecture as you gain experience and your requirements become clearer. You now have a framework for evaluating CrewAI use cases and choosing the right approach based on complexity and precision requirements. This will help you build more effective, maintainable, and scalable AI applications. ## Next Steps * Learn more about [crafting effective agents](/en/guides/agents/crafting-effective-agents) * Explore [building your first crew](/en/guides/crews/first-crew) * Dive into [mastering flow state management](/en/guides/flows/mastering-flow-state) * Check out the [core concepts](/en/concepts/agents) for deeper understanding # Build Your First Crew Source: https://docs.crewai.com/v1.15.13/en/guides/crews/first-crew Step-by-step tutorial to create a collaborative AI team with JSON-first crew configuration. ## Build a Research Crew In this guide, you will create a two-agent research crew that gathers information about a topic and writes a markdown report. New crew projects are JSON-first: agents are defined in `agents/*.jsonc`, tasks and crew settings are defined in `crew.jsonc`, and `crewai run` loads the JSON definition directly. ### Prerequisites Before starting, make sure you have: 1. Installed CrewAI following the [installation guide](/en/installation) 2. Set up your LLM API key following the [LLM setup guide](/en/concepts/llms#setting-up-your-llm) 3. A [Serper.dev](https://serper.dev/) API key if you want the researcher to use web search ## Step 1: Create a New Crew ```bash theme={null} crewai create crew research_crew cd research_crew ``` The CLI creates a JSON-first project: ```text theme={null} research_crew/ ├── .gitignore ├── .env ├── agents/ │ └── researcher.jsonc ├── crew.jsonc ├── knowledge/ ├── pyproject.toml ├── README.md ├── skills/ └── tools/ ``` Need the older `crew.py`, `config/agents.yaml`, and `config/tasks.yaml` layout? Create it with `crewai create crew research_crew --classic`. ## Step 2: Define Your Agents Replace the generated `agents/researcher.jsonc` file and add `agents/analyst.jsonc`. The file names are the names you reference from `crew.jsonc`. ```jsonc agents/researcher.jsonc theme={null} { "role": "Senior Research Specialist for {topic}", "goal": "Find comprehensive and accurate information about {topic}, with a focus on recent developments and key insights.", "backstory": "You are an experienced research specialist who organizes complex information into clear, useful notes.", // Replace with your model, for example "openai/gpt-4o". "llm": "provider/model-id", "tools": ["SerperDevTool"], "settings": { "verbose": true, "allow_delegation": false } } ``` ```jsonc agents/analyst.jsonc theme={null} { "role": "Report Analyst for {topic}", "goal": "Turn research findings into a clear, well-structured report.", "backstory": "You are a careful analyst with strong technical writing skills and a talent for extracting useful insights.", // Replace with your model, for example "openai/gpt-4o". "llm": "provider/model-id", "settings": { "verbose": true, "allow_delegation": false } } ``` Replace `provider/model-id` with the model you use, for example `openai/gpt-4o`, `anthropic/claude-sonnet-4-6`, or `gemini/gemini-2.0-flash-001`. ## Step 3: Define Tasks and Crew Settings Replace `crew.jsonc` with: ```jsonc crew.jsonc theme={null} { "name": "Research Crew", "agents": ["researcher", "analyst"], "tasks": [ { "name": "research_task", "description": "Conduct thorough research on {topic}. Focus on key concepts, recent developments, major challenges, notable applications, and future outlook.", "expected_output": "A comprehensive research document with organized sections, specific facts, and useful examples about {topic}.", "agent": "researcher" }, { "name": "analysis_task", "description": "Analyze the research findings and create a polished report on {topic}. Include an executive summary, key insights, trend analysis, and recommendations.", "expected_output": "A professional markdown report with clear headings, a concise summary, main findings, and recommendations.", "agent": "analyst", "context": ["research_task"], "output_file": "output/report.md", "markdown": true } ], "process": "sequential", "verbose": true, "memory": true, "inputs": { "topic": "Artificial Intelligence in Healthcare" } } ``` `context` points to prior task names, so the analyst receives the research task output. The `inputs` object provides default values for `{topic}`. If you remove a default, `crewai run` prompts for it. ## Step 4: Set Environment Variables Open `.env` and add the keys your model and tools need: ```sh theme={null} SERPER_API_KEY=your_serper_api_key # Add your model provider API key here too. ``` See the [LLM setup guide](/en/concepts/llms#setting-up-your-llm) for provider-specific keys. ## Step 5: Install and Run ```bash theme={null} crewai install crewai run ``` `crewai run` detects `crew.jsonc`, loads the agents from `agents/`, prompts for missing placeholders, and runs the crew. When the run finishes, open `output/report.md`. ## How It Works 1. `crew.jsonc` defines the crew, task order, process, memory, and runtime inputs. 2. `agents/researcher.jsonc` and `agents/analyst.jsonc` define the agents. 3. The researcher runs first. 4. The analyst runs second with `context: ["research_task"]`. 5. The final task writes `output/report.md`. ## Extending Your Crew You can add: * More agents by creating new `agents/.jsonc` files and listing them in `crew.jsonc` * More tasks by appending objects to the `tasks` array * Built-in tools by adding tool class names such as `"FileReadTool"` or `"SerperDevTool"` * Custom tools with `"custom:"`, which loads `tools/.py` * Hierarchical execution with `"process": "hierarchical"` and a `manager_llm` or `manager_agent` Only run JSON crew projects from sources you trust. `custom:` tools and `{"python": "module.attribute"}` references execute local Python code when the crew loads. You now have a working JSON-first crew that researches a topic and writes a report. # Conversational Flows Source: https://docs.crewai.com/v1.15.13/en/guides/flows/conversational-flows Build multi-turn chat apps with handle_turn per turn, message history, intent routing, tracing, and WebSocket bridges. ## Overview Conversational apps treat each user line as a **new flow run** with the **same session id**. CrewAI adds helpers for message history, optional intent routing, deferred tracing, UI bridges, and a local `flow.chat()` REPL for conversational flows. | Concept | Implementation | | ------------------ | --------------------------------------------------------------------------------- | | Session id | `handle_turn(..., session_id=...)` → `kickoff(inputs={"id": ...})` → `state.id` | | User line | `handle_turn(message)` appends to `state.messages` before the graph runs | | Turn complete | `FlowFinished` for **this run** only; chat continues on the next `handle_turn` | | Full-session trace | `ConversationConfig(defer_trace_finalization=True)` + `finalize_session_traces()` | ## Turn APIs Use **`flow.handle_turn(message, session_id=...)`** for every user message from REST, WebSocket, tests, and custom UIs. Use **`flow.chat()`** when you want a local terminal chat loop for a conversational `Flow`. `Flow.kickoff()` does **not** accept `user_message=` or `session_id=` keyword arguments. For conversational flows, `handle_turn()` stores the pending message and calls `kickoff(inputs={"id": session_id})` internally after resetting per-turn execution state. | API | Use for | | -------------------------------------- | ------------------------------------------------------------ | | `handle_turn(message, session_id=...)` | Ergonomic one-turn wrapper for conversational `Flow` | | `stream_turn(message, session_id=...)` | Stream one conversational turn as ordered runtime frames | | `chat()` | Local terminal REPL for conversational `Flow` | | `kickoff(inputs={...})` | Advanced flow execution without conversational turn handling | | `ask()` | Blocking prompt **inside** one step (wizard, clarification) | | `@human_feedback` | Approve/reject **a step output** — not the next chat line | | `ChatSession.handle_turn(...)` | Transport layer over `handle_turn` (SSE / WebSocket) | ## Quick start ```python theme={null} from uuid import uuid4 from crewai import Flow from crewai.flow import listen from crewai.experimental.conversational import ( ConversationConfig, ConversationState, ) @ConversationConfig(defer_trace_finalization=True) class SupportFlow(Flow[ConversationState]): conversational = True def route_turn(self, context): message = (self.state.current_user_message or "").lower() if "order" in message: return "order" if "bye" in message or "goodbye" in message: return "goodbye" return "help" @listen("order") def handle_order(self): reply = "Your order is on the way." self.append_assistant_message(reply) return reply @listen("help") def handle_help(self): reply = "How can I help?" self.append_assistant_message(reply) return reply @listen("goodbye") def handle_goodbye(self): reply = "Goodbye!" self.append_assistant_message(reply) return reply session_id = str(uuid4()) flow = SupportFlow() try: flow.handle_turn("Where is my order?", session_id=session_id) flow.handle_turn("What about returns?", session_id=session_id) finally: flow.finalize_session_traces() # one trace link for the whole chat ``` ## Streaming a turn Use `stream_turn()` when a UI or runtime needs structured events for one chat turn. It returns a stream session with ordered frames for Flow routing, LLM chunks, tool activity, and conversation messages. ```python theme={null} stream = flow.stream_turn("Where is my order?", session_id=session_id) with stream: for frame in stream.events: if frame.channel == "llm" and frame.type == "llm_stream_chunk": print(frame.data.get("chunk", ""), end="", flush=True) result = stream.result ``` For the full frame contract, channel list, and async API, see [Streaming Runtime Contract](/edge/en/learn/streaming-runtime-contract). ## Turn lifecycle Each `handle_turn` runs this pipeline: 1. **Turn setup** — stores the pending user message, resolves the session id, resets per-turn execution tracking, and calls `kickoff(inputs={"id": session_id})`. 2. **State restore** — if `inputs["id"]` exists and `@persist` is configured, loads the latest snapshot. 3. **`FlowStarted`** — emitted on the first deferred session turn only. 4. **Pending turn hydration** — appends the user message to `state.messages`, sets `current_user_message` / `last_user_message`, and optionally classifies when `intents` / `default_intents` + `intent_llm` are set. 5. **Graph execution** — `conversation_start` → `route_conversation` → the selected `@listen` handler. 6. **End of run** — per-turn `flow_finished` and trace finalization are **skipped** when deferral is enabled; nested `Agent.kickoff()` / crews do not close the parent batch either. Handlers should call **`append_assistant_message(reply)`** so the next turn’s `conversation_messages` includes assistant text. The user line is already stored by `handle_turn` — do not append it again in handlers. ## `ConversationConfig` (class-level defaults) Decorate your conversational `Flow` subclass with `ConversationConfig`. | Field | Default | Purpose | | -------------------------- | ----------------- | ---------------------------------------------------------------- | | `system_prompt` | Framework default | System message used by the built-in `converse_turn`. | | `llm` | `None` | Conversation LLM used by `converse_turn` and as router fallback. | | `router` | `None` | `RouterConfig` for LLM-driven routing. | | `intent_llm` | `None` | LLM for `intents=` / `default_intents` pre-classification. | | `default_intents` | `None` | Outcome labels for pre-classification. | | `defer_trace_finalization` | `True` | Keep one trace batch open across `handle_turn()` calls. | Override pre-classification per turn with `handle_turn(..., intents=..., intent_llm=...)`. ## Lower-level `ChatState` helpers `ChatState`, `ConversationalConfig`, and `crewai.flow.conversation` helpers are still importable for advanced orchestration, tests, or custom wrappers. They do not add `user_message=` or `session_id=` keyword arguments to `Flow.kickoff()`. ```python theme={null} from crewai.flow import ChatState class MyChatState(ChatState): # Inherited: id, messages, last_user_message, last_intent, session_ready research_turn_count: int = 0 custom_flag: bool = False ``` | Field | Role | | ------------------- | --------------------------------------------------- | | `id` | Session UUID (same as `inputs["id"]`) | | `messages` | `list` of `{role, content}` for LLM history | | `last_user_message` | Latest user line for this turn | | `last_intent` | Route label after classification (if used) | | `session_ready` | One-time bootstrap flag (permissions, caches, etc.) | `ConversationalInputs` is a `TypedDict` for conventional `kickoff(inputs={...})` keys: `id`, `user_message`, `last_intent`. ## `Flow` conversational API ### `handle_turn` parameters | Parameter | Purpose | | ------------------ | ------------------------------------------------------------------------------------------------------- | | `message` | This turn’s text | | `session_id` | Conversation UUID → `inputs["id"]` / `state.id` | | `intents` | Outcome labels for pre-kickoff `classify_intent` | | `intent_llm` | LLM for classification (required with `intents`) | | `**kickoff_kwargs` | Forwarded to `kickoff()` for options like `input_files`, `from_checkpoint`, and `restore_from_state_id` | ### `kickoff` parameters `Flow.kickoff()` accepts `inputs`, `input_files`, `from_checkpoint`, and `restore_from_state_id`. Pass `inputs={"id": session_id}` when you need raw flow execution, but use `handle_turn()` when the call represents a chat message. ### Instance attributes | Attribute | Purpose | | -------------------------- | ----------------------------------------------------------------------- | | `conversational` | Set to `True` to enable the conversational graph and `handle_turn()` | | `defer_trace_finalization` | Instance flag; set automatically from config on `handle_turn()` | | `suppress_flow_events` | Hides console flow panels; **tracing still records** method/flow events | | `stream` | Enable streaming; use with `ChatSession.handle_turn(..., stream=True)` | ### Methods and properties | Name | Description | | -------------------------------------------------------- | ------------------------------------------------------------------ | | `append_assistant_message(content)` | Append a user-visible assistant reply to `state.messages` | | `append_message(role, content, **extra)` | Lower-level append to `state.messages` | | `conversation_messages` | Read-only history for LLM calls | | `classify_intent(text, outcomes, *, llm, context=None)` | Map text to one outcome (same collapse logic as `@human_feedback`) | | `receive_user_message(text, *, outcomes=None, llm=None)` | Append user message; optionally set `last_intent` | | `finalize_session_traces()` | Emit deferred `flow_finished` and finalize the session trace batch | | `_should_defer_trace_finalization()` | Whether this flow defers per-turn trace finalization | | `input_history` | Audit trail of `ask()` prompts and responses | ### Module helpers (`crewai.flow.conversation`) Importable for tests or custom orchestration: | Function | Description | | ---------------------------------------------------------------------------------------------- | ---------------------------------------------- | | `normalize_kickoff_inputs(inputs, user_message=..., session_id=...)` | Merge conversational kwargs into `inputs` | | `get_conversation_messages(flow)` | Read messages from state or internal buffer | | `append_message(flow, role, content, **extra)` | Same as instance method | | `prepare_conversational_turn(flow, user_message=..., intents=..., intent_llm=..., config=...)` | Lower-level turn hydration for custom wrappers | | `receive_user_message(flow, text, ...)` | Same as instance method | | `set_state_field(flow, name, value)` | Set a field on dict or Pydantic state | | `get_conversational_config(flow)` | Read class `conversational_config` | | `input_history_to_messages(entries)` | Convert `input_history` to LLM message format | ## Intent routing patterns ### A. Pre-classify via `ConversationConfig` (simplest) Set `default_intents` and `intent_llm`. Each `handle_turn()` runs classification before routing; read `self.state.last_intent` in `route_turn()`. ### B. Classify inside `route_turn` (richer prompts) Set `default_intents=None` so `handle_turn()` only appends the user message. In `route_turn()`, call `classify_intent` with a custom prompt or descriptions: ```python theme={null} def route_turn(self, context): intent = self.classify_intent( self._routing_prompt(self.state.current_user_message), ("GREETING", "ORDER", "RESEARCH", "GOODBYE"), llm="gpt-4o-mini", ) self.state.last_intent = intent return intent ``` Use **`@listen("RESEARCH")`** (or similar) for steps that run `Agent.kickoff()` with tools — not bare `LLM.call()` — when you need web research or multi-step tool use. ## When the flow finishes but the user keeps chatting `FlowFinished` means **this graph run** completed. The conversation continues with another `handle_turn()` and the same `session_id`. `@persist` restores `messages`, flags, and context. **Persist pattern:** prefer `@persist` on a **single terminal step** (for example `finalize`) rather than on the whole `Flow` class. Class-level persist saves after every method; `load_state` uses the latest row, which may be a mid-run snapshot (for example right after `bootstrap`) and miss handler updates from the same turn. Do **not** use `@human_feedback` for follow-up chat lines unless a human must approve a specific step output before it is shown. ## Conversational `Flow` (experimental) **This is an experimental feature.** The conversational `Flow` surface (`conversational = True`, `handle_turn`, `ConversationConfig`, `RouterConfig`, `ConversationState`, the built-in graph + helpers) lives under `crewai.experimental` and may change shape before it graduates. Pin your CrewAI version if you depend on specific behavior, and watch the changelog for breaking updates. Open issues / feedback welcome. Opt into the conversational chat graph by setting `conversational = True` on a `Flow` subclass. The base `Flow` then ships a built-in `@start` / `@router` / `converse_turn` / `end_conversation` graph, manages `state.messages`, can drive a router LLM, and keeps the trace batch open across turns. You write the **custom routes**; the framework owns the rest. Use this when you want a multi-turn chat with a router and per-route handlers without wiring the lifecycle yourself. Use `Flow[ChatState]` (the lower-level pattern above) when you need full control. ### Quick example ```python theme={null} from crewai import Flow from crewai.flow import listen from crewai.experimental.conversational import ( ConversationConfig, ConversationState, ) @ConversationConfig(defer_trace_finalization=True) class SupportFlow(Flow[ConversationState]): conversational = True def route_turn(self, context: dict) -> str | None: message = (self.state.current_user_message or "").lower() if "search" in message or "news" in message: return "INTERNET_SEARCH" if "docs" in message or "crewai" in message: return "CREWAI_DOCS" return "converse" @listen("INTERNET_SEARCH") def handle_internet_search(self) -> str: """Fresh web research, current news, real-time lookups.""" reply = "I would run the web research route here." self.append_assistant_message(reply) return reply @listen("CREWAI_DOCS") def handle_crewai_docs(self) -> str: """Look up the CrewAI documentation for framework/API questions.""" reply = "I would look up the CrewAI docs here." self.append_assistant_message(reply) return reply flow = SupportFlow() try: flow.handle_turn("What can you do?") # routes to converse flow.handle_turn("Search the web for AI news.") # routes to INTERNET_SEARCH flow.handle_turn("Check the CrewAI docs.") # routes to CREWAI_DOCS finally: flow.finalize_session_traces() ``` For a local terminal chat, use `chat()`: ```python theme={null} def kickoff() -> None: SupportFlow().chat() ``` `chat()` wraps `handle_turn()` in a REPL, exits on `exit` / `quit`, skips blank lines by default, and calls `finalize_session_traces()` when the session ends. ### `ConversationConfig` Class decorator that attaches per-class chat defaults. | Field | Default | Purpose | | ---------------------------- | ----------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------- | | `system_prompt` | `slices.conversational_system_prompt` from i18n | System message used by the built-in `converse_turn`. Pass `""` to opt out entirely. | | `llm` | `None` | Conversation LLM (used by `converse_turn` and as router fallback). | | `router` | `None` | `RouterConfig` for LLM-driven routing. Without it, the flow always falls through to `converse`. | | `answer_from_history_prompt` | Framework default | System message for the optional `answer_from_history` route. | | `answer_from_history_llm` | `None` | Enables the `answer_from_history` short-circuit when set. | | `intent_llm` | `None` | LLM for legacy `intents=`/`default_intents` pre-classification. | | `default_intents` | `None` | Outcome labels for legacy pre-classification. | | `visible_agent_outputs` | `None` | `"all"`, or a list of agent names whose `append_agent_result()` calls should be promoted to public assistant messages. | | `defer_trace_finalization` | `True` | Keep one trace batch open across `handle_turn()` calls. | ### `RouterConfig` and the auto-built route catalog ```python theme={null} from typing import Literal from pydantic import BaseModel from crewai import LLM from crewai.experimental.conversational import RouterConfig class MyRoute(BaseModel): intent: Literal["INTERNET_SEARCH", "CREWAI_DOCS", "converse"] ROUTER_LLM = LLM(model="gpt-4o-mini") router_config = RouterConfig( prompt="Optional domain framing (policy, voice, persona).", response_format=MyRoute, # optional; auto-generated otherwise llm=ROUTER_LLM, # falls back to ConversationConfig.llm routes=["INTERNET_SEARCH", "CREWAI_DOCS"], # optional; inferred from listeners route_descriptions={ "INTERNET_SEARCH": "Override the docstring for this one route.", }, default_intent="converse", # used when LLM call fails or no LLM available fallback_intent="converse", # used when LLM returns an invalid route intent_field="intent", ) ``` The router prompt that gets sent to the LLM is built automatically. For each route the framework picks a description with this precedence: 1. `RouterConfig.route_descriptions[label]` — explicit override. 2. `Flow.builtin_route_descriptions[label]` — framework-canned text for `converse`, `end`, `answer_from_history` (phrased for the router LLM). 3. First non-empty line of the `@listen(label)` handler's docstring. 4. Empty (the route is listed without a description). So in practice, **adding a new route is `@listen("X")` + a one-line docstring**: ```python theme={null} from crewai.flow import listen @listen("INTERNET_SEARCH") def handle_internet_search(self) -> str: """Fresh web research, current news, real-time lookups.""" ... ``` ### Naming handlers The string in `@listen("…")` is a **router route label** (an event name), not the Python method name. Route labels and method completion events share one trigger namespace, so naming a handler the same as its route causes the handler to re-trigger itself in a loop. Use a different method name — the docs examples use a `handle_*` prefix: ```python theme={null} @listen("create_video") def handle_create_video(self) -> str: """User wants a new video.""" ... ``` Do **not** mirror the route label on the method: ```python theme={null} @listen("create_video") def create_video(self) -> str: # rejected at flow instantiation ... ``` …and the router LLM sees: ``` Routes: - CREWAI_DOCS: Look up the CrewAI documentation for framework/API questions. - INTERNET_SEARCH: Fresh web research, current news, real-time lookups. - converse: Ordinary chat, follow-ups, summaries, clarifications… - end: User signals the conversation is finished (goodbye, exit, done). ``` `RouterConfig.prompt` is for **domain framing** (assistant persona, business rules, voice). The route catalog is auto-built — don't list routes in `prompt`; they'll drift the moment you add a handler. ### Built-in routes | Route | Handler | Purpose | | --------------------- | -------------------------- | ------------------------------------------------------------------------------------------------------------------------------------- | | `converse` | `converse_turn` | Default chat handler. Calls `ConversationConfig.llm` with the system prompt + canonical message history. | | `end` | `end_conversation` | Sets `state.ended = True` and emits a terminator reply. | | `answer_from_history` | `answer_from_history_turn` | Optional. Routes here when `ConversationConfig.answer_from_history_llm` is set and the message can be answered from existing history. | You can override any of these by defining a same-named handler in your subclass. ### `handle_turn()` semantics `flow.handle_turn(message)` runs one turn: 1. Resets per-execution tracking (`_completed_methods`, `_method_outputs`) so the graph re-runs — without this, repeated `kickoff` calls on the same flow instance would short-circuit on turn 2+ because `Flow.kickoff_async` treats `inputs={"id": ...}` as a checkpoint restore. 2. Appends the user message to `state.messages`, sets `current_user_message` / `last_user_message`. `last_intent` is **preserved from the prior turn** so the router LLM can use it as a signal. 3. Runs `conversation_start` → `route_conversation` → the chosen `@listen` handler. 4. The router stores its decision in `state.last_intent` (visible to the next turn's router context). 5. If your handler returned a string and didn't already call `append_assistant_message`, `handle_turn` appends it for you. Call `handle_turn()` for chat messages. Calling `kickoff(inputs={"id": ...})` directly runs the flow graph without applying the conversational turn wrapper. ### `chat()` for local REPLs `flow.chat()` is the batteries-included terminal wrapper around `handle_turn()`: ```python theme={null} flow = SupportFlow() flow.chat() ``` It handles the common local loop: 1. Prompts for a user message. 2. Stops on `exit` / `quit`, `EOFError`, or `KeyboardInterrupt`. 3. Calls `handle_turn(message, session_id=...)`. 4. Prints the assistant result. 5. Finalizes deferred session traces in a `finally` block. Customize the terminal behavior with injectable I/O: ```python theme={null} flow.chat( session_id="demo-session", prompt="You: ", assistant_prefix="Assistant: ", exit_commands=("exit", "quit", "bye"), ) ``` For web apps, background workers, tests, and custom transports, keep using `handle_turn()` directly. ### Custom router behavior To run side effects (event bus setup, telemetry) on every routing decision, override `route_turn`: ```python theme={null} from typing import Any from crewai import Flow from crewai.experimental.conversational import ConversationState class SupportFlow(Flow[ConversationState]): conversational = True def route_turn(self, context: dict[str, Any]) -> str | None: self.event_bus = MyBus(self) return super().route_turn(context) ``` To bypass the LLM router entirely and pick a route programmatically, return a string from `route_turn`; returning `None` falls back to `_route_with_config(...)`. ### `append_assistant_message` and `append_agent_result` Inside a `@listen(label)` handler, choose: * `self.append_assistant_message(text)` — adds a user-visible assistant turn to `state.messages`. The next turn's `converse_turn` sees it. * `self.append_agent_result(agent_name, result, visibility="private")` — records a structured event in `state.events` and a thread in `state.agent_threads[agent_name]`. Public visibility also calls `append_assistant_message` for you. Use private results for scratch work that shouldn't pollute the canonical history. `ConversationConfig.visible_agent_outputs` can promote specific agents' private results to public globally (`"all"`, or a list of agent names). ## Tracing across turns With `defer_trace_finalization=True` (default in `ConversationConfig`): * **One trace batch** for the whole chat session. * **`flow_started`** on the first turn only; **`flow_finished`** once in `finalize_session_traces()`. * **Per-turn** `kickoff` does not print “Trace batch finalized”. * **Nested work** (`Agent.kickoff()`, crews, Exa tools) appends to the **parent** batch; inner `AgentExecutor` flows do not close the session batch early. ```python theme={null} flow.chat(session_id=session_id) ``` `flow.chat()` calls `finalize_session_traces()` for you. When you own the loop with `handle_turn()`, call `finalize_session_traces()` when the session ends. `suppress_flow_events=True` only hides Rich console panels; trace and method events still emit for observability. ### Conversational `Flow` trace lifecycle The experimental [conversational `Flow`](#conversational-flow-experimental) uses the same tracing lifecycle: `defer_trace_finalization` defaults to `True`, so each `handle_turn()` keeps the session trace open. Always finalize at the end of the session — wrap your REPL/loop in `try/finally` and call `flow.finalize_session_traces()` on exit. Without it, the trace batch stays open and the final conversation may never export. ## Streaming Set `stream = True` on the `Flow` class. `kickoff(...)` will then emit `assistant_delta` (and related) events through the standard event bus. ## Imports ```python theme={null} from crewai.flow import ( ChatState, ConversationalConfig, ConversationalInputs, Flow, listen, persist, router, start, ) ``` ## See also * [Mastering Flow State Management](/en/guides/flows/mastering-flow-state) — persistence, Pydantic state, `@persist` * [Build Your First Flow](/en/guides/flows/first-flow) — flow basics * Demo: `lib/crewai/runner_conversational_flow_simple.py` — minimal REPL with `RESEARCH` + Exa agent # Build Your First Flow Source: https://docs.crewai.com/v1.15.13/en/guides/flows/first-flow Learn how to create structured, event-driven workflows with precise control over execution. ## Taking Control of AI Workflows with Flows CrewAI Flows represent the next level in AI orchestration - combining the collaborative power of AI agent crews with the precision and flexibility of procedural programming. While crews excel at agent collaboration, flows give you fine-grained control over exactly how and when different components of your AI system interact. In this guide, we'll walk through creating a powerful CrewAI Flow that generates a comprehensive learning guide on any topic. This tutorial will demonstrate how Flows provide structured, event-driven control over your AI workflows by combining regular code, direct LLM calls, and crew-based processing. ### What Makes Flows Powerful Flows enable you to: 1. **Combine different AI interaction patterns** - Use crews for complex collaborative tasks, direct LLM calls for simpler operations, and regular code for procedural logic 2. **Build event-driven systems** - Define how components respond to specific events and data changes 3. **Maintain state across components** - Share and transform data between different parts of your application 4. **Integrate with external systems** - Seamlessly connect your AI workflow with databases, APIs, and user interfaces 5. **Create complex execution paths** - Design conditional branches, parallel processing, and dynamic workflows ### What You'll Build and Learn By the end of this guide, you'll have: 1. **Created a sophisticated content generation system** that combines user input, AI planning, and multi-agent content creation 2. **Orchestrated the flow of information** between different components of your system 3. **Implemented event-driven architecture** where each step responds to the completion of previous steps 4. **Built a foundation for more complex AI applications** that you can expand and customize This guide creator flow demonstrates fundamental patterns that can be applied to create much more advanced applications, such as: * Interactive AI assistants that combine multiple specialized subsystems * Complex data processing pipelines with AI-enhanced transformations * Autonomous agents that integrate with external services and APIs * Multi-stage decision-making systems with human-in-the-loop processes Let's dive in and build your first flow! ## Prerequisites Before starting, make sure you have: 1. Installed CrewAI following the [installation guide](/en/installation) 2. Set up your LLM API key in your environment, following the [LLM setup guide](/en/concepts/llms#setting-up-your-llm) 3. Basic understanding of Python ## Step 1: Create a New CrewAI Flow Project First, let's create a new CrewAI Flow project using the CLI. This command sets up a scaffolded project with all the necessary directories and template files for your flow. ```bash theme={null} crewai create flow guide_creator_flow cd guide_creator_flow ``` This will generate a project with the basic structure needed for your flow. CrewAI Framework Overview ## Step 2: Understanding the Project Structure The generated project has the following structure. The starter embedded crew uses the classic Python/YAML layout, and in Step 4 we will replace the content crew with a JSONC crew. ``` guide_creator_flow/ ├── .gitignore ├── pyproject.toml ├── README.md ├── .env └── src/ └── guide_creator_flow/ ├── __init__.py ├── main.py ├── crews/ │ └── poem_crew/ │ ├── config/ │ │ ├── agents.yaml │ │ └── tasks.yaml │ └── poem_crew.py └── tools/ └── custom_tool.py ``` This structure provides a clear separation between different components of your flow: * The main flow logic in the `src/guide_creator_flow/main.py` file * Specialized crews in the `src/guide_creator_flow/crews` directory * Custom tools in the `src/guide_creator_flow/tools` directory We'll modify this structure to create our guide creator flow, which will orchestrate the process of generating comprehensive learning guides. ## Step 3: Add a Content Writer Crew Our flow will need a specialized crew to handle the content creation process. Let's use the CrewAI CLI to add a content writer crew: ```bash theme={null} crewai flow add-crew content-crew ``` This command automatically creates the necessary directories and template files for your crew. The content writer crew will be responsible for writing and reviewing sections of our guide, working within the overall flow orchestrated by our main application. ## Step 4: Configure the Content Writer Crew Now, let's configure the content writer crew with JSONC. We'll set up two specialized agents - a writer and a reviewer - that collaborate to create high-quality content for our guide. 1. Create `src/guide_creator_flow/crews/content_crew/agents/content_writer.jsonc`: ```jsonc theme={null} { "role": "Educational Content Writer", "goal": "Create engaging, informative content that thoroughly explains the assigned topic and provides valuable insights to the reader.", "backstory": "You are a talented educational writer who explains complex concepts in accessible language and organizes information clearly.", "llm": "provider/model-id", "settings": { "verbose": true } } ``` 2. Create `src/guide_creator_flow/crews/content_crew/agents/content_reviewer.jsonc`: ```jsonc theme={null} { "role": "Educational Content Reviewer and Editor", "goal": "Ensure content is accurate, comprehensive, well-structured, and consistent with previously written sections.", "backstory": "You are a meticulous editor with an eye for detail, clarity, and coherence.", "llm": "provider/model-id", "settings": { "verbose": true } } ``` Replace `provider/model-id` with the model you use, for example `openai/gpt-4o`, `gemini/gemini-2.0-flash-001`, or `anthropic/claude-sonnet-4-6`. 3. Create `src/guide_creator_flow/crews/content_crew/crew.jsonc`: ```jsonc theme={null} { "name": "Content Crew", "agents": ["content_writer", "content_reviewer"], "tasks": [ { "name": "write_section_task", "description": "Write a comprehensive section on the topic: \"{section_title}\".\n\nSection description: {section_description}\nTarget audience: {audience_level} level learners\n\nYour content should begin with a brief introduction, explain key concepts clearly with examples, include practical applications where appropriate, end with a summary, and be approximately 500-800 words.\n\nPreviously written sections:\n{previous_sections}", "expected_output": "A well-structured, comprehensive section in Markdown format that thoroughly explains the topic and is appropriate for the target audience.", "agent": "content_writer", "markdown": true }, { "name": "review_section_task", "description": "Review and improve this section on \"{section_title}\":\n\n{draft_content}\n\nTarget audience: {audience_level} level learners\nPreviously written sections:\n{previous_sections}\n\nFix errors, improve clarity, verify consistency, enhance structure, and add missing key information.", "expected_output": "An improved, polished version of the section that maintains the original structure but enhances clarity, accuracy, and consistency.", "agent": "content_reviewer", "context": ["write_section_task"], "markdown": true } ], "process": "sequential", "verbose": true } ``` The `context` field lets the reviewer use the writer's output. 4. Replace `src/guide_creator_flow/crews/content_crew/content_crew.py` with a small loader: ```python theme={null} from pathlib import Path from crewai.project import load_crew def kickoff_content_crew(inputs: dict): crew, default_inputs = load_crew(Path(__file__).with_name("crew.jsonc")) return crew.kickoff(inputs={**default_inputs, **inputs}) ``` This loader turns `crew.jsonc` into a `Crew` at runtime. While this crew can function independently, in our flow it will be orchestrated as part of a larger system. ## Step 5: Create the Flow Now comes the exciting part - creating the flow that will orchestrate the entire guide creation process. This is where we'll combine regular Python code, direct LLM calls, and our content creation crew into a cohesive system. Our flow will: 1. Get user input for a topic and audience level 2. Make a direct LLM call to create a structured guide outline 3. Process each section sequentially using the content writer crew 4. Combine everything into a final comprehensive document Let's create our flow in the `main.py` file: ```python theme={null} #!/usr/bin/env python import json import os from typing import List, Dict from pydantic import BaseModel, Field from crewai import LLM from crewai.flow.flow import Flow, listen, start from guide_creator_flow.crews.content_crew.content_crew import kickoff_content_crew # Define our models for structured data class Section(BaseModel): title: str = Field(description="Title of the section") description: str = Field(description="Brief description of what the section should cover") class GuideOutline(BaseModel): title: str = Field(description="Title of the guide") introduction: str = Field(description="Introduction to the topic") target_audience: str = Field(description="Description of the target audience") sections: List[Section] = Field(description="List of sections in the guide") conclusion: str = Field(description="Conclusion or summary of the guide") # Define our flow state class GuideCreatorState(BaseModel): topic: str = "" audience_level: str = "" guide_outline: GuideOutline = None sections_content: Dict[str, str] = {} class GuideCreatorFlow(Flow[GuideCreatorState]): """Flow for creating a comprehensive guide on any topic""" @start() def get_user_input(self): """Get input from the user about the guide topic and audience""" print("\n=== Create Your Comprehensive Guide ===\n") # Get user input self.state.topic = input("What topic would you like to create a guide for? ") # Get audience level with validation while True: audience = input("Who is your target audience? (beginner/intermediate/advanced) ").lower() if audience in ["beginner", "intermediate", "advanced"]: self.state.audience_level = audience break print("Please enter 'beginner', 'intermediate', or 'advanced'") print(f"\nCreating a guide on {self.state.topic} for {self.state.audience_level} audience...\n") return self.state @listen(get_user_input) def create_guide_outline(self, state): """Create a structured outline for the guide using a direct LLM call""" print("Creating guide outline...") # Initialize the LLM llm = LLM(model="openai/gpt-4o-mini", response_format=GuideOutline) # Create the messages for the outline messages = [ {"role": "system", "content": "You are a helpful assistant designed to output JSON."}, {"role": "user", "content": f""" Create a detailed outline for a comprehensive guide on "{state.topic}" for {state.audience_level} level learners. The outline should include: 1. A compelling title for the guide 2. An introduction to the topic 3. 4-6 main sections that cover the most important aspects of the topic 4. A conclusion or summary For each section, provide a clear title and a brief description of what it should cover. """} ] # Make the LLM call with JSON response format response = llm.call(messages=messages) # Parse the JSON response outline_dict = json.loads(response) self.state.guide_outline = GuideOutline(**outline_dict) # Ensure output directory exists before saving os.makedirs("output", exist_ok=True) # Save the outline to a file with open("output/guide_outline.json", "w") as f: json.dump(outline_dict, f, indent=2) print(f"Guide outline created with {len(self.state.guide_outline.sections)} sections") return self.state.guide_outline @listen(create_guide_outline) def write_and_compile_guide(self, outline): """Write all sections and compile the guide""" print("Writing guide sections and compiling...") completed_sections = [] # Process sections one by one to maintain context flow for section in outline.sections: print(f"Processing section: {section.title}") # Build context from previous sections previous_sections_text = "" if completed_sections: previous_sections_text = "# Previously Written Sections\n\n" for title in completed_sections: previous_sections_text += f"## {title}\n\n" previous_sections_text += self.state.sections_content.get(title, "") + "\n\n" else: previous_sections_text = "No previous sections written yet." # Run the content crew for this section result = kickoff_content_crew(inputs={ "section_title": section.title, "section_description": section.description, "audience_level": self.state.audience_level, "previous_sections": previous_sections_text, "draft_content": "" }) # Store the content self.state.sections_content[section.title] = result.raw completed_sections.append(section.title) print(f"Section completed: {section.title}") # Compile the final guide guide_content = f"# {outline.title}\n\n" guide_content += f"## Introduction\n\n{outline.introduction}\n\n" # Add each section in order for section in outline.sections: section_content = self.state.sections_content.get(section.title, "") guide_content += f"\n\n{section_content}\n\n" # Add conclusion guide_content += f"## Conclusion\n\n{outline.conclusion}\n\n" # Save the guide with open("output/complete_guide.md", "w") as f: f.write(guide_content) print("\nComplete guide compiled and saved to output/complete_guide.md") return "Guide creation completed successfully" def kickoff(): """Run the guide creator flow""" GuideCreatorFlow().kickoff() print("\n=== Flow Complete ===") print("Your comprehensive guide is ready in the output directory.") print("Open output/complete_guide.md to view it.") def plot(): """Generate a visualization of the flow""" flow = GuideCreatorFlow() flow.plot("guide_creator_flow") print("Flow visualization saved to guide_creator_flow.html") if __name__ == "__main__": kickoff() ``` Let's analyze what's happening in this flow: 1. We define Pydantic models for structured data, ensuring type safety and clear data representation 2. We create a state class to maintain data across different steps of the flow 3. We implement three main flow steps: * Getting user input with the `@start()` decorator * Creating a guide outline with a direct LLM call * Processing sections with our content crew 4. We use the `@listen()` decorator to establish event-driven relationships between steps This is the power of flows - combining different types of processing (user interaction, direct LLM calls, crew-based tasks) into a coherent, event-driven system. ## Step 6: Set Up Your Environment Variables Create a `.env` file in your project root with your API keys. See the [LLM setup guide](/en/concepts/llms#setting-up-your-llm) for details on configuring a provider. ```sh .env theme={null} OPENAI_API_KEY=your_openai_api_key # or GEMINI_API_KEY=your_gemini_api_key # or ANTHROPIC_API_KEY=your_anthropic_api_key ``` ## Step 7: Install Dependencies Install the required dependencies: ```bash theme={null} crewai install ``` ## Step 8: Run Your Flow Now it's time to see your flow in action! Run it using the CrewAI CLI: ```bash theme={null} crewai run ``` When you run this command, you'll see your flow spring to life: 1. It will prompt you for a topic and audience level 2. It will create a structured outline for your guide 3. It will process each section, with the content writer and reviewer collaborating on each 4. Finally, it will compile everything into a comprehensive guide This demonstrates the power of flows to orchestrate complex processes involving multiple components, both AI and non-AI. ## Step 9: Visualize Your Flow One of the powerful features of flows is the ability to visualize their structure: ```bash theme={null} crewai flow plot ``` This will create an HTML file that shows the structure of your flow, including the relationships between different steps and the data that flows between them. This visualization can be invaluable for understanding and debugging complex flows. ## Step 10: Review the Output Once the flow completes, you'll find two files in the `output` directory: 1. `guide_outline.json`: Contains the structured outline of the guide 2. `complete_guide.md`: The comprehensive guide with all sections Take a moment to review these files and appreciate what you've built - a system that combines user input, direct AI interactions, and collaborative agent work to produce a complex, high-quality output. ## The Art of the Possible: Beyond Your First Flow What you've learned in this guide provides a foundation for creating much more sophisticated AI systems. Here are some ways you could extend this basic flow: ### Enhancing User Interaction You could create more interactive flows with: * Web interfaces for input and output * Real-time progress updates * Interactive feedback and refinement loops * Multi-stage user interactions ### Adding More Processing Steps You could expand your flow with additional steps for: * Research before outline creation * Image generation for illustrations * Code snippet generation for technical guides * Final quality assurance and fact-checking ### Creating More Complex Flows You could implement more sophisticated flow patterns: * Conditional branching based on user preferences or content type * Parallel processing of independent sections * Iterative refinement loops with feedback * Integration with external APIs and services ### Applying to Different Domains The same patterns can be applied to create flows for: * **Interactive storytelling**: Create personalized stories based on user input * **Business intelligence**: Process data, generate insights, and create reports * **Product development**: Facilitate ideation, design, and planning * **Educational systems**: Create personalized learning experiences ## Key Features Demonstrated This guide creator flow demonstrates several powerful features of CrewAI: 1. **User interaction**: The flow collects input directly from the user 2. **Direct LLM calls**: Uses the LLM class for efficient, single-purpose AI interactions 3. **Structured data with Pydantic**: Uses Pydantic models to ensure type safety 4. **Sequential processing with context**: Writes sections in order, providing previous sections for context 5. **Multi-agent crews**: Leverages specialized agents (writer and reviewer) for content creation 6. **State management**: Maintains state across different steps of the process 7. **Event-driven architecture**: Uses the `@listen` decorator to respond to events ## Understanding the Flow Structure Let's break down the key components of flows to help you understand how to build your own: ### 1. Direct LLM Calls Flows allow you to make direct calls to language models when you need simple, structured responses: ```python theme={null} llm = LLM( model="model-id-here", # gpt-4o, gemini-2.0-flash, anthropic/claude... response_format=GuideOutline ) response = llm.call(messages=messages) ``` This is more efficient than using a crew when you need a specific, structured output. ### 2. Event-Driven Architecture Flows use decorators to establish relationships between components: ```python theme={null} @start() def get_user_input(self): # First step in the flow # ... @listen(get_user_input) def create_guide_outline(self, state): # This runs when get_user_input completes # ... ``` This creates a clear, declarative structure for your application. ### 3. State Management Flows maintain state across steps, making it easy to share data: ```python theme={null} class GuideCreatorState(BaseModel): topic: str = "" audience_level: str = "" guide_outline: GuideOutline = None sections_content: Dict[str, str] = {} ``` This provides a type-safe way to track and transform data throughout your flow. ### 4. Crew Integration Flows can seamlessly integrate with crews for complex collaborative tasks: ```python theme={null} result = kickoff_content_crew(inputs={ "section_title": section.title, # ... }) ``` This allows you to use the right tool for each part of your application - direct LLM calls for simple tasks and crews for complex collaboration. ## Next Steps Now that you've built your first flow, you can: 1. Experiment with more complex flow structures and patterns 2. Try using `@router()` to create conditional branches in your flows 3. Explore the `and_` and `or_` functions for more complex parallel execution 4. Connect your flow to external APIs, databases, or user interfaces 5. Combine multiple specialized crews in a single flow 6. Build multi-turn chat apps with [Conversational Flows](/en/guides/flows/conversational-flows) (`kickoff` per message, `ChatSession`, deferred tracing) Congratulations! You've successfully built your first CrewAI Flow that combines regular code, direct LLM calls, and crew-based processing to create a comprehensive guide. These foundational skills enable you to create increasingly sophisticated AI applications that can tackle complex, multi-stage problems through a combination of procedural control and collaborative intelligence. # Mastering Flow State Management Source: https://docs.crewai.com/v1.15.13/en/guides/flows/mastering-flow-state A comprehensive guide to managing, persisting, and leveraging state in CrewAI Flows for building robust AI applications. ## Understanding the Power of State in Flows State management is the backbone of any sophisticated AI workflow. In CrewAI Flows, the state system allows you to maintain context, share data between steps, and build complex application logic. Mastering state management is essential for creating reliable, maintainable, and powerful AI applications. This guide will walk you through everything you need to know about managing state in CrewAI Flows, from basic concepts to advanced techniques, with practical code examples along the way. ### Why State Management Matters Effective state management enables you to: 1. **Maintain context across execution steps** - Pass information seamlessly between different stages of your workflow 2. **Build complex conditional logic** - Make decisions based on accumulated data 3. **Create persistent applications** - Save and restore workflow progress 4. **Handle errors gracefully** - Implement recovery patterns for more robust applications 5. **Scale your applications** - Support complex workflows with proper data organization 6. **Enable conversational applications** - Store and access conversation history for context-aware AI interactions For multi-turn chat (`kickoff` per user line, `ChatState`, intent routing, deferred tracing, and `ChatSession`), see [Conversational Flows](/en/guides/flows/conversational-flows). Let's explore how to leverage these capabilities effectively. ## State Management Fundamentals ### The Flow State Lifecycle In CrewAI Flows, the state follows a predictable lifecycle: 1. **Initialization** - When a flow is created, its state is initialized (either as an empty dictionary or a Pydantic model instance) 2. **Modification** - Flow methods access and modify the state as they execute 3. **Transmission** - State is passed automatically between flow methods 4. **Persistence** (optional) - State can be saved to storage and later retrieved 5. **Completion** - The final state reflects the cumulative changes from all executed methods Understanding this lifecycle is crucial for designing effective flows. ### Two Approaches to State Management CrewAI offers two ways to manage state in your flows: 1. **Unstructured State** - Using dictionary-like objects for flexibility 2. **Structured State** - Using Pydantic models for type safety and validation Let's examine each approach in detail. ## Unstructured State Management Unstructured state uses a dictionary-like approach, offering flexibility and simplicity for straightforward applications. ### How It Works With unstructured state: * You access state via `self.state` which behaves like a dictionary * You can freely add, modify, or remove keys at any point * All state is automatically available to all flow methods ### Basic Example Here's a simple example of unstructured state management: ```python theme={null} from crewai.flow.flow import Flow, listen, start class UnstructuredStateFlow(Flow): @start() def initialize_data(self): print("Initializing flow data") # Add key-value pairs to state self.state["user_name"] = "Alex" self.state["preferences"] = { "theme": "dark", "language": "English" } self.state["items"] = [] # The flow state automatically gets a unique ID print(f"Flow ID: {self.state['id']}") return "Initialized" @listen(initialize_data) def process_data(self, previous_result): print(f"Previous step returned: {previous_result}") # Access and modify state user = self.state["user_name"] print(f"Processing data for {user}") # Add items to a list in state self.state["items"].append("item1") self.state["items"].append("item2") # Add a new key-value pair self.state["processed"] = True return "Processed" @listen(process_data) def generate_summary(self, previous_result): # Access multiple state values user = self.state["user_name"] theme = self.state["preferences"]["theme"] items = self.state["items"] processed = self.state.get("processed", False) summary = f"User {user} has {len(items)} items with {theme} theme. " summary += "Data is processed." if processed else "Data is not processed." return summary # Run the flow flow = UnstructuredStateFlow() result = flow.kickoff() print(f"Final result: {result}") print(f"Final state: {flow.state}") ``` ### When to Use Unstructured State Unstructured state is ideal for: * Quick prototyping and simple flows * Dynamically evolving state needs * Cases where the structure may not be known in advance * Flows with simple state requirements While flexible, unstructured state lacks type checking and schema validation, which can lead to errors in complex applications. ## Structured State Management Structured state uses Pydantic models to define a schema for your flow's state, providing type safety, validation, and better developer experience. ### How It Works With structured state: * You define a Pydantic model that represents your state structure * You pass this model type to your Flow class as a type parameter * You access state via `self.state`, which behaves like a Pydantic model instance * All fields are validated according to their defined types * You get IDE autocompletion and type checking support ### Basic Example Here's how to implement structured state management: ```python theme={null} from crewai.flow.flow import Flow, listen, start from pydantic import BaseModel, Field from typing import List, Dict, Optional # Define your state model class UserPreferences(BaseModel): theme: str = "light" language: str = "English" class AppState(BaseModel): user_name: str = "" preferences: UserPreferences = UserPreferences() items: List[str] = [] processed: bool = False completion_percentage: float = 0.0 # Create a flow with typed state class StructuredStateFlow(Flow[AppState]): @start() def initialize_data(self): print("Initializing flow data") # Set state values (type-checked) self.state.user_name = "Taylor" self.state.preferences.theme = "dark" # The ID field is automatically available print(f"Flow ID: {self.state.id}") return "Initialized" @listen(initialize_data) def process_data(self, previous_result): print(f"Processing data for {self.state.user_name}") # Modify state (with type checking) self.state.items.append("item1") self.state.items.append("item2") self.state.processed = True self.state.completion_percentage = 50.0 return "Processed" @listen(process_data) def generate_summary(self, previous_result): # Access state (with autocompletion) summary = f"User {self.state.user_name} has {len(self.state.items)} items " summary += f"with {self.state.preferences.theme} theme. " summary += "Data is processed." if self.state.processed else "Data is not processed." summary += f" Completion: {self.state.completion_percentage}%" return summary # Run the flow flow = StructuredStateFlow() result = flow.kickoff() print(f"Final result: {result}") print(f"Final state: {flow.state}") ``` ### Benefits of Structured State Using structured state provides several advantages: 1. **Type Safety** - Catch type errors at development time 2. **Self-Documentation** - The state model clearly documents what data is available 3. **Validation** - Automatic validation of data types and constraints 4. **IDE Support** - Get autocomplete and inline documentation 5. **Default Values** - Easily define fallbacks for missing data ### When to Use Structured State Structured state is recommended for: * Complex flows with well-defined data schemas * Team projects where multiple developers work on the same code * Applications where data validation is important * Flows that need to enforce specific data types and constraints ## The Automatic State ID Both unstructured and structured states automatically receive a unique identifier (UUID) to help track and manage state instances. ### How It Works * For unstructured state, the ID is accessible as `self.state["id"]` * For structured state, the ID is accessible as `self.state.id` * This ID is generated automatically when the flow is created * The ID remains the same throughout the flow's lifecycle * The ID can be used for tracking, logging, and retrieving persisted states This UUID is particularly valuable when implementing persistence or tracking multiple flow executions. ## Dynamic State Updates Regardless of whether you're using structured or unstructured state, you can update state dynamically throughout your flow's execution. ### Passing Data Between Steps Flow methods can return values that are then passed as arguments to listening methods: ```python theme={null} from crewai.flow.flow import Flow, listen, start class DataPassingFlow(Flow): @start() def generate_data(self): # This return value will be passed to listening methods return "Generated data" @listen(generate_data) def process_data(self, data_from_previous_step): print(f"Received: {data_from_previous_step}") # You can modify the data and pass it along processed_data = f"{data_from_previous_step} - processed" # Also update state self.state["last_processed"] = processed_data return processed_data @listen(process_data) def finalize_data(self, processed_data): print(f"Received processed data: {processed_data}") # Access both the passed data and state last_processed = self.state.get("last_processed", "") return f"Final: {processed_data} (from state: {last_processed})" ``` This pattern allows you to combine direct data passing with state updates for maximum flexibility. ## Persisting Flow State One of CrewAI's most powerful features is the ability to persist flow state across executions. This enables workflows that can be paused, resumed, and even recovered after failures. ### The @persist() Decorator The `@persist()` decorator automates state persistence, saving your flow's state at key points in execution. #### Class-Level Persistence When applied at the class level, `@persist()` saves state after every method execution: ```python theme={null} from crewai.flow.flow import Flow, listen, start from crewai.flow.persistence import persist from pydantic import BaseModel class CounterState(BaseModel): value: int = 0 @persist() # Apply to the entire flow class class PersistentCounterFlow(Flow[CounterState]): @start() def increment(self): self.state.value += 1 print(f"Incremented to {self.state.value}") return self.state.value @listen(increment) def double(self, value): self.state.value = value * 2 print(f"Doubled to {self.state.value}") return self.state.value # First run flow1 = PersistentCounterFlow() result1 = flow1.kickoff() print(f"First run result: {result1}") # Second run - pass the ID to load the persisted state flow2 = PersistentCounterFlow() result2 = flow2.kickoff(inputs={"id": flow1.state.id}) print(f"Second run result: {result2}") # Will be higher due to persisted state ``` #### Method-Level Persistence For more granular control, you can apply `@persist()` to specific methods: ```python theme={null} from crewai.flow.flow import Flow, listen, start from crewai.flow.persistence import persist class SelectivePersistFlow(Flow): @start() def first_step(self): self.state["count"] = 1 return "First step" @persist() # Only persist after this method @listen(first_step) def important_step(self, prev_result): self.state["count"] += 1 self.state["important_data"] = "This will be persisted" return "Important step completed" @listen(important_step) def final_step(self, prev_result): self.state["count"] += 1 return f"Complete with count {self.state['count']}" ``` #### Forking Persisted State `@persist` supports two distinct hydration modes on `kickoff` / `kickoff_async`. Use **resume** (`inputs["id"]`) to continue the same lineage; use **fork** (`restore_from_state_id`) to start a new lineage seeded from a snapshot: | | `state.id` after kickoff | `@persist` writes land under | | ------------------------------ | ------------------------------------- | ----------------------------- | | `inputs["id"]` (resume) | supplied id | supplied id (extends history) | | `restore_from_state_id` (fork) | fresh id, or `inputs["id"]` if pinned | new id (source preserved) | ```python theme={null} from crewai.flow.flow import Flow, start from crewai.flow.persistence import persist from pydantic import BaseModel class CounterState(BaseModel): id: str = "" counter: int = 0 @persist class CounterFlow(Flow[CounterState]): @start() def step(self): self.state.counter += 1 # Run 1: fresh state, counter 0 -> 1 flow_1 = CounterFlow() flow_1.kickoff() # Fork: hydrate from flow_1's latest snapshot, but write under a NEW state.id flow_2 = CounterFlow() flow_2.kickoff(restore_from_state_id=flow_1.state.id) # flow_2 starts with counter=1 (hydrated), then step() bumps it to 2. # flow_1's flow_uuid history is unchanged. ``` Behavior notes: * `restore_from_state_id` not found in persistence → the kickoff falls back silently to default behavior (mirrors the existing `inputs["id"]` resume not-found behavior). No exception is raised. * Combining `restore_from_state_id` with `from_checkpoint` raises a `ValueError` — they target different state systems (`@persist` vs. Checkpointing) and cannot be combined. * `restore_from_state_id=None` (default) is byte-identical to a kickoff without the parameter. * Pinning `inputs["id"]` while forking means the new run shares a persistence key with another flow — usually you want only `restore_from_state_id`. ## Advanced State Patterns ### Conditional starts and resumable execution Flows support conditional `@start()` and resumable execution for HITL/cyclic scenarios: ```python theme={null} from crewai.flow.flow import Flow, start, listen, and_, or_ class ResumableFlow(Flow): @start() # unconditional start def init(self): ... # Conditional start: run after "init" or external trigger name @start("init") def maybe_begin(self): ... @listen(and_(init, maybe_begin)) def proceed(self): ... ``` * Conditional `@start()` accepts a method name, a router label, or a callable condition. * During resume, listeners continue from prior checkpoints; cycle/router branches honor resumption flags. ### State-Based Conditional Logic You can use state to implement complex conditional logic in your flows: ```python theme={null} from crewai.flow.flow import Flow, listen, router, start from pydantic import BaseModel class PaymentState(BaseModel): amount: float = 0.0 is_approved: bool = False retry_count: int = 0 class PaymentFlow(Flow[PaymentState]): @start() def process_payment(self): # Simulate payment processing self.state.amount = 100.0 self.state.is_approved = self.state.amount < 1000 return "Payment processed" @router(process_payment) def check_approval(self, previous_result): if self.state.is_approved: return "approved" elif self.state.retry_count < 3: return "retry" else: return "rejected" @listen("approved") def handle_approval(self): return f"Payment of ${self.state.amount} approved!" @listen("retry") def handle_retry(self): self.state.retry_count += 1 print(f"Retrying payment (attempt {self.state.retry_count})...") # Could implement retry logic here return "Retry initiated" @listen("rejected") def handle_rejection(self): return f"Payment of ${self.state.amount} rejected after {self.state.retry_count} retries." ``` ### Handling Complex State Transformations For complex state transformations, you can create dedicated methods: ```python theme={null} from crewai.flow.flow import Flow, listen, start from pydantic import BaseModel from typing import List, Dict class UserData(BaseModel): name: str active: bool = True login_count: int = 0 class ComplexState(BaseModel): users: Dict[str, UserData] = {} active_user_count: int = 0 class TransformationFlow(Flow[ComplexState]): @start() def initialize(self): # Add some users self.add_user("alice", "Alice") self.add_user("bob", "Bob") self.add_user("charlie", "Charlie") return "Initialized" @listen(initialize) def process_users(self, _): # Increment login counts for user_id in self.state.users: self.increment_login(user_id) # Deactivate one user self.deactivate_user("bob") # Update active count self.update_active_count() return f"Processed {len(self.state.users)} users" # Helper methods for state transformations def add_user(self, user_id: str, name: str): self.state.users[user_id] = UserData(name=name) self.update_active_count() def increment_login(self, user_id: str): if user_id in self.state.users: self.state.users[user_id].login_count += 1 def deactivate_user(self, user_id: str): if user_id in self.state.users: self.state.users[user_id].active = False self.update_active_count() def update_active_count(self): self.state.active_user_count = sum( 1 for user in self.state.users.values() if user.active ) ``` This pattern of creating helper methods keeps your flow methods clean while enabling complex state manipulations. ## State Management with Crews One of the most powerful patterns in CrewAI is combining flow state management with crew execution. ### Passing State to Crews You can use flow state to parameterize crews: ```python theme={null} from crewai.flow.flow import Flow, listen, start from crewai import Agent, Crew, Process, Task from pydantic import BaseModel class ResearchState(BaseModel): topic: str = "" depth: str = "medium" results: str = "" class ResearchFlow(Flow[ResearchState]): @start() def get_parameters(self): # In a real app, this might come from user input self.state.topic = "Artificial Intelligence Ethics" self.state.depth = "deep" return "Parameters set" @listen(get_parameters) def execute_research(self, _): # Create agents researcher = Agent( role="Research Specialist", goal=f"Research {self.state.topic} in {self.state.depth} detail", backstory="You are an expert researcher with a talent for finding accurate information." ) writer = Agent( role="Content Writer", goal="Transform research into clear, engaging content", backstory="You excel at communicating complex ideas clearly and concisely." ) # Create tasks research_task = Task( description=f"Research {self.state.topic} with {self.state.depth} analysis", expected_output="Comprehensive research notes in markdown format", agent=researcher ) writing_task = Task( description=f"Create a summary on {self.state.topic} based on the research", expected_output="Well-written article in markdown format", agent=writer, context=[research_task] ) # Create and run crew research_crew = Crew( agents=[researcher, writer], tasks=[research_task, writing_task], process=Process.sequential, verbose=True ) # Run crew and store result in state result = research_crew.kickoff() self.state.results = result.raw return "Research completed" @listen(execute_research) def summarize_results(self, _): # Access the stored results result_length = len(self.state.results) return f"Research on {self.state.topic} completed with {result_length} characters of results." ``` ### Handling Crew Outputs in State When a crew completes, you can process its output and store it in your flow state: ```python theme={null} @listen(execute_crew) def process_crew_results(self, _): # Parse the raw results (assuming JSON output) import json try: results_dict = json.loads(self.state.raw_results) self.state.processed_results = { "title": results_dict.get("title", ""), "main_points": results_dict.get("main_points", []), "conclusion": results_dict.get("conclusion", "") } return "Results processed successfully" except json.JSONDecodeError: self.state.error = "Failed to parse crew results as JSON" return "Error processing results" ``` ## Best Practices for State Management ### 1. Keep State Focused Design your state to contain only what's necessary: ```python theme={null} # Too broad class BloatedState(BaseModel): user_data: Dict = {} system_settings: Dict = {} temporary_calculations: List = [] debug_info: Dict = {} # ...many more fields # Better: Focused state class FocusedState(BaseModel): user_id: str preferences: Dict[str, str] completion_status: Dict[str, bool] ``` ### 2. Use Structured State for Complex Flows As your flows grow in complexity, structured state becomes increasingly valuable: ```python theme={null} # Simple flow can use unstructured state class SimpleGreetingFlow(Flow): @start() def greet(self): self.state["name"] = "World" return f"Hello, {self.state['name']}!" # Complex flow benefits from structured state class UserRegistrationState(BaseModel): username: str email: str verification_status: bool = False registration_date: datetime = Field(default_factory=datetime.now) last_login: Optional[datetime] = None class RegistrationFlow(Flow[UserRegistrationState]): # Methods with strongly-typed state access ``` ### 3. Document State Transitions For complex flows, document how state changes throughout the execution: ```python theme={null} @start() def initialize_order(self): """ Initialize order state with empty values. State before: {} State after: {order_id: str, items: [], status: 'new'} """ self.state.order_id = str(uuid.uuid4()) self.state.items = [] self.state.status = "new" return "Order initialized" ``` ### 4. Handle State Errors Gracefully Implement error handling for state access: ```python theme={null} @listen(previous_step) def process_data(self, _): try: # Try to access a value that might not exist user_preference = self.state.preferences.get("theme", "default") except (AttributeError, KeyError): # Handle the error gracefully self.state.errors = self.state.get("errors", []) self.state.errors.append("Failed to access preferences") user_preference = "default" return f"Used preference: {user_preference}" ``` ### 5. Use State for Progress Tracking Leverage state to track progress in long-running flows: ```python theme={null} class ProgressTrackingFlow(Flow): @start() def initialize(self): self.state["total_steps"] = 3 self.state["current_step"] = 0 self.state["progress"] = 0.0 self.update_progress() return "Initialized" def update_progress(self): """Helper method to calculate and update progress""" if self.state.get("total_steps", 0) > 0: self.state["progress"] = (self.state.get("current_step", 0) / self.state["total_steps"]) * 100 print(f"Progress: {self.state['progress']:.1f}%") @listen(initialize) def step_one(self, _): # Do work... self.state["current_step"] = 1 self.update_progress() return "Step 1 complete" # Additional steps... ``` ### 6. Use Immutable Operations When Possible Especially with structured state, prefer immutable operations for clarity: ```python theme={null} # Instead of modifying lists in place: self.state.items.append(new_item) # Mutable operation # Consider creating new state: from pydantic import BaseModel from typing import List class ItemState(BaseModel): items: List[str] = [] class ImmutableFlow(Flow[ItemState]): @start() def add_item(self): # Create new list with the added item self.state.items = [*self.state.items, "new item"] return "Item added" ``` ## Debugging Flow State ### Logging State Changes When developing, add logging to track state changes: ```python theme={null} import logging logging.basicConfig(level=logging.INFO) class LoggingFlow(Flow): def log_state(self, step_name): logging.info(f"State after {step_name}: {self.state}") @start() def initialize(self): self.state["counter"] = 0 self.log_state("initialize") return "Initialized" @listen(initialize) def increment(self, _): self.state["counter"] += 1 self.log_state("increment") return f"Incremented to {self.state['counter']}" ``` ### State Visualization You can add methods to visualize your state for debugging: ```python theme={null} def visualize_state(self): """Create a simple visualization of the current state""" import json from rich.console import Console from rich.panel import Panel console = Console() if hasattr(self.state, "model_dump"): # Pydantic v2 state_dict = self.state.model_dump() elif hasattr(self.state, "dict"): # Pydantic v1 state_dict = self.state.dict() else: # Unstructured state state_dict = dict(self.state) # Remove id for cleaner output if "id" in state_dict: state_dict.pop("id") state_json = json.dumps(state_dict, indent=2, default=str) console.print(Panel(state_json, title="Current Flow State")) ``` ## Conclusion Mastering state management in CrewAI Flows gives you the power to build sophisticated, robust AI applications that maintain context, make complex decisions, and deliver consistent results. Whether you choose unstructured or structured state, implementing proper state management practices will help you create flows that are maintainable, extensible, and effective at solving real-world problems. As you develop more complex flows, remember that good state management is about finding the right balance between flexibility and structure, making your code both powerful and easy to understand. You've now mastered the concepts and practices of state management in CrewAI Flows! With this knowledge, you can create robust AI workflows that effectively maintain context, share data between steps, and build sophisticated application logic. ## Next Steps * Experiment with both structured and unstructured state in your flows * Try implementing state persistence for long-running workflows * Explore [building your first crew](/en/guides/crews/first-crew) to see how crews and flows can work together * Check out the [Flow reference documentation](/en/concepts/flows) for more advanced features # Installation Source: https://docs.crewai.com/v1.15.13/en/installation Get started with CrewAI - Install, configure, and build your first AI crew

Coding agent setup

Set up CrewAI in your coding agent

Copy a ready-to-paste setup prompt for Claude Code, Codex, Cursor, or any coding agent. It installs the official CrewAI skills, checks the CLI, and points the agent at the right docs before it edits code.

View coding-agent guide
### Watch: Building CrewAI Agents & Flows with Coding Agent Skills