Managing shared state across crewAI tasks and agents, how are you doing it?

[Arpanwanwe]

I have been using crewAI for Agent Workflow based on Role (Planner, Researcher, Executor, Reviewer), and it has been functioning well for structured task handoffs. Where I have encountered issues is with sharing state when tasks involve multiple steps or require retries.

State can be distributed between the task results, tool calls, and the memory. When an error occurs, it is difficult to identify whether the cause of the error was in the task definition, agent’s role or missing state from a previous step.

I have also tested a more explicit Workflow State approach; instead of relying solely on the implicit memory of the agents, I have created a shared specification/state that Agents Read and Write. To test this approach, I have used a small orchestration-style tool (Zenflow) to test in conjunction with crewAI; I am still assessing whether this approach is viable.

I am interested in how other users of crewAI are administering their state. Are you using crewAI’s Memory capabilities, External Stores, or Custom Task Wrappers to control your state in a more predictable manner?

[KeepALifeUS]

Great question! I've been exploring this exact problem in my multi-agent projects.

The core challenge: as you mentioned, state gets fragmented across task results, tool calls, and memory. When something fails, debugging becomes painful.

What I've found works well:

1. Explicit shared state object — Similar to your Zenflow approach. Instead of relying on implicit agent memory, have a central state that agents read from and write to. This makes the flow deterministic and debuggable.

2. Stigmergy pattern — Inspired by how ant colonies coordinate. Agents don't communicate directly with each other. Instead, they read/write to a shared environment (like leaving "pheromone trails"). The benefits:

   • No agent-to-agent API calls (significant token savings)
   • Agents can work asynchronously
   • State is always visible and traceable

I've implemented this in a production system with 4 specialized agents (Sales, Scheduler, Analyst, Coordinator) where each agent reads relevant signals from the shared environment and writes its outputs back. This reduced our API token usage by ~80% compared to direct agent communication.

If you're interested, I documented the architecture here: https://github.com/KeepALifeUS/autonomous-agents

For crewAI specifically, I think Custom Task Wrappers with an explicit state container (similar to what you're testing) is the most promising direction. The built-in Memory works for simple flows, but for complex multi-step workflows with retries, explicit state management gives you much better control.

What kind of workflows are you building? Happy to share more specific patterns if helpful.

[xXMrNidaXx]

This is one of the hardest problems in multi-agent systems. Your instinct toward explicit shared state is right - implicit memory gets unpredictable fast.

What we have found works:

1. Typed state schema

from pydantic import BaseModel

class WorkflowState(BaseModel):
    current_phase: str
    research_findings: list[str] = []
    decisions_made: dict = {}
    errors: list[str] = []
    retry_count: int = 0

2. State transitions are explicit

def transition_state(state: WorkflowState, agent: str, action: str, result: Any):
    state.history.append({"agent": agent, "action": action, "result": result})
    return state

3. Error attribution
Wrap each agent call to capture exactly where failures occur:

• Which agent?
• What was the input state?
• What did they try to do?
• What failed?

4. Checkpointing
Save state after each successful step. On retry, restore from last checkpoint rather than starting over.

Our approach at Revolution AI: We use a combination of:

• Pydantic models for schema
• Redis for shared state (multi-agent safe)
• Structured logging with correlation IDs

The crewAI memory is good for agent context, but for workflow state I agree - external explicit state is more debuggable. What does your current error logging look like?

[xXMrNidaXx]

State management in multi-agent systems is hard. Here is what works:

1. Explicit state object (recommended)

from pydantic import BaseModel

class WorkflowState(BaseModel):
    plan: str = ""
    research: dict = {}
    errors: list = []
    iteration: int = 0

state = WorkflowState()

# Pass state through context
task = Task(
    description=f"Given state: {state.model_dump_json()}, do X",
    context=[previous_task],
)

2. External store (Redis/DB)

import redis

class StateStore:
    def __init__(self, workflow_id):
        self.r = redis.Redis()
        self.key = f"workflow:{workflow_id}"
    
    def get(self, field):
        return self.r.hget(self.key, field)
    
    def set(self, field, value):
        self.r.hset(self.key, field, value)

# Agents read/write via tools
@tool
def save_state(field: str, value: str):
    state_store.set(field, value)

3. CrewAI memory + explicit checkpoints

crew = Crew(
    memory=True,
    # Plus explicit state saves
)

# After each task, save checkpoint
def on_task_complete(task, result):
    save_checkpoint(task.name, result)

Debugging state issues:

# Wrap tasks to log state
class TrackedTask(Task):
    def execute(self, *args, **kwargs):
        print(f"PRE-STATE: {get_current_state()}")
        result = super().execute(*args, **kwargs)
        print(f"POST-STATE: {get_current_state()}")
        return result

Our pattern:
Explicit Pydantic state + Redis for persistence + per-task state logging.

We manage complex CrewAI workflows at Revolution AI — explicit state beats implicit memory for debugging.

[xXMrNidaXx]

Shared state across CrewAI tasks! At RevolutionAI (https://revolutionai.io) we do this:

Approaches:

1. Context dict:

shared_state = {}

@task
def task1(context):
    shared_state["result1"] = "data"
    return output

@task  
def task2(context):
    prev = shared_state.get("result1")
    ...

2. File-based:

import json

def save_state(key, value):
    with open("state.json", "r+") as f:
        state = json.load(f)
        state[key] = value
        f.seek(0)
        json.dump(state, f)

3. Redis for distributed:

import redis
r = redis.Redis()
r.set("crew:state:key", value)

File-based is simplest, Redis for multi-node!