DEV Community: Pavel Koryagin

Maximize efficiency: CrewAI Task isolation problem

Pavel Koryagin — Wed, 29 Jan 2025 10:48:35 +0000

I’m a TDD fan!

Don’t get me wrong: I don’t perform literally every change in the code by editing a test first. But I’ve mastered running parts individually without the whole product. This saves me time and even money.

Stuff must be tested in isolation

Of course, in the practical approach to CrewAI, the first obvious thing to me was to run tasks individually.

Really, when I start, I don’t know what tasks I need. Will a task as I imagine it work in one shot, or should I split it into many? No one knows until we try. And the same is true for improvement ideas regarding tasks we already have.

Additionally, we tune our internal formats and data transformations, and we debug tools. All these things should be runnable in isolation. And all the cloud vendor connections should be mockable. If they are not, then we’re losing on waiting for irrelevant tasks and LLM calls.

But how?

I am only on the way to assembling a TDD-friendly CrewAI environment. You can check my topic in the CrewAI community and share your tricks, architecture, and code snippets.

Here’re my examples.

A tool-testing script

from dotenv import load_dotenv  
from src.publishing.tools.get_articles_backlog import GetArticlesBacklog  

def main():  
    # Load environment variables  
    load_dotenv()  

    # Initialize the tool  
    backlog_tool = GetArticlesBacklog()  

    print("\n=== Testing Articles Backlog Tool ===")  
    try:  
        results = backlog_tool._run()  
        print(results)  
        print(f"\nFound {len(results)} articles in backlog:")  
        for i, article in enumerate(results, 1):  
            print(f"\nArticle {i}:")  
            for key, value in article.items():  
                print(f"{key}: {value}")  
    except Exception as e:  
        print(f"Error occurred: {str(e)}")  

if __name__ == "__main__":  
    main()

Invoking partial crews

I don’t have an extensive guide on how to run tasks and their combinations individually. I am trying various hacks. They work, but none of them could be recommended as a good practice.

Here’s the extract from crew.py, though:

@CrewBase
class Publishing():
...
    def mock_tasks(self, context: Dict):
        for task_name, prev_result in context.items():
            prev_task = getattr(self, task_name)()
            pydantic_output, json_output = prev_task._export_output(prev_result)
            prev_task.output = TaskOutput(
                name=prev_task.name,
                description=prev_task.description,
                expected_output=prev_task.expected_output,
                raw=prev_result,
                pydantic=pydantic_output,
                json_dict=json_output,
                agent='Mock',
                output_format=prev_task._get_output_format(),
            )

    @crew
    def crew(self, mode: str = None) -> Crew:
        if not mode:
            return Crew(
                agents=self.agents,
                tasks=self.tasks,
                verbose=True,
            )

        if mode == 'ideate':
            return Crew(
                agents=[
                    self.idea_hunter(),
                ],
                tasks=[
                    self.ideate_free(),
                ],
                verbose=True,
            )

        ...    

        if mode == 'research_idea':
            return Crew(
                agents=[
                    self.scientific_researcher(),
                ],
                tasks=[
                    self.prepare_idea_research(),
                    self.research_idea(),
                ],
                verbose=True,
            )

        # An example of a completely detached agent for testing-only purposes
        if mode == 'test_tool':
            tool_tester = Agent(
                role="Tool Tester",
                goal="Test and validate the provided tool's functionality",
                backstory="I am an expert at testing tools and validating their outputs. I ensure tools work as expected. Use word 'dog' as a part of the product name and 'language learning' as domain to search for knowledge.",
                tools=[self.search_in_knowledge_base()],
                verbose=True,
                allow_delegation=False,
            )

            tool_test_task = Task(
                description="Test the {tool} tool by using it and validating its output",
                expected_output="A detailed report of the tool's functionality and performance",
                agent=tool_tester,
            )

            return Crew(
                agents=[tool_tester],
                tasks=[tool_test_task],
                process=Process.sequential,
                verbose=True,
            )

        raise ValueError(f"Unknown mode: {mode}")

This way I can:

Define named sub-sets of tasks.
Mock results of the preceding tasks.

And here’s how it call that Crew in main.py:

def assemble_regular_inputs():
    with open('current_run/input.md', 'r') as f:
        input = f.read()

    return {
        # Stuff for very different tasks is mixed together,
        # because CrewAI interpolates all the task strings, not only the used ones.
        'session_date': '2025-01-12',
        'task': 'Inspect content plan, suggest ideas to fill the gaps',
        'calendar_principles': get_publication_calendar_principles(),
        'input': input,
    }

def run():
    """
    Run the crew.
    """
    agentops.init()

    cmd = os.getenv('RUN_MODE')

    if not cmd:
        inputs = assemble_regular_inputs()
        Publishing().crew().kickoff(inputs=inputs)
        return

    if cmd == 'ideate':
        inputs = assemble_regular_inputs()
        Publishing().crew(mode='ideate').kickoff(inputs=inputs)
        return

    ...

    if cmd.startswith('test_tool:'):
        tool_name = cmd.split(':', 1)[1]
        inputs = {'tool': tool_name}
        Publishing().crew(mode='test_tool').kickoff(inputs=inputs)
        return

    raise ValueError(f"Unknown command: {cmd}")

This post is rather a heads-up that CrewAI won’t make your crew test-friendly in traditional sense. It comes with its own “test” concept that is fantastic, but is not a replacement for traditional test engineering.

The end of the series

As you could see, I dive deep into CrewAI, but also found some problems I’d prefer not to have.

I plan to continue my AI-agents the following flows:

Keep using CrewAI for at least one of my projects.
Try to adapt a schema, where CrewAI is a node in LangGraph.
Study new frameworks.

Happy building!

Flow vs. Crew: The clash of CrewAI templates

Pavel Koryagin — Tue, 28 Jan 2025 10:38:46 +0000

Welcome to the least informative post of the whole series! 😁 Just a small heads-up. The really complex topic is for tomorrow.

These two commands create entry points that are structured very differently:

crewai create crew <name>
crewai create flow <name>

create crew version main.py:

def run():
    ...

def train():
    ...

def replay():
    ...

def test():
    ...

# No `if __name__ == "__main__"` statement at all

create flow version main.py:

def kickoff():
    poem_flow = PoemFlow()
    poem_flow.kickoff()

def plot():
    poem_flow = PoemFlow()
    poem_flow.plot()

if __name__ == "__main__":
    kickoff()

The difference is that the flow version is structured as a script entry point and the crew version is nonfunctional without the crewai run command.

Additionally, the crew version exports some interesting entrypoints, while the flow version does not. Not sure why—maybe because they design that the crews should be trained independently.

Stay tuned

In the next post: CrewAI hurts TDD fans! Let’s invent a cure.

CrewAI’s output_pydantic is almost perfect. Here’s the missing part

Pavel Koryagin — Mon, 27 Jan 2025 11:13:43 +0000

output_pydantic is awesome and guarantees the output format to you. But its implementation is missing an important obvious thing.

If you use monitoring, you could make interesting discoveries, and one of mine is this.

When you use output_pydantic, the LLM does not see your field descriptions.

Code:

class FillContentPlanGapsOutput(BaseModel):
    date: str
    time: str
    channel: str = Field(..., description="Channel ID. Lowercase name, e.g. twitter, youtube, etc.")
    typeOfContent: str

LLM prompt from monitoring:

Here’s my topic where I approach the solutions.

And here’s a very early draft of a solution:

@task
def fill_content_plan_gaps(self) -> Task:
    field_info = "\nOutput fields:\n"
    for field_name, field_instance in FillContentPlanGapsOutput.model_fields.items():
        field_info += "- " + field_name + ((": " + field_instance.description) if field_instance.description is not None else "") + "\n"

    return Task(
        config=self.tasks_config['fill_content_plan_gaps'],
        expected_output=self.tasks_config['fill_content_plan_gaps']['expected_output'] + field_info,
        output_pydantic=FillContentPlanGapsOutput,
    )

Logs:

This is the expect criteria for your final answer: List of content requests with channel, content type, and time slot.

Output fields:
- date
- time
- channel: Channel ID. Lowercase name, e.g. twitter, youtube, etc.
- typeOfContent

you MUST return the actual complete content as the final answer, not a summary.
Ensure your final answer contains only the content in the following format: {
  "date": str,
  "time": str,
  "channel": str,
  "typeOfContent": str
}

When you read it, the problem might be already fixed in the framework. Otherwise, make a generic implementation of the prompt customizer that I haven’t made yet.

Stay tuned

In the next post: No revolutionary observations, just a heads-up about a discrepancy in official boilerplate versions.

The dual nature of a CrewAI Task. Have you overlooked it too?

Pavel Koryagin — Fri, 24 Jan 2025 15:49:29 +0000

Tasks in Crew AI are not what they seem. There’s more to them.

Let’s take a look at a task definition.

We have a separate YAML file with string templates where the values are yet to be interpolated. This YAML file goes into a Python object of class Task.

What do I (and, as it turns out, others too) expect when we see this? The contents of the Task object seem to be a definition of the task, right? With experience in things like n8n or Redux, we expect that the actual data will flow somewhere in memory, according to this definition, and will be returned by kickoff() in the end.

The reality is different.

Yes, it holds the template strings yet to be interpolated. But it also has the output property of type TaskOutput that gets populated at runtime when the task has reached its final answer.

You can access the output value by reading it on the respective task object.

researcher = Agent(
    config=agents_config['researcher'],
)

research_task = Task(
    config=tasks_config['research_task'],
    agent=researcher,
)

crew = Crew(
    agents=[researcher],
    tasks=[research_task],
    verbose=True,
)

crew.kickoff(inputs=inputs) # Ignoring the return value  

# Reading the output on the task object
print(research_task.output.raw)
# print(research_task.output.json)
# print(research_task.output.to_dict())
# print(research_task.output.pydantic)

Look, in this example, I don’t take the result from the kickoff function. Instead, it is accessed on the task object.

With decorators, it also works, although it looks even more cryptic because you extract the runtime data from a function call that instantiates a Task object... but alright, I complained about the decorators two days ago.

crew_base = SampleCrew()

crew_base.crew().kickoff(inputs=inputs) # Ignoring the return value  

# Reading the output on the task object
print(crew_base.research_task().output.raw)
# print(crew_base.research_task().output.json)
# print(crew_base.research_task().output.to_dict())
# print(crew_base.research_task().output.pydantic)

Knowing this, we can better predict how our crew works:

Every task is always a single instance per execution.
- Maybe in the future, they will make the tasks optional.
- It is impossible to make them plural without breaking the contract.
When you are puzzled about which result you are actually getting from your crew, use your_task.output.raw/json/pydantic for granular control, rather than the result of kickoff().
With decorators, you have to create different instances of "CrewBase" for separate runs. There are other bugs with multiple instances, but it mostly works.

And I cannot say this is not reflected in the docs. It is:

But to me, this doc makes sense only when I already know how it works. Before that, these parts felt out of context. The courses did not help either. And those Task methods/properties do not exist today, maybe they are from TaskOutput or reserved for the future?

Stay tuned

In the next post: output_pydantic is awesome! But you have to patch the framework.

CrewAI Agent is not a team player. It craves a conveyor, not a crew

Pavel Koryagin — Thu, 23 Jan 2025 16:11:10 +0000

I like CrewAI, but I must admit its name has fooled me.

Problems with AI frameworks start before you’ve even chosen one. There are a ton of frameworks already, so we guess their properties based on very fragmented knowledge.

How did the name of CrewAI fool me? I imagined agents interacting like a crew, but in practice, it turned out to be built for sequential processes, and the behavior that you expect for a virtual crew—resembling a human crew—is either not working or is absent completely.

What is already available

Objecting to this thesis, CrewAI offers asynchronous tasks and hierarchical processes, but these features are underwhelming.

Asynchronous tasks only optimize execution time without affecting the outcomes.

The hierarchical process is buggy and poorly documented. For instance, tasks calling other tasks often produce errors, and the manager meant to dispatch tasks sometimes mishandles them. Telemetry reveals bizarre interactions, like a Frankenstein-style combination of a task definition with a wrong agent. Join my topic to see the research I conducted.

What is a “crew,” according to my intuition

The agents should decide what they do and what they don’t. The tasks and task dependencies should be a set of best practices rather than a strict to-do list. Agents or their dispatcher should decide which sub-tasks (or predefined processes) are relevant to the current request and which are not.

This isn’t available right now in CrewAI. All the tasks in a crew are mandatory. The sequential process is the only type that works reliably.

Can I implement it manually?

Look, the process in CrewAI is a sort of state machine. And CrewAI also comes with an explicit state machine—the Flow class. It might be the base for DIY solutions.

Wrapping up, if you want something more complex than a sequential process with CrewAI, implement it yourself. Do you know a framework that allows agents to decide which task definitions are relevant to the current user request? Please share.

Stay tuned

In the next post: You are probably getting CrewAI’s Task object wrong, as I did. Let’s clarify how it physically works.

The evil decorators of CrewAI. Fight or flight?

Pavel Koryagin — Wed, 22 Jan 2025 12:11:33 +0000

After completing the course “Multi AI Agent Systems with crewAI,” mentioned earlier, you’re confident that CrewAI is a piece of cake, and you’re ready to start solving your own tasks! That’s the moment where the first obstacle arises.

To start, you need to create a project locally, right? But the course didn’t teach you how to do that.

You decide to follow the official Quickstart guide to set up your local project, but you see a discrepancy with the course: YAML, a class, and decorators.

This is a limbo state: “Multi AI Agent Systems with crewAI” does not show decorators at all, then “Practical Multi AI Agents and Advanced Use Cases with crewAI” shows YAML with manual processing, and demos like “How To Create AI Agents From Scratch” show them without much explanation.

So what to do?

Option 1. My Quickstart

Recommended for your first project right after completing the course “Multi AI Agent Systems with crewAI.” This option uses a plain and straightforward crew boilerplate.

Make a copy of this directory: instagram_post example
Initialize a git repo and make the initial commit.
Create an empty .env file.
In pyproject.toml:
- Update crewai to the latest version,
- Add crewai_tools.
- Run poetry lock,
- Resolve version conflicts and execute poetry lock again.
Run poetry install --no-root.
Find self.llm = Ollama(model=os.environ['MODEL']) and replace it with a provider-agnostic version self.llm = LLM(model=os.environ['MODEL']), where LLM is imported from crewai.
Do not try to run it as it is.
- If you do, you’ll have to resolve several dependencies (run Ollama locally, obtain two API keys, install a package missing in pyproject.toml). These complications are not what you need when you start a project.
Instead, start editing the tasks and agents to your needs right at this point.
- Start small (one task and one agent; not all the tasks you imagine at once).
- Fill the .env file, using your knowledge from the course and the providers you’re already familiar with.
- Either delete .env.example or synchronize it with your version of .env.
- The same is for README.md–delete or adjust.
Run python main.py.
Fix errors.
Enjoy yourself.

At this point, I’d recommend adding monitoring as the previous post suggests.

Option 2. Official Quickstart

Do this only when you have solved a couple of your own problems via Option 1 and gained confidence in using CrewAI elements.

Take the official Quickstart
Ensure that the process with all the tasks at once works.
Fight with problems as you add something a bit less trivial.

I did many experiments of code organization with decorators to avoid the problems I’m sharing below, but so far, I have no robust solution to share. I will show you some hacks in snippets in the post “Running tools and tasks individually or in subsets” of this series.

This is it! But you probably want to know why I don’t like decorators? There’s more than just discrepancy with the course.

The good, the bad, and the fragile

YAML idea is okay, generally intuitive, and maybe even neat. But in order to make it work, you need to do some implicit transformations.

In plain code, you cannot reference an agent by name:

the_task = Task(
    ...
    agent=the_agent # A variable, this works
)
the_task = Task(
    ...
    agent='the_agent' # A string id, does not work
)

The same is true for tools. But with YAML, this is exactly what you do. To make it work, CrewAI introduces some decorators...

At first glance, those five decorators look alright (@tool, the 5th, is not in the screenshot). The caveat is that they introduce unexpected behaviors. The basic sequential task flow with a complete crew as in the boilerplate works, but any deviation results in cryptic error messages.

As a software architect, I insist on testing app components separately (and there is a post on it in this series down the road). However, creating subsets of tasks for focused testing is problematic with CrewAI decorators.

If you create a separate crew function for testing, it fails because the agents list hasn’t been initialized by the decorator.
Adding a parameter to the API might work, but only for scalar values.
Passing more complex data structures results in cryptic errors:

Note that things might fail in a different way if you’re reading this post months later than it was written.

And one more thing: the problem is not in the “decorator” pattern per se! The problematic ones are these five (CrewBase, agent, task, tool, crew). As opposed to this, you could find a good example of decorators in Flow of the same framework. They are elegant, flexible, and I haven’t faced any problem with them so far.

So, if you want full control or simply fewer surprises, consider building your own architecture instead of relying on CrewAI’s default setup. In this series, down the road, you will find more solutions to incorporate.

Stay tuned

In the next post: CrewAI’s name has fooled me. Let’s see what its Process variants can and what they cannot.

Add monitoring ASAP! Smooth out your dev experience with CrewAI

Pavel Koryagin — Tue, 21 Jan 2025 13:36:10 +0000

With AI, our app’s quality is becoming less predictable than ever. In this post, we’re going to reduce the frameworks’ contribution to the chaos.

We choose CrewAI for its unique prompt crafting and some hard-coded logic. These features make it stand out, but their opacity is also a profound source of frustration and misconceptions.

When the agent produces incorrect or even nonsensical conclusions, it’s not always clear whether it’s an LLM hallucination, missing context, or a bug in the prompt-building code (all three are real.) These cases assume very different mitigation actions, so we need to know for sure what’s actually happening inside the agent.

Another challenge is designing your agent crew. Questions like how data from one task will transfer to another, whether tasks will share the same chat history, and similar issues are all confusing. Documentation is also vague on these topics, so we need an X-ray vision to see the reality.

    verbose=True

Verbose mode is insufficient in CrewAI, but there’s a superior method—monitoring (aka telemetry, aka observability.) For experienced developers, the need for observability is obvious. For newcomers, CrewAI might become your first experience with this kind of tool.

Let’s start. I won’t dig here into how to connect telemetry—just check the relevant sections on the CrewAI website. All these one-page guides are simple and straightforward.

Instead, I’ll show you how to use it. I started with AgentOps, simply because it was the first on the list. Although I'm disappointed and will try the alternatives, it is still a fine option to start with.

Here’s a screenshot of my session. When I run crewai run, I get a link at the beginning and end of the session.

Clicking the link opens a screen where chats are rendered per agent and separately per LLM call.

Here's my take on its UI. Don’t rely on the rendered chats—they are misleading. Instead, hover over a block in the LLM and tools diagram of the "Session Replay" tab, click to freeze the selection, and navigate to the “Raw JSON” tab.

This is the true representation of the request sent to the LLM, revealing how CrewAI actually works under the hood in your particular case. Isn't that awesome to access this secret?

For example, by pressing [Cmd+F], I can search for expected strings from previous tasks. This allows me to verify how tools and task data are being fed into the LLM. See how they are wrapped into the prompts and does their formatting conflict with other parts of the prompt.

You've got the point: with AI agents, monitoring is not for production only. It’s eye-opening to see the actual source of unexpected LLM responses.

You should enable monitoring asap even locally, to make debugging straightforward. Fortunately, it is simple.

Stay tuned

In the next post: Decorators in CrewAI frustrate the beginners. I am going to smooth things over on your learning journey.

I’ve learned CrewAI. Now I want to make your learning smoother

Pavel Koryagin — Mon, 20 Jan 2025 21:30:44 +0000

Is CrewAI friendly to fast-learners and software architects?

Let me tell you my story. After some struggles in adapting LangChain with fast-learner techniques, I decided to take a traditional course for the next thing I’m going to learn.

The next thing to try was CrewAI, and Google brought me the course “Multi AI Agent Systems with crewAI.” It was a great learning experience, and CrewAI’s neat, straightforward approach really won me over. It felt like that moment when you realize you’ve finally found a tool that you can just take and use.

But of course, nothing’s ever that simple. Once I started building my own project, I hit roadblock after roadblock—turns out there’s a big difference between reading about how something works and actually getting it to work. Through plenty of trial, error, and dig-the-sources, I picked up a bunch of insights I wish I’d had from the start.

That’s why I’m kicking off this series of posts. I want to share with you what I’ve learned in practice that either didn’t catch enough attention when I was learning or is not properly documented at all.

This series is especially for those of you who’ve started tinkering with CrewAI and found yourself wondering, “Now what?”

But if you’re just starting out, then this first post is for you. As a fast-learner by spirit and a dev mentor, I continuously look for ways to turn my experience into an optimal learning plan.

My version of the shortest way to master CrewAI

Here are the three mandatory and two optional steps.

Step 1. 🎓 Take the free course Multi AI Agent Systems with crewAI. I started with it, and I see nothing better. It is from the CrewAI creator.

Step 2. 📃 Read through this my series of posts. Apparently 😊

Step 3. 🔨 Start your own real project with CrewAI. Better yet, kick it off while you’re still doing the course from step 1.

However, there’s a critical discrepancy between their course and their boilerplate, discussed in my post “Decorators.” Recipe what to do is also there.

Optional step 4. Complete (or skim over) another free course of the same author Practical Multi AI Agents and Advanced Use Cases with crewAI.

Optional step out of the order. If you’re like me and enjoy podcast-style demonstrations, consider watching this at any stage of your learning: How To Create AI Agents From Scratch (CrewAI, Zapier, Cursor).

Stay tuned

In the next post: How to transform CrewAI debugging from guesswork into engineering and peek under the hood of your AI agents.