ИИ-агенты для начинающих. Часть 5. Создание задач для агентов в CrewAI

При подготовке статьи использовалась публикация «Задачи в Crew AI».

Что такое задачи (task) в CrewAI?

В CrewAI, Задача (Task) — это конкретное поручение, выполняемое Агентом.Задачи предоставляют все необходимые детали для выполнения, такие как описание, ответственный агент, требуемые инструменты и многое другое, что позволяет реализовывать действия различной сложности.Задачи в CrewAI могут быть совместными, требующими работы нескольких агентов вместе. Это управляется через свойства задачи и координируется процессом Команды (Crew), что улучшает командную работу и эффективность.В CrewAI есть два основных способа выполнения задач:
1. Последовательный (Sequential) — задачи выполняются строго одна за другой, как в конвейере. Удобно, когда результат одной задачи нужен для следующей.
1. Иерархический (Hierarchical) — задачи распределяются между агентами с учетом их специализации и уровня «экспертности».

Атрибуты задач

Атрибут	Параметр	Тип	Описание
Description	description	str	Четкая и краткая формулировка того, что включает в себя задача.
Expected Output	expected_output	str	Подробное описание того, как выглядит выполнение задачи.
Name*	name	Optional[str]	Идентификатор (имя) задачи.
Agent*	agent	Optional[BaseAgent]	Агент, ответственный за выполнение задачи.
Tools*	tools	List[BaseTool]	Инструменты/ресурсы, которыми ограничен агент для выполнения этой задачи.
Context*	context	Optional[List["Task"]]	Другие задачи, результаты которых будут использоваться как контекст для этой задачи.
Async Execution*	async_execution	Optional[bool]	Должна ли задача выполняться асинхронно. По умолчанию False.
Human Input*	human_input	Optional[bool]	Должен ли человек проверять окончательный ответ агента. По умолчанию False.
Config*	config	Optional[Dict[str, Any]]	Параметры конфигурации, специфичные для задачи.
Output File*	output_file	Optional[str]	Путь к файлу для хранения результата задачи.
Output JSON*	output_json	Optional[Type[BaseModel]]	Pydantic-модель для структурирования JSON-вывода.
Output Pydantic*	output_pydantic	Optional[Type[BaseModel]]	Pydantic-модель для вывода задачи.
Callback*	callback	Optional[Any]	Функция/объект, которые должны быть выполнены после завершения задачи.
*опционально

Способы создания задачи

В CrewAI можно создавать задачи двумя способами:
- Через YAML-конфигурацию (рекомендуемый способ)
- Напрямую в коде (когда хочется пожить опасно)

YAML-конфигурация — путь джедая

Если вы хотите, чтобы ваш код был чистым и поддерживаемым (а кто не хочет?), то YAML-конфигурация — ваш лучший друг.После создания проекта:
1. Найдем файл конфигурации: src/hmhm_project/config/tasks.yaml
1. Отредактируем шаблон под свои задачи

Пример YAML-файла

research_task:
  description: >
    Проведите тщательное исследование по теме {topic}
    Убедитесь, что вы нашли всю интересную и актуальную информацию,
    учитывая, что текущий год - 2025.
  expected_output: >
    Список из 10 пунктов с наиболее актуальной информацией по теме {topic}
  agent: researcher

reporting_task:
  description: >
    Просмотрите полученный контекст и расширьте каждую тему в полноценный раздел отчета.
    Убедитесь, что отчет детализирован и содержит всю релевантную информацию.
  expected_output: >
    Полноценный отчет с основными темами, каждая из которых представлена полным разделом информации.
    Отформатировано в markdown без использования '```'
 agent: reporting_analyst
 output_file: report.md

Для использования этой YAML конфигурации в вашем коде создайте класс crew, который наследуется от CrewBase:

# src/latest_ai_development/crew.py

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai_tools import SerperDevTool

@CrewBase
class HmHmCrew:

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

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

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

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

 @crew
 def crew(self) -> Crew:
 return Crew(
 agents=[
 self.researcher(),
 self.reporting_analyst()
 ],
 tasks=[
 self.research_task(),
 self.reporting_task()
 ],
 process=Process.sequential
 )

Объявление задач в коде

Вот альтернативный способ определения задач напрямую в коде без использования YAML конфигурации:

from crewai import Task

research_task = Task(
 description="""
 Проведите тщательное исследование по теме {topic}
 Убедитесь, что вы нашли всю интересную и актуальную информацию,
 учитывая,

Формат вывода задачи

CrewAI предоставляет структурированный способ обработки результатов задач через класс TaskOutput, который поддерживает несколько форматов вывода и может легко передаваться между задачами.Вывод задачи в фреймворке CrewAI инкапсулируется в классе TaskOutput. Этот класс предоставляет структурированный способ доступа к результатам задачи, включая различные форматы, такие как необработанный вывод, JSON и Pydantic-модели.По умолчанию TaskOutput будет включать только необработанный вывод. TaskOutput будет включать вывод pydantic или json_dict только в том случае, если исходный объект Task был настроен с параметрами output_pydantic или output_json соответственно.

Атрибуты вывода задачи

Атрибут	Параметр	Тип	Описание
Description	description	str	Описание задачи.
Summary	summary	Optional[str]	Краткое содержание задачи, автоматически сгенерированное из первых 10 слов описания.
Raw	raw	str	Необработанный вывод задачи. Это формат вывода по умолчанию.
Pydantic	pydantic	Optional[BaseModel]	Объект Pydantic-модели, представляющий структурированный вывод задачи.
JSON Dict	json_dict	Optional[Dict[str, Any]]	Словарь, представляющий вывод задачи в формате JSON.
Agent	agent	str	Агент, который выполнил задачу.
Output Format	output_format	OutputFormat	Формат вывода задачи с вариантами, включающими RAW (необработанный), JSON и Pydantic. По умолчанию используется RAW.

Свойства задачи

Свойство	Описание
json	Возвращает строковое представление вывода задачи в формате JSON, если формат вывода установлен как JSON.
to_dict	Преобразует выводы JSON и Pydantic в словарь.
str	Возвращает строковое представление вывода задачи, отдавая приоритет сначала Pydantic, затем JSON, и наконец необработанному формату.

Доступ к выводу/результату задачи

После выполнения задачи к ее выводу можно получить доступ через атрибут output объекта Task. Класс TaskOutput предоставляет различные способы взаимодействия с этим выводом и его представления.

task = Task(
 description='Найти и обобщить последние новости в сфере ИИ',
 expected_output='Маркированный список из 5 самых важных новостей в сфере ИИ',
 agent=research_agent,
 tools=[search_tool]
)

crew = Crew(
 agents=[research_agent],
 tasks=[task],
 verbose=True
)

result = crew.kickoff()

task_output = task.output

print(f"Описание задачи: {task_output.description}")
print(f"Краткое содержание задачи: {task_output.summary}")
print(f"Необработанный вывод: {task_output.raw}")
if task_output.json_dict:
 print(f"Вывод в формате JSON: {json.dumps(task_output.json_dict, indent=2)}")
if task_output.pydantic:
 print(f"Вывод в формате Pydantic: {task_output.pydantic}")

Зависимости и контекст задач

Задачи могут зависеть от вывода других задач, используя атрибут context. Например:

research_task = Task(
 description="Исследовать последние разработки в сфере ИИ",
 expected_output="Список последних разработок в области ИИ",
 agent=researcher
)

analysis_task = Task(
 description="Проанализировать результаты исследования и определить ключевые тенденции",
 expected_output="Аналитический отчет о тенденциях в ИИ",
 agent=analyst,
 context=[research_task] # Эта задача будет ждать завершения research_task
)

Ограничения задач (Task Guardrails)

Ограничения задач (Task guardrails) предоставляют способ проверки и преобразования выводов задач перед их передачей следующей задаче. Эта функция помогает обеспечить качество данных и предоставляет обратную связь агентам, когда их вывод не соответствует определенным критериям.

Использование ограничений задач

Чтобы добавить ограничение к задаче, необходимо предоставить функцию проверки через параметр guardrail.

from typing import Tuple, Union, Dict, Any

def validate_blog_content(result: str) -> Tuple[bool, Union[Dict[str, Any], str]]:
 """Проверить, что содержание блога соответствует требованиям."""
 try:
 # Проверить количество слов
 word_count = len(result.split())
 if word_count > 200:
 return False, {
 "error": "Содержание блога превышает 200 слов",
 "code": "WORD_COUNT_ERROR",
 "context": {"word_count": word_count},
 }

 # Дополнительная проверка
 return True, result.strip()
 except Exception as e:
 return False, {
 "error": "Непредвиденная ошибка во время

Требования к Функции-ограничителю

Сигнатура Функции:
- Должна принимать ровно один параметр (выходные данные задачи)
- Должна возвращать кортеж из (bool, Any)
- Рекомендуются подсказки типов, но они необязательны
Возвращаемые значения:
- Успех: возвращает (True, validated_result)
- Ошибка: возвращает (False, error_details)

Обработка результатов ограничителя

Когда ограничитель возвращает (False, error):
- Ошибка отправляется обратно агенту
- Агент пытается исправить проблему
- Процесс повторяется до тех пор, пока:
- Ограничитель не вернет (True, result)
- Не будет достигнуто максимальное количество попыток
Пример с обработкой повторных попыток:

import json
from typing import Any, Dict, Tuple, Union

def validate_json_output(result: str) -> Tuple[bool, Union[Dict[str, Any], str]]:
 try:
 # попытка загрузить JSON
 data = json.loads(result)
 return (True, data)
 except json.JSONDecodeError as e:
 return (False, {
 "error": "Неправильный JSON-формат",
 "code": "JSON_ERROR",
 "context": {"line": e.lineno, "column": e.colno}
 })

task = Task(
 description="Создать JSON-отчет",
 expected_output="Корректный JSON-объект",
 agent=analyst,
 guardrail=validate_json_output,
 max_retries=3 # Ограничить количество попыток
)

Использование output_json

Свойство output_json позволяет определить ожидаемый вывод в формате JSON. Это гарантирует, что вывод задачи является действительной JSON-структурой, которая может быть легко распарсена и использована в вашем приложении.Вот пример, демонстрирующий как использовать output_json:

from crewai import Agent, Crew, Process, Task
from pydantic import BaseModel

class Blog(BaseModel):
 title: str
 content: str

blog_agent = Agent(
 role="Агент-генератор контента блога",
 goal="Создать заголовок и содержание блога",
 backstory="Вы - опытный создатель контента, умеющий создавать увлекательные и информативные записи в блоге.",
 verbose=False,
 allow_delegation=False,
 llm="gpt-4o",
)

task1 = Task(
 description="Создайте заголовок и содержание блога на заданную тему. Убедитесь, что содержание не превышает 200 слов.",
 expected_output="JSON-объект с полями 'title' и 'content'.",
 agent=blog_agent,
 output_json=Blog,
)

crew = Crew(
 agents=[blog_agent],
 tasks=[task1],
 verbose=True,
 process=Process.sequential,
)

result = crew.kickoff()

print("Доступ к свойствам - Вариант 1")
title = result["title"]
content = result["content"]
print("Заголовок:", title)
print("Содержание:", content)

print("Доступ к свойствам - Вариант 2")
print("Блог:", result)

В этом примере:Определена Pydantic-модель Blog с полями title и content, которая используется для указания структуры JSON-вывода. Задача task1 использует свойство output_json, чтобы указать, что ожидается JSON-вывод, соответствующий модели Blog. После выполнения команды (crew) вы можете получить доступ к структурированному JSON-выводу двумя способами, как показано.Объяснение доступа к выводу:
1. Доступ к свойствам через индексацию словаря:
- Вы можете получить доступ к полям напрямую, используя result[«field_name»]
- Это возможно, потому что класс CrewOutput реализует метод getitem, позволяющий обращаться с выводом как со словарем
- В этом варианте мы получаем заголовок (title) и содержание (content) из результата
1. Вывод всего объекта Blog:
- При выводе result вы получаете строковое представление объекта CrewOutput
- Поскольку метод str реализован для возврата JSON-вывода, это отобра

Интегрирование инструментов в задачи

Используйте инструменты из CrewAI Toolkit и LangChain Tools для взаимодействия агентов.

Создание задачи с инструментами

import os

os.environ["OPENAI_API_KEY"] = "Ваш Ключ"
os.environ["SERPER_API_KEY"] = "Ваш Ключ" # ключ API serper.dev

from crewai import Agent, Task, Crew
from crewai_tools import SerperDevTool

research_agent = Agent(
 role='Исследователь',
 goal='Найти и обобщить последние новости об ИИ',
 backstory="""Вы - исследователь в крупной компании.
 Вы отвечаете за анализ данных и предоставление
 аналитических выводов для бизнеса.""",
 verbose=True
)

# для выполнения семантического поиска по заданному запросу в содержании текстов по всему интернету
search_tool = SerperDevTool()

task = Task(
 description='Найти и обобщить последние новости об ИИ',
 expected_output='Маркированный список с кратким содержанием 5 самых важных новостей об ИИ',
 agent=research_agent,
 tools=[search_tool]
)

crew = Crew(
 agents=[research_agent],
 tasks=[task],
 verbose=True
)

result = crew.kickoff()
print(result)

Передача вывода другой задаче

В CrewAI вывод одной задачи автоматически передается в следующую, но вы можете специально определить, какие выходные данные задач (включая множественные) должны использоваться как контекст для другой задачи.Это полезно, когда у вас есть задача, которая зависит от вывода другой задачи, которая выполняется не сразу после нее. Это делается через атрибут context задачи.

# ...

research_ai_task = Task(
 description="Исследовать последние разработки в области ИИ",
 expected_output="Список последних разработок в области ИИ",
 async_execution=True,
 agent=research_agent,
 tools=[search_tool],
)

research_ops_task = Task(
 description="Исследовать последние разработки в области AI Ops",
 expected_output="Список последних разработок в области AI Ops",
 async_execution=True,
 agent=research_agent,
 tools=[search_tool],
)

write_blog_task = Task(
 description="Написать полноценный пост в блог о важности ИИ и последних новостях",
 expected_output="Полный пост в блог длиной в 4 абзаца",
 agent=writer_agent,
 context=[research_ai_task, research_ops_task],
)

# ...

Асинхронное выполнение

Вы можете определить задачу для асинхронного выполнения. Это означает, что команда (crew) не будет ждать ее завершения, прежде чем перейти к следующей задаче. Это полезно в следующих случаях:
- для задач, которые требуют длительного времени выполнения
- для задач, которые не являются критически важными для выполнения следующих задач
Затем вы можете использовать атрибут context в будущей задаче, чтобы указать, что она должна дождаться завершения и получения результатов асинхронной задачи.

# ...

list_ideas = Task(
 description="Список из 5 интересных идей для исследования в статье об ИИ.",
 expected_output="Маркированный список из 5 идей для статьи.",
 agent=researcher,
 async_execution=True # Будет выполняться асинхронно
)

list_important_history = Task(
 description="Исследовать историю ИИ и предоставить 5 самых важных событий.",
 expected_output="Маркированный список из 5 важных событий.",
 agent=researcher,
 async_execution=True # Будет выполняться асинхронно
)

write_article = Task(
 description="Написать статью об ИИ, его истории и интересных идеях.",
 expected_output="Статья об ИИ из 4 абзацев.",
 agent=writer,
 context=[list_ideas, list_important_history] # Будет ждать завершения выполнения двух задач
)

# ...

Механизм обратного вызова

Функция обратного вызова (callback) выполняется после завершения задачи, позволяя запускать действия или уведомления на основе результата выполнения задачи.

# ...

def callback_function(output: TaskOutput):
 # Выполнить что-то после завершения задачи
 # Пример: Отправить email менеджеру
 print(f"""
 Задача завершена!
 Задача: {output.description}
 Результат: {output.raw}
 """)

research_task = Task(
 description='Найти и обобщить последние новости об ИИ',
 expected_output='Маркированный список с кратким содержанием 5 самых важных новостей об ИИ',
 agent=research_agent,
 tools=[search_tool],
 callback=callback_function
)

#...

Доступ к результату конкретной задачи

После завершения работы crew (команды) вы можете получить доступ к результату конкретной задачи, используя атрибут output объекта задачи.

Механизм переопределения инструментов

Указание инструментов (tools) в задаче позволяет динамически адаптировать возможности агента, что подчеркивает гибкость CrewAI.

Механизмы обработки ошибок и валидации

При создании и выполнении задач действуют определенные механизмы валидации, обеспечивающие надежность и корректность атрибутов задачи. Они включают, но не ограничиваются следующим:
- Обеспечение только одного типа вывода для каждой задачи для поддержания четких ожиданий результата
- Предотвращение ручного назначения атрибута id для сохранения целостности системы уникальных идентификаторов
Эти проверки помогают поддерживать согласованность и надежность выполнения задач в рамках фреймворка crewAI.

Ограничители задач (Task Guardrails)

Ограничители задач — способ проверки, преобразования или фильтрации результатов задач перед их передачей следующей задаче. Ограничители — это опциональные функции, которые выполняются перед началом следующей задачи, позволяя убедиться, что результаты задач соответствуют определенным требованиям или форматам.

from typing import Tuple, Union
from crewai import Task

def validate_json_output(result: str) -> Tuple[bool, Union[dict, str]]:
 """Проверяет, что вывод является корректным JSON."""
 try:
 json_data =.loads(result)
 return (True, json_data)
 except.JSONDecodeError:
 return (False, "Вывод должен быть корректным JSON")

task = Task(
 description="Сгенерировать данные в формате JSON",
 expected_output="Корректный JSON объект",
 guardrail=validate_json_output,
)

Как работают ограничители (Guardrails)

Опциональный атрибут

Ограничители являются необязательным атрибутом на уровне задачи, позволяя добавлять валидацию только там, где это необходимо.

Время выполнения

Функция-ограничитель выполняется перед началом следующей задачи, обеспечивая корректный поток данных между задачами.

Формат возвращаемого значения

Ограничители должны возвращать кортеж из двух элементов (success, data):
- Если success равен True, то data содержит проверенный/преобразованный результат
- Если success равен False, то data содержит сообщение об ошибке

Маршрутизация результатов

• При успехе (True): результат автоматически передается следующей задаче
• При неудаче (False): ошибка отправляется обратно агенту для генерации нового ответа

Типовые сценарии использования ограничителей

1. Валидация формата данных

def validate_email_format(result: str) -> Tuple[bool, Union[str, str]]:
 """Проверяет, что вывод содержит корректный email-адрес."""
 import re
 email_pattern = '^[\w\.-]+@[\w\.-]+\.\w+$'
 if re.match(email_pattern, result.strip()):
 return (True, result.strip())
 return (False, "Вывод должен быть корректным email-адресом")

2. Проверка на конфиденциальную информацию:

sensitive_patterns = ['SSN:', 'password:', 'secret:']
for pattern in sensitive_patterns:
 if pattern.lower() in result.lower():
 return (False, f"Вывод содержит конфиденциальную информацию ({pattern})")
return (True, result)

3. Нормализация номера телефона:

def normalize_phone_number(result: str) -> Tuple[bool, Union[str, str]]:
 """Обеспечивает единый формат номеров телефона."""
 import re

 digits = re.sub('\D', '', result)
 if len(digits) == 10:
 formatted = f"({digits[:3]}) {digits[3:6]}-{digits[6:]}"
 return (True, formatted)
 return (False, "Вывод должен быть 10-значным номером телефона")

---

Содержание

1. Что такое ИИ-агенты и где они применяются
2. Агентный фреймворк CrewAI
3. Установка CrewAI и создание нового проекта
4. Агенты в CrewAI
5. Создание задач для агентов в CrewAI

Практикум

1. Создание системы автоматического перевода и редактирования текстов