Why do I get KeyError when trying to set the context parameter in the tasks.yaml file?

CrewAI Community Support
1

Hi all, I’m running into an issue with context defined in my tasks.yaml in crewai 0.83.0

Traceback (most recent call last):
  File "<path_to_script>/run_crew", line 8, in <module>
sys.exit(run())
         ^^^^^
  File "<path_to_project>/main.py", line 23, in run
CrewaiAirtableDemo().crew().kickoff(inputs=inputs)
^^^^^^^^^^^^^^^^^^^^
  File "<path_to_library>/crew_base.py", line 35, in __init__
self.map_all_task_variables()
  File "<path_to_library>/crew_base.py", line 166, in map_all_task_variables
self._map_task_variables(
  File "<path_to_library>/crew_base.py", line 189, in _map_task_variables
self.tasks_config[task_name]["context"] = [
                                          ^
  File "<path_to_library>/crew_base.py", line 190, in <listcomp>
tasks[context_task_name]() for context_task_name in context_list
~~~~~^^^^^^^^^^^^^^^^^^^
KeyError: 'review_product_website'

An excerpt from tasks.yaml:

review_product_website:
  description: >
    Retrieve information about {product_name} from {product_url}.
    Extract key details from the website.
  expected_output: >
    Raw text or product information from the website.
  agent: researcher

write_product_description:
  description: >
    Write a concise and objective product description and value proposition based on gathered data.
    Highlight important features etc etc.
  expected_output: >
    A paragraph summarizing the product description and value proposition from an objective point of view.
  context:
    - review_product_website
    - another_task_defined_above
  agent: product_analyst
  async_execution: True

This is intended to be a sequential process, and the tasks are ordered as such. crew.py follows the same order.

I installed with pip and am running the crew with crewai run, in case that makes a difference. (I’m a little confused on what the current recommendation is with pip vs. poetry.)

Thanks, and let me know if more information would be helpful.

2

@petari Set only agent, description, and expected_output in the YAML file. Set everything else in the crew.py file.

Note: I updated this answer after the second issue was solved in case anyone will read this thread in the future. Initially, my answer solved the issue with the KeyError in the tasks.yaml file, but after that issue was solved, the OP got another error (i.e., AttributeError: 'function' object has no attribute 'get') which my initial answer didn’t solve. Following the code below should now solve both(!) errors.

tasks.yaml

task_a_name:
  agent: your_agent
  description: >
    Write description here...
  expected_output: >
    Write expected output here...

task_b_name:
  agent: your_agent
  description: >
    Write description here...
  expected_output: >
    Write expected output here...

crew.py

@task
def task_a(self) -> Task:
  return Task(
    config=self.tasks_config['task_a_name'],
)

@task
def task_b(self) -> Task:
  return Task(
    config=self.tasks_config['task_b_name'],
    context=[self.task_a()],
)
3

Thanks, that got me unstuck and on to the next error. From the testing I did, it seems like it doesn’t like the way I set context in crew.py with self.task_a

@task
def task_a(self) -> Task:
  return Task(
    config=self.tasks_config['task_a_name'],
)

@task
def task_b(self) -> Task:
  return Task(
    config=self.tasks_config['task_b_name'],
    context=[self.task_a], # note that task_a is undefined within the class
)

This results in the following error:

Traceback (most recent call last):
  File "<path_to_script>/run_crew", line 8, in <module>
    sys.exit(run())
             ^^^^^
  File "<path_to_project>/main.py", line 23, in run
    CrewaiAirtableDemo().crew().kickoff(inputs=inputs)
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<path_to_library>/utils.py", line 7, in memoized_func
    cache[key] = func(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^^
  File "<path_to_library>/annotations.py", line 95, in wrapper
    task_instance = task_method(self)
                    ^^^^^^^^^^^^^^^^^
  File "<path_to_library>/utils.py", line 7, in memoized_func
    cache[key] = func(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^^
  File "<path_to_library>/annotations.py", line 23, in wrapper
    result = func(*args, **kwargs)
             ^^^^^^^^^^^^^^^^^^^^^
  File "<path_to_project>/crew.py", line 91, in write_product_description_task
    return Task(
           ^^^^^
  File "<path_to_library>/pydantic/main.py", line 214, in __init__
    validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<path_to_library>/task.py", line 124, in process_model_config
    return process_config(values, cls)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<path_to_library>/utilities/config.py", line 19, in process_config
    config = values.get("config", {})
             ^^^^^^^^^^
AttributeError: 'function' object has no attribute 'get'
4

@petari Can you show the code?

5

Yep, here’s the crew.py:

@CrewBase
class CrewaiAirtableDemo():
	"""CrewaiAirtableDemo crew"""

	agents_config = 'config/agents.yaml'
	tasks_config = 'config/tasks.yaml'

	@agent
	def researcher(self) -> Agent:
		return Agent(
			config=self.agents_config['researcher'],
			tools=[
				ScrapeWebsiteTool(),
				SerperDevTool()
			],
			verbose=True
		)

	@agent
	def product_analyst(self) -> Agent:
		return Agent(
			config=self.agents_config['product_analyst'],
			verbose=True
		)
	
	@agent
	def data_entry_specialist(self) -> Agent:
		return Agent(
			config=self.agents_config['data_entry_specialist'],
			verbose=True,
			llm=json_llm
		)

	
	@task
	def review_product_website_task(self) -> Task:
		return Task(
			config=self.tasks_config['review_product_website'],
		)
	
	@task
	def search_product_information_task(self) -> Task:
		return Task(
			config=self.tasks_config['search_product_information'],
		)
	
	@task
	def write_product_description_task(self) -> Task:
		return Task(
			config=self.tasks_config['write_product_description'],
			async_execution=True,
			context=[self.review_product_website_task, self.search_product_information_task],
		)
	
	@task
	def categorize_product_type_task(self) -> Task:
		return Task(
			config=self.tasks_config['categorize_product_type'],
			async_execution=True,
			context=[self.review_product_website_task, self.search_product_information_task],
		)
	
	@task
	def categorize_technology_task(self) -> Task:
		return Task(
			config=self.tasks_config['categorize_technology'],
			async_execution=True,
			context=[self.review_product_website_task, self.search_product_information_task]
		)
	
	@task
	def categorize_interest_area_task(self) -> Task:
		return Task(
			config=self.tasks_config['categorize_client_interest_areas'],
			async_execution=True,
			context=[self.review_product_website_task, self.search_product_information_task]
		)
	
	@task
	def create_product_record_task(self) -> Task:
		return Task(
			config=self.tasks_config['create_product_record'],
			context=[
				self.write_product_description_task,
				self.categorize_technology_task,
				self.categorize_product_type_task,
				self.categorize_client_interest_area_task
			],
			output_pydantic=ProductRecord
		)
	
	@crew
	def crew(self) -> Crew:
		"""Creates the CrewaiAirtableDemo crew"""
		return Crew(
			agents=self.agents, # Automatically created by the @agent decorator
			tasks=self.tasks, # Automatically created by the @task decorator
			process=Process.sequential,
			verbose=True,
		)

The code runs fine when I keep context commented out. The note above was just that I can’t reference the task definition without including self.task_name as it’s all in the same class.

Thanks again for your help.

7

Hy frient.

Try this way:

context=[
self.write_product_description_task(),
self.categorize_technology_task(),
self.categorize_product_type_task(),
self.categorize_client_interest_area_task()
],

8

@GuilhermeHenriqueBil That was it. It’s running correctly now.

In summary:
This didn’t work

context = [task_a, task_b] → task_a is undefined

This also didn’t work:

context = [self.task_a, self.task_b] → AttributeError: ‘function’ object has no attribute ‘get’

This works

context = [self.task_a(), self.task_b()]

@rokbenko The full error was

Traceback (most recent call last):
  File "<path_to_script>/run_crew", line 8, in <module>
    sys.exit(run())
             ^^^^^
  File "<path_to_project>/main.py", line 23, in run
    CrewaiAirtableDemo().crew().kickoff(inputs=inputs)
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<path_to_library>/utils.py", line 7, in memoized_func
    cache[key] = func(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^^
  File "<path_to_library>/annotations.py", line 95, in wrapper
    task_instance = task_method(self)
                    ^^^^^^^^^^^^^^^^^
  File "<path_to_library>/utils.py", line 7, in memoized_func
    cache[key] = func(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^^^
  File "<path_to_library>/annotations.py", line 23, in wrapper
    result = func(*args, **kwargs)
             ^^^^^^^^^^^^^^^^^^^^^
  File "<path_to_project>/crew.py", line 91, in write_product_description_task
    return Task(
           ^^^^^
  File "<path_to_library>/pydantic/main.py", line 214, in __init__
    validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<path_to_library>/task.py", line 124, in process_model_config
    return process_config(values, cls)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<path_to_library>/utilities/config.py", line 19, in process_config
    config = values.get("config", {})
             ^^^^^^^^^^
AttributeError: 'function' object has no attribute 'get'

Thanks all for your help!