motleycrew.applications.research_agent.answer_task

Classes

AnswerTask(crew[, answer_length, llm])

Task to answer a question based on the notes and sub-questions.

QuestionAnsweringTaskUnit(*[, status, output])

class motleycrew.applications.research_agent.answer_task.QuestionAnsweringTaskUnit(*, 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.answer_task.AnswerTask(crew: MotleyCrew, answer_length: int = 1000, llm: BaseLanguageModel | None = None)

Bases: Task

Task to answer a question based on the notes and sub-questions.

__init__(crew: MotleyCrew, answer_length: int = 1000, llm: BaseLanguageModel | None = None)

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() → QuestionAnsweringTaskUnit | None

Choose an unanswered question to answer.

The question should have a context and no unanswered subquestions.

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

Get the worker that will run the task units.

This method is called by the crew when a unit of the task is dispatched. The unit will be converted to a dictionary and passed to the worker’s invoke method.

Typically, the worker is an agent, but it can be any object that implements the Langchain Runnable interface.

Parameters:

tools – Tools to be used by the worker.

Returns:

Worker that will run the task units.