Any suggestions why agents cant seem to delegate properly?

[rschoos]

Love crewai, but have a hard time making it work (it runs... but making it work...)
not so much an expert with python (using javascript/typescript mostly), but with the help of claude and gpt4o all works out well...

it tuns... and i like the dynamic setup of things. But there seems to be some 'configuration' options (or limitations i am not sure how to deal with) - perhaps using the wrong models or need to specify some prompt templates, but in all examples it works out of the box. And it did work for a short period of time better than now. so i messed up somehing...

hope the message and explanation is familiar and enough to give to some pointers :-)

Any feedback or tips are welcome. I am trying to create some crews that run locally with ollama (tried llama3:8b, mixtral, and some others)

here we go:

it seems the action input is nicely formatted (as a json...)
but it seems the "delegation tool" is being 'called' with (or expects?) something else, so validation seems to fail.

Here is the terminal output:

Entering new CrewAgentExecutor chain...
Thought: To create a comprehensive AI newsletter, I'll need to gather information about the top 5 AI news stories from the past 24 hours. This requires me to delegate tasks to coworkers.

Action: Delegate work to coworker

Action Input: { "task": "fetch top 5 ai news stories", "context": "Please provide a list of the top 5 AI news stories from the past 24 hours, including headlines, summaries, and sources.", "coworker": "NewsFetcher"

I encountered an error while trying to use the tool. This was the error: 3 validation errors for Delegate work to coworkerSchema
task
str type expected (type=type_error.str)
context
str type expected (type=type_error.str)
coworker
str type expected (type=type_error.str).
Tool Delegate work to coworker accepts these inputs: Delegate a specific task to one of the following coworkers: NewsFetcher, NewsAnalyzer, NewsletterCompiler
The input to this tool should be the coworker, the task you want them to do, and ALL necessary context to execute the task, they know nothing about the task, so share absolute everything you know, don't reference things but instead explain them.

// extra log on agent step callback:

Agent Step Callback: Agent
Output: [(AgentAction(tool='Delegate work to coworker', tool_input='{"task": "fetch top 5 ai news stories", "context": "Please provide a list of the top 5 AI news stories from the past 24 hours, including headlines, summaries, and sources.", "coworker": "NewsFetcher"}', log='Thought: To create a comprehensive AI newsletter, I'll need to gather information about the top 5 AI news stories from the past 24 hours. This requires me to delegate tasks to coworkers.\n\nAction: Delegate work to coworker\n\nAction Input: { "task": "fetch top 5 ai news stories", "context": "Please provide a list of the top 5 AI news stories from the past 24 hours, including headlines, summaries, and sources.", "coworker": "NewsFetcher"'), "\nI encountered an error while trying to use the tool. This was the error: 3 validation errors for Delegate work to coworkerSchema\ntask\n str type expected (type=type_error.str)\ncontext\n str type expected (type=type_error.str)\ncoworker\n str type expected (type=type_error.str).\n Tool Delegate work to coworker accepts these inputs: Delegate a specific task to one of the following coworkers: NewsFetcher, NewsAnalyzer, NewsletterCompiler\nThe input to this tool should be the coworker, the task you want them to do, and ALL necessary context to execute the task, they know nothing about the task, so share absolute everything you know, don't reference things but instead explain them..\nMoving on then. I MUST either use a tool (use one at time) OR give my best final answer. To Use the following format:\n\nThought: you should always think about what to do\nAction: the action to take, should be one of [Delegate work to coworker, Ask question to coworker]\nAction Input: the input to the action, dictionary enclosed in curly braces\nObservation: the result of the action\n... (this Thought/Action/Action Input/Observation can repeat N times)\nThought: I now can give a great answer\nFinal Answer: my best complete final answer to the task.\nYour final answer must be the great and the most complete as possible, it must be outcome described\n\n ")]

[tool.poetry.dependencies]
python = ">=3.10,<=3.13"
crewai = {version = "^0.41.1", extras = ["tools"]}
pydantic = "^2.8.2"
requests = "^2.32.3"
langchain-community = "^0.2.9"
python-dotenv = "^1.0.1"
langchain = "^0.2.10"
crewai-tools = "^0.4.26"

"""Newsletter crew module for AI newsletter generation."""

from typing import List

from file_io import save_markdown
from models.newsletter_models import NewsletterContent
from models.output import TaskOutput
from schemas import AgentResponse
from tools.search_tools import SearchTools
from crews.base_crew import BaseCrew

from crewai import Agent, Process, Task, Crew
from crewai.project import agent, task, crew
from crewai.agent import BaseAgent

from llm import creative_llm, instructional_llm, tool_llm

class NewsletterCrew(BaseCrew):
"""Crew for generating AI newsletters."""

def __init__(self):
    super().__init__()
    self.creative_llm = creative_llm
    self.instructional_llm = instructional_llm
    self.tool_llm = tool_llm
    self.agents_config = self.load_config("config/agents.yaml")
    self.tasks_config = self.load_config("config/tasks.yaml")
    self.search_tools = SearchTools()
    self.agents: List[BaseAgent] = []
    self.tasks: List[Task] = []
    self.build_agents()
    self.build_tasks()

def print_gray(self, text):
    """Print text in gray."""
    print(f"\033[90m{text}\033[0m")

def agent_step_callback(self, output):
    """Print agent step callback information."""
    self.print_gray("Agent Step Callback: Agent")
    self.print_gray(f"Output: {output}")
    if output.action == "Delegate work to coworker":
        self.print_gray(f"Delegation attempt: {output.action_input}")
        self.print_gray(f"Delegation attempt: {output.action_input}")

def task_callback(self, output: TaskOutput):
    """Print task callback information."""
    self.print_gray("Task Callback: Task completed")
    self.print_gray(f"Output: {output}")

def crew_step_callback(self, output: AgentResponse):
    """ "Print crew step callback information."""
    self.print_gray(f"Crew Step Callback: Crew is working on task '{output}'")

def crew_task_callback(self, output: TaskOutput):
    """Print crew task callback information."""
    self.print_gray("Crew Task Callback: Crew completed task")
    self.print_gray(f"Output summary: {output}")

def crew_manager_callback(self, output):
    """Print crew manager callback information."""
    self.print_gray(f"Crew Manager Callback: {output}")

def call_callbacks(self, callbacks, task_output):
    """
    Call each callback function in the given list with the provided arguments.

    :param callbacks: List of callback functions to call
    :param task_output: Positional arguments to pass to each callback
    """
    for callback in callbacks:
        callback(task_output)

@agent
def editor(self) -> Agent:
    """Create the editor agent."""
    return Agent(
        **self.agents_config["editor"],
        llm=self.creative_llm,
        function_calling_llm=self.tool_llm,
        max_iter=15,
        memory=False,
        cache=True,
        max_rpm=30,
        # tools=[],
        step_callback=self.agent_step_callback,
        # callbacks=[],
        allow_code_execution=False,
        # max_retry_limit=3,
    )

@agent
def news_fetcher(self) -> Agent:
    """Create the news fetcher agent."""
    config = self.agents_config["news_fetcher"]
    return Agent(
        **config,
        tools=[self.search_tools.search_internet],
        llm=self.creative_llm,
        function_calling_llm=self.tool_llm,
        max_iter=15,
        memory=False,
        cache=True,
        max_rpm=30,
        step_callback=self.agent_step_callback,
        # callbacks=[],
        allow_code_execution=False,
        # max_retry_limit=3,
    )

@agent
def news_analyzer(self) -> Agent:
    """Create the news analyzer agent."""
    return Agent(
        **self.agents_config["news_analyzer"],
        llm=self.creative_llm,
        function_calling_llm=self.tool_llm,
        max_iter=15,
        memory=False,
        cache=True,
        max_rpm=30,
        tools=[self.search_tools.search_internet],
        step_callback=self.agent_step_callback,
        # callbacks=[],
        allow_code_execution=False,
        # max_retry_limit=3,
    )

@agent
def newsletter_compiler(self) -> Agent:
    """Create the newsletter compiler agent."""
    return Agent(
        **self.agents_config["newsletter_compiler"],
        llm=self.instructional_llm,
        function_calling_llm=self.tool_llm,
        max_iter=15,
        memory=False,
        cache=True,
        max_rpm=30,
        # tools=[],
        step_callback=self.agent_step_callback,
        # callbacks=[],
        allow_code_execution=False,
        # max_retry_limit=3,
    )

@task
def project_task(self) -> Task:
    """Create the main project task."""
    config = self.tasks_config["project_task"]
    return Task(
        **config,
        agent=self.editor(),
        # context=[self.compile_newsletter_task()],
        async_execution=False,
        callback=self.task_callback,
        output_file="output/project_task.md",
        # output_json=None,
        # output_pydantic=None,
        # tools=None,
    )

@task
def fetch_news_task(self) -> Task:
    """Create the research task."""
    config = self.tasks_config["fetch_news_task"]
    return Task(
        **config,
        agent=self.news_fetcher(),
        context=None,
        async_execution=False,
        callback=self.task_callback,
        output_file="output/fetch_news_task.md",
        # output_json=None,
        # output_pydantic=None,
        # tools=None,
    )

@task
def analyze_news_task(self) -> Task:
    """Create the analyse task."""
    config = self.tasks_config["analyze_news_task"]
    return Task(
        **config,
        agent=self.news_analyzer(),
        context=[self.fetch_news_task()],
        async_execution=False,
        callback=self.task_callback,
        output_file="output/analyze_news_task.md",
        # output_json=None,
        # output_pydantic=None,
        # tools=None,
    )

@task
def compile_newsletter_task(self) -> Task:
    """Create the compile newsletter task task."""
    config = self.tasks_config["compile_newsletter_task"]
    return Task(
        **config,
        agent=self.newsletter_compiler(),
        context=[self.analyze_news_task()],
        output_json=NewsletterContent,
        callback=lambda task_output: self.call_callbacks(
            [
                self.task_callback,
                save_markdown,
            ],
            task_output,
        ),
        async_execution=False,
        output_file="output/compile_newsletter_task.md",
        # output_pydantic=None,
        # tools=None,
    )

def build_agents(self):
    """Build all agents for the crew."""
    self.agents = [
        self.editor(),
        self.news_fetcher(),
        self.news_analyzer(),
        self.newsletter_compiler(),
    ]

def build_tasks(self):
    """Build all tasks for the crew."""
    self.tasks = [
        self.fetch_news_task(),
        self.analyze_news_task(),
        self.compile_newsletter_task(),
    ]

@crew
def get_crew(self) -> Crew:
    """Get the configured crew."""
    return Crew(
        agents=self.agents,
        tasks=self.tasks,
        process=Process.sequential,
        verbose=True,
        embedder=None,
        cache=True,
        memory=False,
        # manager_llm=self.instructional_llm,
        manager_agent=None,
        manager_callbacks=None,
        function_calling_llm=self.instructional_llm,
        config=None,
        share_crew=False,
        step_callback=self.crew_step_callback,
        task_callback=self.crew_task_callback,
        max_rpm=None,
        # prompt_file=None,
        output_log_file="output/crew_log.md",
        planning=False,
        task_execution_output_json_files=None,
        # execution_logs=[],
    )

def run(self):
    """Execute the crew's tasks."""
    print("Running newsletter crew...")
    return self.get_crew().kickoff()

agents.yaml

editor:
role: Editor
goal: Oversee the creation of the AI Newsletter and return the full final compiled formatted newsletter.
backstory: With a keen eye for detail and a passion for storytelling, you ensure that the newsletter not only informs but also engages and inspires the readers.
allow_delegation: True
verbose: True

news_fetcher:
role: NewsFetcher
goal: Fetch the top AI news stories for the day
backstory: As a digital sleuth, you scour the internet for the latest and most impactful developments in the world of AI, ensuring that our readers are always in the know.
allow_delegation: True
verbose: True

news_analyzer:
role: NewsAnalyzer
goal: Analyze each news story and generate a detailed markdown summary
backstory: With a critical eye and a knack for distilling complex information, you provide insightful analyses of AI news stories, making them accessible and engaging for our audience.
allow_delegation: True
verbose: True

newsletter_compiler:
role: NewsletterCompiler
goal: Compile the analyzed news stories into a final newsletter format
backstory: As the final architect of the newsletter, you meticulously arrange and format the content, ensuring a coherent and visually appealing presentation that captivates our readers.
allow_delegation: True
verbose: True

tasks.yaml

project_task:
description: >
Create a newsletter with the top 5 AI news stories from the past 24 hours.
Make sure to include a rundown, detailed bullet points, and a "Why it matters" section for each story.

If you do your BEST WORK, I'll give you a $10,000 commission!

expected_output: >
A complete newsletter in specific markdown format, with a consistent style and layout.

fetch_news_task:
description: >
Fetch top AI news stories from the past 24 hours.
The current time is {current_time}.

If you do your BEST WORK, I'll give you a $10,000 commission!

expected_output: >
A list of top AI news story titles, URLs,
and a brief summary for each story from the past 24 hours.

Example Output: 
[
    {
      'title': 'AI takes spotlight in Super Bowl commercials', 
      'url': 'https://example.com/story1', 
      'summary': 'AI made a splash in this year\'s Super Bowl commercials...'
    }, 
    {...}
]

analyze_news_task:
description: >
Analyze each news story and ensure there are at least 5 well-formatted articles.
Compile the newsletter from the analyzed news stories.
The final answer must be the full newsletter in markdown format,
only the content of the newsletter and nothing else.

If you do your BEST WORK, I'll give you a $10,000 commission!

expected_output: >
A complete newsletter in markdown format, with a consistent style and layout.
Each article should include a rundown, detailed bullet points, and a "Why it matters" section.

Example Output:
# Top stories in AI today:
- AI takes spotlight in Super Bowl commercials
- Altman seeks TRILLIONS for global AI chip initiative

## AI takes spotlight in Super Bowl commercials

**The Rundown:** AI made a splash in this year's Super Bowl commercials...
**The details:**
- Microsoft's Copilot spot showcased its AI assistant...
**Why it matters:** While AI-related ads have been rampant over the last year, 
its Super Bowl presence is a big mainstream moment.
**Sources:**
- [Link to source 1](https://example.com/source1)
- [Link to source 2](https://example.com/source2)

compile_newsletter_task:
description: >
Compile the newsletter from the analyzed news stories.
The final answer must be the full newsletter in markdown format,
only the content of the newsletter and nothing else.

If you do your BEST WORK, I'll give you a $10,000 commission!

expected_output: >
A complete newsletter in markdown format, with a consistent style and layout.

Example Output:
'# Top stories in AI today:
- AI takes spotlight in Super Bowl commercials
- Altman seeks TRILLIONS for global AI chip initiative

## AI takes spotlight in Super Bowl commercials\\n\\n
**The Rundown:** AI made a splash in this year\'s Super Bowl commercials...
**The details:**...
**Why it matters::**...
**Sources:**
- [Link to source 1](https://example.com/source1)
- [Link to source 2](https://example.com/source2)
## Altman seeks TRILLIONS for global AI chip initiative
**The Rundown:** OpenAI CEO Sam Altman is reportedly angling 
to raise TRILLIONS of dollars...
**The details:**...
**Why it matters::**...
**Sources:**
- [Link to source 3](https://example.com/source3)
- [Link to source 4](https://example.com/source4)

newsletter_models

"""Models for the AI newsletter content."""

from typing import List
from pydantic import BaseModel, Field

class NewsItem(BaseModel):
"""Represents a single news item in the newsletter."""

title: str = Field(..., description="Title of the news item")
content: str = Field(..., description="Content of the news item")
source: str = Field(..., description="Source of the news item")

class NewsletterContent(BaseModel):
"""Represents the full content of the newsletter."""

title: str = Field(..., description="Title of the newsletter")
introduction: str = Field(
    ..., description="Introduction paragraph for the newsletter"
)
news_items: List[NewsItem] = Field(
    ..., description="List of news items in the newsletter"
)
conclusion: str = Field(..., description="Concluding paragraph for the newsletter")

crewai llama model:

FROM llama3:8b
PARAMETER temperature 0.8
PARAMETER stop Result
SYSTEM """"""

main:
def main():
"""Main function to run the AI newsletter generation process."""

load_dotenv()

crew = NewsletterCrew()
crew_output = crew.run()

thanks in advance!