motleycrew.applications.research_agent.question_task

Classes

QuestionGenerationTaskUnit(*[, status, output])
QuestionTask(question, query_tool, crew[, ...])	Task to generate subquestions based on a given question.

class motleycrew.applications.research_agent.question_task.QuestionGenerationTaskUnit(*, status: str = 'pending', output: Any | None = None, question: Question)

Bases: TaskUnit

question: Question

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'output': FieldInfo(annotation=Union[Any, NoneType], required=False, default=None), 'question': FieldInfo(annotation=Question, required=True), 'status': FieldInfo(annotation=str, required=False, default='pending')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class motleycrew.applications.research_agent.question_task.QuestionTask(question: str, query_tool: MotleyTool, crew: MotleyCrew, max_iter: int = 10, allow_async_units: bool = False, llm: BaseLanguageModel | None = None, name: str = 'QuestionTask')

Bases: Task

Task to generate subquestions based on a given question.

__init__(question: str, query_tool: MotleyTool, crew: MotleyCrew, max_iter: int = 10, allow_async_units: bool = False, llm: BaseLanguageModel | None = None, name: str = 'QuestionTask')

Initialize the task.

Parameters:

• name – Name of the task.

• task_unit_class – Class for representing task units.

• crew – Crew to which the task belongs. If not provided, the task should be registered with a crew later.

• allow_async_units – Whether the task allows asynchronous units. Default is False. If True, the task may be queried for the next unit even if it has other units in progress.

get_next_unit() → QuestionGenerationTaskUnit | None

Choose the most pertinent question to generate subquestions for.

on_unit_dispatch(unit: TaskUnitType) → None

Increment the iteration count when a unit is dispatched.

on_unit_completion(unit: TaskUnitType) → None

Check if the task is done after each unit completion.

The task is done if the maximum number of iterations is reached.

get_worker(tools: List[MotleyTool] | None) → Runnable

Return the worker that will process the task units.

get_unanswered_questions(only_without_children: bool = False) → list[Question]