Class AssignmentConfig

src/typings/config.py:31–160

class AssignmentConfig(BaseModel):
    assignments: List[Assignment]
    concurrency: ConcurrencyConfig
    definition: DefinitionConfig
    output: str = None

    @validator("assignments", pre=True)
    def assignments_validation(cls, v):
        assert isinstance(v, list), f"'assignments' must be a list, but got {type(v)}"
        ret = []
        for item in v:
            assert isinstance(
                item, dict
            ), f"Each item in 'assignments' must be a dict, but got {type(item)}"
            agent = item.get("agent", None)
            if agent is None:
                raise ValueError("'agent' must be specified")
            if isinstance(agent, str):
                agent = [agent]
            task = item.get("task")
            if task is None:
                raise ValueError("'task' must be specified")
            if isinstance(task, str):
                task = [task]
            for a in agent:
                for t in task:
                    ret.append(Assignment(agent=a, task=t))
        return ret

    @validator("output", pre=True)
    def output_validation(cls, v):
        predefined_structure = get_predefined_structure()
        if v is None:
            v = "output/{TIMESTAMP}"
        assert isinstance(v, str), f"'output' must be a string, but got {type(v)}"
        return v.format(**predefined_structure)

    @classmethod
    def post_validate(cls, instance: "AssignmentConfig"):

        REMOVE_UNUSED_IN_DEFINITION = True
        REMOVE_UNUSED_IN_CONCURRENCY = True

        # Step 1. Check if all agents and tasks are defined, and remove unused agents and tasks

        agent_in_assignment = set()
        task_in_assignment = set()
        for assignment in instance.assignments:
            assert (
                assignment.agent in instance.definition.agent
            ), f"Agent {assignment.agent} is not defined."
            agent_in_assignment.add(assignment.agent)
            assert (
                assignment.task in instance.definition.task
            ), f"Task {assignment.task} is not defined."
            task_in_assignment.add(assignment.task)

        for agent in agent_in_assignment: