(cls,
config_dir_or_id: Optional[str] = None,
config: Optional[DictConfig] = None,
env: Optional[Dict[str, str]] = None,
trust_remote_code: bool = False,
**kwargs)
| 9 | |
| 10 | @classmethod |
| 11 | def build(cls, |
| 12 | config_dir_or_id: Optional[str] = None, |
| 13 | config: Optional[DictConfig] = None, |
| 14 | env: Optional[Dict[str, str]] = None, |
| 15 | trust_remote_code: bool = False, |
| 16 | **kwargs): |
| 17 | wf_config: Optional[DictConfig] = None |
| 18 | if config_dir_or_id is not None: |
| 19 | wf_config: DictConfig = Config.from_task(config_dir_or_id, env) |
| 20 | if config is not None: |
| 21 | if wf_config is not None: |
| 22 | wf_config = OmegaConf.merge(wf_config, config) |
| 23 | else: |
| 24 | wf_config = config |
| 25 | |
| 26 | from ms_agent.workflow.chain_workflow import ChainWorkflow |
| 27 | from ms_agent.workflow.dag_workflow import DagWorkflow |
| 28 | wf_type = ChainWorkflow.WORKFLOW_NAME.lower() |
| 29 | wf_type = getattr(wf_config, 'type', '').lower() or wf_type |
| 30 | |
| 31 | if wf_type == ChainWorkflow.WORKFLOW_NAME.lower(): |
| 32 | wf_instance = ChainWorkflow( |
| 33 | config_dir_or_id=config_dir_or_id, |
| 34 | config=wf_config, |
| 35 | env=env, |
| 36 | mcp_server_file=kwargs.get('mcp_server_file'), |
| 37 | load_cache=kwargs.get('load_cache', False), |
| 38 | trust_remote_code=trust_remote_code) |
| 39 | elif wf_type == DagWorkflow.WORKFLOW_NAME.lower(): |
| 40 | wf_instance = DagWorkflow( |
| 41 | config_dir_or_id=config_dir_or_id, |
| 42 | config=wf_config, |
| 43 | env=env, |
| 44 | mcp_server_file=kwargs.get('mcp_server_file'), |
| 45 | load_cache=kwargs.get('load_cache', False), |
| 46 | trust_remote_code=trust_remote_code) |
| 47 | elif wf_type == 'ResearchWorkflow'.lower(): |
| 48 | # TODO |
| 49 | raise NotImplementedError() |
| 50 | else: |
| 51 | raise ValueError(f'Unknown agent type: {wf_type}') |
| 52 | |
| 53 | return wf_instance |
no test coverage detected