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

@
#, , ,

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

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

В CrewAI, Задача (Task) — это конкретное поручение, выполняемое Агентом.

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

Задачи в CrewAI могут быть совместными, требующими работы нескольких агентов вместе. Это управляется через свойства задачи и координируется процессом Команды (Crew), что улучшает командную работу и эффективность.

В CrewAI есть два основных способа выполнения задач:

  1. Последовательный (Sequential) — задачи выполняются строго одна за другой, как в конвейере. Удобно, когда результат одной задачи нужен для следующей.
  2. Иерархический (Hierarchical) — задачи распределяются между агентами с учетом их специализации и уровня «экспертности».

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

АтрибутПараметрТипОписание
DescriptiondescriptionstrЧеткая и краткая формулировка того, что включает в себя задача.
Expected Outputexpected_outputstrПодробное описание того, как выглядит выполнение задачи.
Name*nameOptional[str]Идентификатор (имя) задачи.
Agent*agentOptional[BaseAgent]Агент, ответственный за выполнение задачи.
Tools*toolsList[BaseTool]Инструменты/ресурсы, которыми ограничен агент для выполнения этой задачи.
Context*contextOptional[List["Task"]]Другие задачи, результаты которых будут использоваться как контекст для этой задачи.
Async Execution*async_executionOptional[bool]Должна ли задача выполняться асинхронно. По умолчанию False.
Human Input*human_inputOptional[bool]Должен ли человек проверять окончательный ответ агента. По умолчанию False.
Config*configOptional[Dict[str, Any]]Параметры конфигурации, специфичные для задачи.
Output File*output_fileOptional[str]Путь к файлу для хранения результата задачи.
Output JSON*output_jsonOptional[Type[BaseModel]]Pydantic-модель для структурирования JSON-вывода.
Output Pydantic*output_pydanticOptional[Type[BaseModel]]Pydantic-модель для вывода задачи.
Callback*callbackOptional[Any]Функция/объект, которые должны быть выполнены после завершения задачи.
*опционально

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

В CrewAI можно создавать задачи двумя способами:

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

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

Если вы хотите, чтобы ваш код был чистым и поддерживаемым (а кто не хочет?), то YAML-конфигурация — ваш лучший друг.

После создания проекта:

  1. Найдем файл конфигурации: src/hmhm_project/config/tasks.yaml
  2. Отредактируем шаблон под свои задачи

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

Python
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:

Python
# 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 конфигурации:

Python
from crewai import Task

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

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

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

CrewAI предоставляет структурированный способ обработки результатов задач через класс TaskOutput, который поддерживает несколько форматов вывода и может легко передаваться между задачами.

Вывод задачи в фреймворке CrewAI инкапсулируется в классе TaskOutput. Этот класс предоставляет структурированный способ доступа к результатам задачи, включая различные форматы, такие как необработанный вывод, JSON и Pydantic-модели.

По умолчанию TaskOutput будет включать только необработанный вывод. TaskOutput будет включать вывод pydantic или json_dict только в том случае, если исходный объект Task был настроен с параметрами output_pydantic или output_json соответственно.

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

АтрибутПараметрТипОписание
DescriptiondescriptionstrОписание задачи.
SummarysummaryOptional[str]Краткое содержание задачи, автоматически сгенерированное из первых 10 слов описания.
RawrawstrНеобработанный вывод задачи. Это формат вывода по умолчанию.
PydanticpydanticOptional[BaseModel]Объект Pydantic-модели, представляющий структурированный вывод задачи.
JSON Dictjson_dictOptional[Dict[str, Any]]Словарь, представляющий вывод задачи в формате JSON.
AgentagentstrАгент, который выполнил задачу.
Output Formatoutput_formatOutputFormatФормат вывода задачи с вариантами, включающими RAW (необработанный), JSON и Pydantic. По умолчанию используется RAW.

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

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

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

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

Python
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. Например:

Python
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.

Python
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": "Непредвиденная ошибка во время проверки",
            "code": "SYSTEM_ERROR"
        })

blog_task = Task(
    description="Написать пост в блог об ИИ",
    expected_output="Пост в блог менее 200 слов,
    agent=blog_agent,
    guardrail=validate_blog_content  # Добавить функцию ограничения
)

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

Сигнатура Функции:

  • Должна принимать ровно один параметр (выходные данные задачи)
  • Должна возвращать кортеж из (bool, Any)
  • Рекомендуются подсказки типов, но они необязательны

Возвращаемые значения:

  • Успех: возвращает (True, validated_result)
  • Ошибка: возвращает (False, error_details)

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

Когда ограничитель возвращает (False, error):

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

Пример с обработкой повторных попыток:

Python
from typing import Optional, 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:

Python
import json

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


# Определяем Pydantic модель для блога
class Blog(BaseModel):
    title: str
    content: str


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

# Определяем задачу с output_json, установленным на модель Blog
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()

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

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

В этом примере:

Определена Pydantic-модель Blog с полями title и content, которая используется для указания структуры JSON-вывода. Задача task1 использует свойство output_json, чтобы указать, что ожидается JSON-вывод, соответствующий модели Blog. После выполнения команды (crew) вы можете получить доступ к структурированному JSON-выводу двумя способами, как показано.

Объяснение доступа к выводу:

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

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

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

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

Python
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 задачи.

Python
# ...

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 в будущей задаче, чтобы указать, что она должна дождаться завершения и получения результатов асинхронной задачи.

Python
#...

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) выполняется после завершения задачи, позволяя запускать действия или уведомления на основе результата выполнения задачи.

Python
# ...

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)

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

Python
from typing import Tuple, Union
from crewai import Task

def validate_json_output(result: str) -> Tuple[bool, Union[dict, str]]:
    """Проверяет, что вывод является корректным JSON."""
    try:
        json_data = json.loads(result)
        return (True, json_data)
    except json.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. Валидация формата данных

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

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

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

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

Python
def normalize_phone_number(result: str) -> Tuple[bool, Union[str, str]]:
    """Обеспечивает единый формат номеров телефона."""
    import re
    digits = re.sub(r'\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. Создание системы автоматического перевода и редактирования текстов

Политика конфиденциальности / ВК