diff --git a/docs/ColabNotebook/tutorial_notebooks/first_workflow.ipynb b/docs/ColabNotebook/tutorial_notebooks/first_workflow.ipynb index 000df16f..6c1dbcbe 100644 --- a/docs/ColabNotebook/tutorial_notebooks/first_workflow.ipynb +++ b/docs/ColabNotebook/tutorial_notebooks/first_workflow.ipynb @@ -207,14 +207,17 @@ "workflow = WorkFlow(graph=graph, agent_manager=agent_manager, llm=llm)\n", "\n", "# Execute the workflow with inputs\n", - "output = workflow.execute(\n", + "result = workflow.execute(\n", " inputs = {\n", " \"problem\": \"Write a function to find the longest palindromic substring in a given string.\"\n", " }\n", ")\n", "\n", - "print(\"Workflow completed!\")\n", - "print(\"Workflow output:\\n\", output)" + "if result.status == \"success\":\n", + " print(\"Workflow completed!\")\n", + " print(\"Workflow output:\\n\", result.result)\n", + "else:\n", + " print(\"Workflow failed:\\n\", result.displayable_error)" ] }, { @@ -276,4 +279,4 @@ }, "nbformat": 4, "nbformat_minor": 5 -} \ No newline at end of file +} diff --git a/docs/ColabNotebook/tutorial_notebooks/quickstart.ipynb b/docs/ColabNotebook/tutorial_notebooks/quickstart.ipynb index e9df5845..e84e5c67 100644 --- a/docs/ColabNotebook/tutorial_notebooks/quickstart.ipynb +++ b/docs/ColabNotebook/tutorial_notebooks/quickstart.ipynb @@ -446,8 +446,11 @@ "outputs": [], "source": [ "workflow = WorkFlow(graph=workflow_graph, agent_manager=agent_manager, llm=llm)\n", - "output = workflow.execute()\n", - "print(output)" + "result = workflow.execute()\n", + "if result.status == \"success\":\n", + " print(result.result)\n", + "else:\n", + " print(result.displayable_error)" ] }, { @@ -467,8 +470,11 @@ "\n", "nest_asyncio.apply()\n", "workflow = WorkFlow(graph=workflow_graph, agent_manager=agent_manager, llm=llm)\n", - "output = workflow.execute()\n", - "print(output)" + "result = workflow.execute()\n", + "if result.status == \"success\":\n", + " print(result.result)\n", + "else:\n", + " print(result.displayable_error)" ], "metadata": { "id": "iU2124jInCdy" @@ -500,4 +506,4 @@ }, "nbformat": 4, "nbformat_minor": 5 -} \ No newline at end of file +} diff --git a/docs/ColabNotebook/tutorial_notebooks/textgrad_optimizer.ipynb b/docs/ColabNotebook/tutorial_notebooks/textgrad_optimizer.ipynb index 254fafb7..230e41f9 100644 --- a/docs/ColabNotebook/tutorial_notebooks/textgrad_optimizer.ipynb +++ b/docs/ColabNotebook/tutorial_notebooks/textgrad_optimizer.ipynb @@ -443,7 +443,7 @@ " \"system_prompt\": \"You are a math-focused assistant dedicated to providing clear, concise, and educational solutions to mathematical problems. Your goal is to deliver structured and pedagogically sound explanations, ensuring mathematical accuracy and logical reasoning. Begin with a brief overview of the problem-solving approach, followed by detailed calculations, and conclude with a verification step. Use precise mathematical notation and consider potential edge cases. Present the final answer clearly, using the specified format, and incorporate visual aids or analogies where appropriate to enhance understanding and engagement. \\n\\nExplicitly include geometric explanations when applicable, describing the geometric context and relationships. Emphasize the importance of visual aids, such as diagrams or sketches, to enhance understanding. Ensure consistency in formatting and mathematical notation. Provide a brief explanation of the reference angle concept and its significance. Include contextual explanations of trigonometric identities and their applications. Critically evaluate initial assumptions and verify geometric properties before proceeding. Highlight the use of symmetry and conjugate pairs in complex numbers. Encourage re-evaluation and verification of steps, ensuring logical flow and clarity. Focus on deriving the correct answer and consider problem-specific strategies or known techniques.\",\n", " \"parse_mode\": \"str\",\n", " \"parse_func\": null,\n", - " \"parse_title\": null\n", + " \"title_format\": \"## {title}\"\n", " }\n", " ]\n", "}" @@ -472,4 +472,4 @@ }, "nbformat": 4, "nbformat_minor": 5 -} \ No newline at end of file +} diff --git a/docs/ColabNotebook/tutorial_notebooks_zh/first_workflow.ipynb b/docs/ColabNotebook/tutorial_notebooks_zh/first_workflow.ipynb index b5648d56..609f3149 100644 --- a/docs/ColabNotebook/tutorial_notebooks_zh/first_workflow.ipynb +++ b/docs/ColabNotebook/tutorial_notebooks_zh/first_workflow.ipynb @@ -208,14 +208,17 @@ "workflow = WorkFlow(graph=graph, agent_manager=agent_manager, llm=llm)\n", "\n", "# Execute the workflow with inputs\n", - "output = workflow.execute(\n", + "result = workflow.execute(\n", " inputs = {\n", " \"problem\": \"Write a function to find the longest palindromic substring in a given string.\"\n", " }\n", ")\n", "\n", - "print(\"Workflow completed!\")\n", - "print(\"Workflow output:\\n\", output)" + "if result.status == \"success\":\n", + " print(\"Workflow completed!\")\n", + " print(\"Workflow output:\\n\", result.result)\n", + "else:\n", + " print(\"Workflow failed:\\n\", result.displayable_error)" ] }, { @@ -281,4 +284,4 @@ }, "nbformat": 4, "nbformat_minor": 5 -} \ No newline at end of file +} diff --git a/docs/ColabNotebook/tutorial_notebooks_zh/quickstart.ipynb b/docs/ColabNotebook/tutorial_notebooks_zh/quickstart.ipynb index 32193963..be3f17f3 100644 --- a/docs/ColabNotebook/tutorial_notebooks_zh/quickstart.ipynb +++ b/docs/ColabNotebook/tutorial_notebooks_zh/quickstart.ipynb @@ -450,8 +450,11 @@ "outputs": [], "source": [ "workflow = WorkFlow(graph=workflow_graph, agent_manager=agent_manager, llm=llm)\n", - "output = workflow.execute()\n", - "print(output)" + "result = workflow.execute()\n", + "if result.status == \"success\":\n", + " print(result.result)\n", + "else:\n", + " print(result.displayable_error)" ] }, { @@ -471,8 +474,11 @@ "\n", "nest_asyncio.apply()\n", "workflow = WorkFlow(graph=workflow_graph, agent_manager=agent_manager, llm=llm)\n", - "output = workflow.execute()\n", - "print(output)" + "result = workflow.execute()\n", + "if result.status == \"success\":\n", + " print(result.result)\n", + "else:\n", + " print(result.displayable_error)" ], "metadata": { "id": "xIB1AIbbm5Dt" @@ -508,4 +514,4 @@ }, "nbformat": 4, "nbformat_minor": 5 -} \ No newline at end of file +} diff --git a/docs/ColabNotebook/tutorial_notebooks_zh/textgrad_optimizer.ipynb b/docs/ColabNotebook/tutorial_notebooks_zh/textgrad_optimizer.ipynb index f3f28653..9f3d3b26 100644 --- a/docs/ColabNotebook/tutorial_notebooks_zh/textgrad_optimizer.ipynb +++ b/docs/ColabNotebook/tutorial_notebooks_zh/textgrad_optimizer.ipynb @@ -423,7 +423,7 @@ " \"system_prompt\": \"You are a math-focused assistant dedicated to providing clear, concise, and educational solutions to mathematical problems. Your goal is to deliver structured and pedagogically sound explanations, ensuring mathematical accuracy and logical reasoning. Begin with a brief overview of the problem-solving approach, followed by detailed calculations, and conclude with a verification step. Use precise mathematical notation and consider potential edge cases. Present the final answer clearly, using the specified format, and incorporate visual aids or analogies where appropriate to enhance understanding and engagement. \\n\\nExplicitly include geometric explanations when applicable, describing the geometric context and relationships. Emphasize the importance of visual aids, such as diagrams or sketches, to enhance understanding. Ensure consistency in formatting and mathematical notation. Provide a brief explanation of the reference angle concept and its significance. Include contextual explanations of trigonometric identities and their applications. Critically evaluate initial assumptions and verify geometric properties before proceeding. Highlight the use of symmetry and conjugate pairs in complex numbers. Encourage re-evaluation and verification of steps, ensuring logical flow and clarity. Focus on deriving the correct answer and consider problem-specific strategies or known techniques.\",\n", " \"parse_mode\": \"str\",\n", " \"parse_func\": null,\n", - " \"parse_title\": null\n", + " \"title_format\": \"## {title}\"\n", " }\n", " ]\n", "}" @@ -456,4 +456,4 @@ }, "nbformat": 4, "nbformat_minor": 5 -} \ No newline at end of file +} diff --git a/docs/modules/workflow_graph.md b/docs/modules/workflow_graph.md index 24cbf416..8aa62194 100644 --- a/docs/modules/workflow_graph.md +++ b/docs/modules/workflow_graph.md @@ -74,10 +74,10 @@ The `SequentialWorkFlowGraph` accepts a simplified input format that makes it ea - `outputs` (required): List of output parameters produced by the task - `prompt` (required): The prompt template to guide the agent's behavior - `system_prompt` (optional): System message to provide context to the agent -- `output_parser` (optional): The output parser to parse the output of the task +- `output_parser` (optional): The output parser to parse the output of the task - `parse_mode` (optional): Mode for parsing outputs, defaults to "str" - `parse_func` (optional): Custom function for parsing outputs -- `parse_title` (optional): Title for the parsed output +- `title_format` (optional): Title format used when `parse_mode` is "title", e.g. "## {title}" The parameters related to prompts and parsing will be used to create a `CustomizeAgent` instance in the `agent_manager`. Please refer to the [Customize Agent](./customize_agent.md) documentation for more details about the agent configuration. @@ -183,9 +183,13 @@ agent_manager.add_agents_from_workflow(workflow_graph, llm_config=llm_config) # create a workflow instance for execution workflow = WorkFlow(graph=workflow_graph, agent_manager=agent_manager, llm=llm) -workflow.execute(inputs={"data_source": "xxx"}) +result = workflow.execute(inputs={"data_source": "xxx"}) +if result.status == "success": + print(result.result) ``` +`WorkFlow.execute()` returns a `WorkflowResult` object. By default, the workflow result is the structured output data as a `dict`. Pass `extract_output=True` if you want the older text-extraction behavior. + ### Creating a SequentialWorkFlowGraph ```python @@ -255,4 +259,4 @@ sequential_workflow_graph.save_module("examples/output/my_sequential_workflow.js workflow_graph.display() ``` -The `WorkFlowGraph` and `SequentialWorkFlowGraph` classes provide a flexible and powerful way to design complex agent workflows, track their execution, and manage the flow of data between tasks. \ No newline at end of file +The `WorkFlowGraph` and `SequentialWorkFlowGraph` classes provide a flexible and powerful way to design complex agent workflows, track their execution, and manage the flow of data between tasks. diff --git a/docs/quickstart.md b/docs/quickstart.md index 46c763d1..b3fbcc0e 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -132,9 +132,13 @@ agent_manager.add_agents_from_workflow(workflow_graph, llm_config=openai_config) Once agents are ready, you can create a `WorkFlow` instance and run it: ```python workflow = WorkFlow(graph=workflow_graph, agent_manager=agent_manager, llm=llm) -output = workflow.execute() -print(output) +result = workflow.execute() +if result.status == "success": + print(result.result) +else: + print(result.displayable_error) ``` -For a complete working example, check out the [full workflow demo](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/workflow_demo.py). +`WorkFlow.execute()` returns a `WorkflowResult` object. By default, `result.result` is the structured workflow output as a `dict`. If you want the older text-extraction behavior, call `workflow.execute(extract_output=True)`. +For a complete working example, check out the [full workflow demo](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/workflow_demo.py). diff --git a/docs/tutorial/first_workflow.md b/docs/tutorial/first_workflow.md index cf6f08ce..793c042d 100644 --- a/docs/tutorial/first_workflow.md +++ b/docs/tutorial/first_workflow.md @@ -111,17 +111,22 @@ agent_manager.add_agents_from_workflow( workflow = WorkFlow(graph=graph, agent_manager=agent_manager, llm=llm) # Execute the workflow with inputs -output = workflow.execute( +result = workflow.execute( inputs = { "problem": "Write a function to find the longest palindromic substring in a given string." } ) -print("Workflow completed!") -print("Workflow output:\n", output) +if result.status == "success": + print("Workflow completed!") + print("Workflow output:\n", result.result) +else: + print("Workflow failed:\n", result.displayable_error) ``` -You should specify all the required inputs for the workflow in the `inputs` argument of the `execute` method. +`WorkFlow.execute()` returns a `WorkflowResult` object. By default, `result.result` is the structured workflow output as a `dict`. If you want the older text-extraction behavior, call `workflow.execute(extract_output=True)`. + +You should specify all the required inputs for the workflow in the `inputs` argument of the `execute` method. For a complete working example, please refer to the [Sequential Workflow example](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/sequential_workflow.py). diff --git a/docs/tutorial/hitl.md b/docs/tutorial/hitl.md index 5878b1ea..2c44d330 100644 --- a/docs/tutorial/hitl.md +++ b/docs/tutorial/hitl.md @@ -159,6 +159,11 @@ workflow = WorkFlow(graph=graph, llm=llm, agent_manager=manager, hitl_manager=hi result = await workflow.async_execute(inputs={ "data_source": "2025Q2 financial report ..." }) + +if result.status == "success": + print(result.result) +else: + print(result.displayable_error) ``` When the interceptor runs you will see a prompt like below. Type `a` (approve) or `r` (reject): diff --git a/docs/tutorial/mcp.md b/docs/tutorial/mcp.md index 8f653440..7b4f3226 100644 --- a/docs/tutorial/mcp.md +++ b/docs/tutorial/mcp.md @@ -428,7 +428,10 @@ result = workflow.execute(inputs={ "output_format": "summary" }) -print(f"Workflow result: {result}") +if result.status == "success": + print(f"Workflow result: {result.result}") +else: + print(f"Workflow failed: {result.displayable_error}") # Clean up toolkit.disconnect() diff --git a/docs/tutorial/textgrad_optimizer.md b/docs/tutorial/textgrad_optimizer.md index b0f86d04..b9b22bcb 100644 --- a/docs/tutorial/textgrad_optimizer.md +++ b/docs/tutorial/textgrad_optimizer.md @@ -216,10 +216,10 @@ Below is an example of a saved workflow graph after optimization using `TextGrad "system_prompt": "You are a math-focused assistant dedicated to providing clear, concise, and educational solutions to mathematical problems. Your goal is to deliver structured and pedagogically sound explanations, ensuring mathematical accuracy and logical reasoning. Begin with a brief overview of the problem-solving approach, followed by detailed calculations, and conclude with a verification step. Use precise mathematical notation and consider potential edge cases. Present the final answer clearly, using the specified format, and incorporate visual aids or analogies where appropriate to enhance understanding and engagement. \n\nExplicitly include geometric explanations when applicable, describing the geometric context and relationships. Emphasize the importance of visual aids, such as diagrams or sketches, to enhance understanding. Ensure consistency in formatting and mathematical notation. Provide a brief explanation of the reference angle concept and its significance. Include contextual explanations of trigonometric identities and their applications. Critically evaluate initial assumptions and verify geometric properties before proceeding. Highlight the use of symmetry and conjugate pairs in complex numbers. Encourage re-evaluation and verification of steps, ensuring logical flow and clarity. Focus on deriving the correct answer and consider problem-specific strategies or known techniques.", "parse_mode": "str", "parse_func": null, - "parse_title": null + "title_format": "## {title}" } ] } ``` -For a complete working example, please refer to [examples/textgrad/math_textgrad.py](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/optimization/textgrad/math_textgrad.py). Additional TextGrad optimization scripts for other datasets (e.g., [`hotpotqa_textgrad.py`](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/optimization/textgrad/hotpotqa_textgrad.py) and [`mbqq_textgrad.py`](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/optimization/textgrad/mbpp_textgrad.py)) are available in the [examples/optimization/textgrad](https://github.com/EvoAgentX/EvoAgentX/tree/main/examples/optimization/textgrad) directory. \ No newline at end of file +For a complete working example, please refer to [examples/textgrad/math_textgrad.py](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/optimization/textgrad/math_textgrad.py). Additional TextGrad optimization scripts for other datasets (e.g., [`hotpotqa_textgrad.py`](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/optimization/textgrad/hotpotqa_textgrad.py) and [`mbqq_textgrad.py`](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/optimization/textgrad/mbpp_textgrad.py)) are available in the [examples/optimization/textgrad](https://github.com/EvoAgentX/EvoAgentX/tree/main/examples/optimization/textgrad) directory. diff --git a/docs/zh/modules/workflow_graph.md b/docs/zh/modules/workflow_graph.md index ea6a4ec2..014a1091 100644 --- a/docs/zh/modules/workflow_graph.md +++ b/docs/zh/modules/workflow_graph.md @@ -76,7 +76,7 @@ - `output_parser`（可选）：用于解析任务输出的解析器 - `parse_mode`（可选）：解析输出的模式，默认为 "str" - `parse_func`（可选）：用于解析输出的自定义函数 -- `parse_title`（可选）：解析输出的标题 +- `title_format`（可选）：当 `parse_mode` 为 "title" 时使用的标题格式，例如 "## {title}" 与提示和解析相关的参数将用于在 `agent_manager` 中创建 `CustomizeAgent` 实例。有关代理配置的更多详细信息，请参阅[自定义代理](./customize_agent.md)文档。 @@ -182,9 +182,13 @@ agent_manager.add_agents_from_workflow(workflow_graph, llm_config=llm_config) # 创建工作流实例以执行 workflow = WorkFlow(graph=workflow_graph, agent_manager=agent_manager, llm=llm) -workflow.execute(inputs={"data_source": "xxx"}) +result = workflow.execute(inputs={"data_source": "xxx"}) +if result.status == "success": + print(result.result) ``` +`WorkFlow.execute()` 会返回一个 `WorkflowResult` 对象。默认情况下，工作流结果是结构化的 `dict` 输出。若想保留旧的文本抽取行为，可以传入 `extract_output=True`。 + ### 创建顺序工作流图 ```python @@ -254,4 +258,4 @@ sequential_workflow_graph.save_module("examples/output/my_sequential_workflow.js workflow_graph.display() ``` -`WorkFlowGraph` 和 `SequentialWorkFlowGraph` 类提供了一种灵活而强大的方式来设计复杂的代理工作流、跟踪其执行并管理任务之间的数据流。 \ No newline at end of file +`WorkFlowGraph` 和 `SequentialWorkFlowGraph` 类提供了一种灵活而强大的方式来设计复杂的代理工作流、跟踪其执行并管理任务之间的数据流。 diff --git a/docs/zh/quickstart.md b/docs/zh/quickstart.md index 74189c1d..cf864cda 100644 --- a/docs/zh/quickstart.md +++ b/docs/zh/quickstart.md @@ -130,8 +130,13 @@ agent_manager.add_agents_from_workflow(workflow_graph, llm_config=openai_config) 代理准备就绪后，可以创建 `WorkFlow` 实例并运行： ```python workflow = WorkFlow(graph=workflow_graph, agent_manager=agent_manager, llm=llm) -output = workflow.execute() -print(output) +result = workflow.execute() +if result.status == "success": + print(result.result) +else: + print(result.displayable_error) ``` -更多示例请参见 [完整工作流演示](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/workflow_demo.py)。 \ No newline at end of file +`WorkFlow.execute()` 会返回一个 `WorkflowResult` 对象。默认情况下，`result.result` 是结构化的工作流输出，类型为 `dict`。如果你想保留旧的文本抽取行为，可以调用 `workflow.execute(extract_output=True)`。 + +更多示例请参见 [完整工作流演示](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/workflow_demo.py)。 diff --git a/docs/zh/tutorial/first_workflow.md b/docs/zh/tutorial/first_workflow.md index 4d2bd6e1..82657a8d 100644 --- a/docs/zh/tutorial/first_workflow.md +++ b/docs/zh/tutorial/first_workflow.md @@ -111,16 +111,21 @@ agent_manager.add_agents_from_workflow( workflow = WorkFlow(graph=graph, agent_manager=agent_manager, llm=llm) # Execute the workflow with inputs -output = workflow.execute( +result = workflow.execute( inputs = { "problem": "Write a function to find the longest palindromic substring in a given string." } ) -print("Workflow completed!") -print("Workflow output:\n", output) +if result.status == "success": + print("Workflow completed!") + print("Workflow output:\n", result.result) +else: + print("Workflow failed:\n", result.displayable_error) ``` +`WorkFlow.execute()` 会返回一个 `WorkflowResult` 对象。默认情况下，`result.result` 是结构化的工作流输出，类型为 `dict`。如果你想保留旧的文本抽取行为，可以调用 `workflow.execute(extract_output=True)`。 + 你应该在 `execute` 方法的 `inputs` 参数中指定工作流所需的所有输入。 有关完整的工作示例，请参考 [顺序工作流示例](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/sequential_workflow.py)。 diff --git a/docs/zh/tutorial/textgrad_optimizer.md b/docs/zh/tutorial/textgrad_optimizer.md index 20fa04a4..d76b6be4 100644 --- a/docs/zh/tutorial/textgrad_optimizer.md +++ b/docs/zh/tutorial/textgrad_optimizer.md @@ -196,10 +196,10 @@ print(f"Evaluation result (after optimization):\n{result}") "system_prompt": "You are a math-focused assistant dedicated to providing clear, concise, and educational solutions to mathematical problems. Your goal is to deliver structured and pedagogically sound explanations, ensuring mathematical accuracy and logical reasoning. Begin with a brief overview of the problem-solving approach, followed by detailed calculations, and conclude with a verification step. Use precise mathematical notation and consider potential edge cases. Present the final answer clearly, using the specified format, and incorporate visual aids or analogies where appropriate to enhance understanding and engagement. \n\nExplicitly include geometric explanations when applicable, describing the geometric context and relationships. Emphasize the importance of visual aids, such as diagrams or sketches, to enhance understanding. Ensure consistency in formatting and mathematical notation. Provide a brief explanation of the reference angle concept and its significance. Include contextual explanations of trigonometric identities and their applications. Critically evaluate initial assumptions and verify geometric properties before proceeding. Highlight the use of symmetry and conjugate pairs in complex numbers. Encourage re-evaluation and verification of steps, ensuring logical flow and clarity. Focus on deriving the correct answer and consider problem-specific strategies or known techniques.", "parse_mode": "str", "parse_func": null, - "parse_title": null + "title_format": "## {title}" } ] } ``` -完整的工作示例，请参阅 [examples/textgrad/math_textgrad.py](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/optimization/textgrad/math_textgrad.py)。其他数据集的 TextGrad 优化脚本（例如，[`hotpotqa_textgrad.py`](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/optimization/textgrad/hotpotqa_textgrad.py) 和 [`mbqq_textgrad.py`](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/optimization/textgrad/mbpp_textgrad.py)）可以在 [examples/optimization/textgrad](https://github.com/EvoAgentX/EvoAgentX/tree/main/examples/optimization/textgrad) 目录中找到。 \ No newline at end of file +完整的工作示例，请参阅 [examples/textgrad/math_textgrad.py](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/optimization/textgrad/math_textgrad.py)。其他数据集的 TextGrad 优化脚本（例如，[`hotpotqa_textgrad.py`](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/optimization/textgrad/hotpotqa_textgrad.py) 和 [`mbqq_textgrad.py`](https://github.com/EvoAgentX/EvoAgentX/blob/main/examples/optimization/textgrad/mbpp_textgrad.py)）可以在 [examples/optimization/textgrad](https://github.com/EvoAgentX/EvoAgentX/tree/main/examples/optimization/textgrad) 目录中找到。 diff --git a/evoagentx/actions/customize_action.py b/evoagentx/actions/customize_action.py index ab8b2b1e..ce8badda 100644 --- a/evoagentx/actions/customize_action.py +++ b/evoagentx/actions/customize_action.py @@ -3,7 +3,6 @@ import re import uuid from collections.abc import Callable -from concurrent.futures import ThreadPoolExecutor from typing import List, Optional, Union from pydantic import Field, PositiveInt @@ -28,6 +27,7 @@ ) from ..prompts.utils import DEFAULT_SYSTEM_PROMPT from ..tools.tool import Tool, Toolkit, ToolMetadata, ToolResult +from ..utils.async_utils import run_coroutine_sync from ..utils.utils import compile_tool_schemas, pydantic_to_parameters from .action import Action @@ -322,18 +322,7 @@ def execute( return_prompt=return_prompt, **kwargs ) - - try: - asyncio.get_running_loop() - except RuntimeError: - # No event loop is running in this thread: drive the coroutine directly. - return asyncio.run(coro) - - # We are already inside a running event loop (e.g. `execute` was called from - # async code). `asyncio.run()` would raise `RuntimeError` here, so run the - # coroutine to completion on a dedicated thread that owns its own event loop. - with ThreadPoolExecutor(max_workers=1) as pool: - return pool.submit(asyncio.run, coro).result() + return run_coroutine_sync(coro) async def async_execute( self, diff --git a/evoagentx/agents/agent.py b/evoagentx/agents/agent.py index 242ae6ab..72fbae5d 100644 --- a/evoagentx/agents/agent.py +++ b/evoagentx/agents/agent.py @@ -1,5 +1,4 @@ import asyncio -import inspect from collections.abc import Coroutine from typing import Any, Dict, List, Optional, Tuple, Type, Union @@ -16,6 +15,7 @@ from ..models.base_model import BaseLLM from ..models.model_configs import LLMConfig from ..storages.base import StorageHandler +from ..utils.async_utils import call_maybe_async, is_method_overridden from ..utils.utils import add_llm_config_to_agent_dict @@ -198,24 +198,22 @@ async def async_execute( **kwargs ) - # execute action asynchronously - async_execute_source = inspect.getsource(action.async_execute) - if "NotImplementedError" in async_execute_source: - # if the async_execute method is not implemented, use the execute method instead - execution_results = action.execute( - llm=self.llm, - inputs=action_input_data, - sys_msg=self.system_prompt, - return_prompt=True, - **kwargs - ) + if is_method_overridden(action, Action, "async_execute"): + execute_function = action.async_execute + elif is_method_overridden(action, Action, "execute"): + execute_function = action.execute else: - execution_results = await action.async_execute( - llm=self.llm, - inputs=action_input_data, - sys_msg=self.system_prompt, - return_prompt=True, - **kwargs + raise NotImplementedError( + f"The action '{type(action).__name__}' must implement `execute` or `async_execute`." + ) + + execution_results = await call_maybe_async( + execute_function, + llm=self.llm, + inputs=action_input_data, + sys_msg=self.system_prompt, + return_prompt=True, + **kwargs ) action_output, prompt = execution_results diff --git a/evoagentx/core/base_config.py b/evoagentx/core/base_config.py index e258753c..c7d93d0f 100644 --- a/evoagentx/core/base_config.py +++ b/evoagentx/core/base_config.py @@ -73,9 +73,14 @@ class Parameter(BaseModule): @model_validator(mode="after") def _validate_type_and_schema(self): - from ..utils.utils import string_to_json_schema_type, string_to_python_type + from ..utils.utils import normalize_param_type, string_to_json_schema_type, string_to_python_type if self.type not in string_to_python_type: - raise ValueError(f"Invalid `type`: {self.type}. Allowed: {list(string_to_python_type.keys())}") + # LLM-generated specs may emit synonyms (e.g. "List[str]", "text"); map those to + # canonical types. Truly unrecognized types (e.g. "other_type") still raise. + normalized = normalize_param_type(self.type) + if normalized is None: + raise ValueError(f"Invalid `type`: {self.type}. Allowed: {list(string_to_python_type.keys())}") + self.type = normalized if self.json_schema is not None: try: Draft7Validator.check_schema(self.json_schema) diff --git a/evoagentx/evaluators/evaluator.py b/evoagentx/evaluators/evaluator.py index 986ad7bb..a6790196 100644 --- a/evoagentx/evaluators/evaluator.py +++ b/evoagentx/evaluators/evaluator.py @@ -129,7 +129,13 @@ def _execute_workflow_graph(self, graph: WorkFlowGraph, inputs: dict, return_tra graph_copy = WorkFlowGraph(goal=graph.goal, graph=graph) graph_copy.reset_graph() # reset the status of all nodes to pending workflow = WorkFlow(llm=self.llm, graph=graph_copy, agent_manager=self.agent_manager, **kwargs) - output: str = workflow.execute(inputs=inputs, **kwargs) + result = workflow.execute(inputs=inputs, extract_output=True, **kwargs) + if result.status == "success": + output = result.result + if not isinstance(output, str): + output = str(output) + else: + output = result.displayable_error or "Workflow Execution Failed" if return_trajectory: return output, workflow.environment.get() return output @@ -509,8 +515,13 @@ async def _async_execute_workflow_graph(self, graph: WorkFlowGraph, inputs: dict **kwargs ) - output: str = await workflow.async_execute(inputs=inputs, **kwargs) + result = await workflow.async_execute(inputs=inputs, extract_output=True, **kwargs) + if result.status == "success": + output = result.result + if not isinstance(output, str): + output = str(output) + else: + output = result.displayable_error or "Workflow Execution Failed" if return_trajectory: return output, workflow.environment.get() return output - diff --git a/evoagentx/optimizers/mipro_optimizer.py b/evoagentx/optimizers/mipro_optimizer.py index 43ebca80..02e5df89 100644 --- a/evoagentx/optimizers/mipro_optimizer.py +++ b/evoagentx/optimizers/mipro_optimizer.py @@ -1313,7 +1313,13 @@ def __call__(self, **input_data): else: # use the original executor llm to execute the graph workflow = WorkFlow(llm=self.executor_llm, graph=new_graph, agent_manager=self.agent_manager) - output: str = workflow.execute(inputs=self.collate_func(input_data)) + result = workflow.execute(inputs=self.collate_func(input_data), extract_output=True) + if result.status == "success": + output = result.result + if not isinstance(output, str): + output = str(output) + else: + output = result.displayable_error or "Workflow Execution Failed" output = self.output_postprocess_func(output) # extract all the input and output data from the workflow execution @@ -1608,4 +1614,4 @@ def _register_optimizable_parameters(self, program: WorkFlowGraphProgram): return registry - \ No newline at end of file + diff --git a/evoagentx/prompts/agent_generator.py b/evoagentx/prompts/agent_generator.py index f5584197..af3a1dcd 100644 --- a/evoagentx/prompts/agent_generator.py +++ b/evoagentx/prompts/agent_generator.py @@ -31,7 +31,7 @@ "inputs": [ {{ "name": "the input's name", - "type": "string/int/float/other_type", + "type": "string/integer/number/boolean/object/array", "description": "Description of the input's purpose and usage." }}, ... @@ -39,7 +39,7 @@ "outputs": [ {{ "name": "the output's name", - "type": "string/int/float/other_type", + "type": "string/integer/number/boolean/object/array", "description": "Description of the output produced by this sub-task." }}, ... @@ -62,7 +62,7 @@ "inputs": [ {{ "name": "the input's name", - "type": "string/int/float/other_type", + "type": "string/integer/number/boolean/object/array", "description": "Description of the input's purpose and usage." }}, ... @@ -70,7 +70,7 @@ "outputs": [ {{ "name": "the output's name", - "type": "string/int/float/other_type", + "type": "string/integer/number/boolean/object/array", "description": "Description of the output produced by this agent." }}, ... @@ -181,7 +181,7 @@ "inputs": [ {{ "name": "the input's name", - "type": "string/int/float/other_type", + "type": "string/integer/number/boolean/object/array", "required": true/false (`false` means the input is the feedback from later sub-task, or the previous output for the current sub-task), "description": "Description of the input's purpose and usage." }}, @@ -190,7 +190,7 @@ "outputs": [ {{ "name": "the output's name", - "type": "string/int/float/other_type", + "type": "string/integer/number/boolean/object/array", "required": true (the `required` field of outputs are always true), "description": "Description of the output produced by this sub-task." }}, @@ -214,7 +214,7 @@ "inputs": [ {{ "name": "the input's name", - "type": "string/int/float/other_type", + "type": "string/integer/number/boolean/object/array", "required": true/false (only set to `false` when this input is the feedback from later sub-task, or the previous generated output for the current sub-task), "description": "Description of the input's purpose and usage." }}, @@ -223,7 +223,7 @@ "outputs": [ {{ "name": "the output's name", - "type": "string/int/float/other_type", + "type": "string/integer/number/boolean/object/array", "required": true (always set the `required` field of outputs as true), "description": "Description of the output produced by this agent." }}, diff --git a/evoagentx/prompts/task_planner.py b/evoagentx/prompts/task_planner.py index 71808cec..6b6176ae 100644 --- a/evoagentx/prompts/task_planner.py +++ b/evoagentx/prompts/task_planner.py @@ -39,7 +39,7 @@ "inputs": [ {{ "name": "the input's name", - "type": "string/int/float/other_type", + "type": "string/integer/number/boolean/object/array", "description": "Description of the input's purpose and usage." }}, ... @@ -47,7 +47,7 @@ "outputs": [ {{ "name": "the output's name", - "type": "string/int/float/other_type", + "type": "string/integer/number/boolean/object/array", "description": "Description of the output produced by this sub-task." }}, ... @@ -148,7 +148,7 @@ "inputs": [ {{ "name": "the input's name", - "type": "string/int/float/other_type", + "type": "string/integer/number/boolean/object/array", "required": true/false (only set to `false` when this input is the feedback from later sub-task, or the previous generated output for the current sub-task), "description": "Description of the input's purpose and usage." }}, @@ -157,7 +157,7 @@ "outputs": [ {{ "name": "the output's name", - "type": "string/int/float/other_type", + "type": "string/integer/number/boolean/object/array", "required": true (always set the `required` field of outputs as true), "description": "Description of the output produced by this sub-task." }}, diff --git a/evoagentx/prompts/workflow/workflow_manager.py b/evoagentx/prompts/workflow/workflow_manager.py index a813fdc0..92b778e4 100644 --- a/evoagentx/prompts/workflow/workflow_manager.py +++ b/evoagentx/prompts/workflow/workflow_manager.py @@ -3,6 +3,55 @@ DEFAULT_TASK_SCHEDULER_PROMPT = """ ### objective Your task is to analyze the given workflow graph, current execution information, and candidate subtasks to decide one of the following actions: +- Select a subtask for iterative execution (if there is a loop or iterative context in the workflow). You may only iterate on each task up to {max_num_turns} times, also determined by the `Workflow Execution History`. +- Select a new subtask from the candidates to move the workflow forward (if the workflow should proceed without re-executing or iterating). + +### Instructions +1. Review the Workflow Information for the structure and details of the tasks and any potential loops or iterative sections. +2. Check the Current Execution Information for evidence of errors or missing data from previously executed subtasks. +3. If the workflow graph indicates there is a loop or iterative context, select a subtask from the Candidate Subtasks that aligns with that iterative or looping goal. +4. Otherwise, select a subtask from the Candidate Subtasks that best moves the workflow forward. +5. Finally, output the decision in the required format. + +### Output Format +Your final output should ALWAYS in the following format: + +## Thought +Provide a brief explanation of your reasoning for scheduling the next task. + +## Scheduled Subtask +Produce your answer in valid JSON with the following structure: +```json +{{ + "decision": "iterate | forward", + "task_name": "name of the scheduled subtask", + "reason": "the reasoning for scheduling this subtask" +}} +``` + +----- +lets' begin + +Here is the information for your decision: + +### Workflow Information: +{workflow_graph_representation} + +### Workflow Execution History: +{execution_history} + +### Workflow Execution Outputs: +{execution_outputs} + +### Candidate Subtasks: +{candidate_tasks} + +Output: +""" + +OLD_DEFAULT_TASK_SCHEDULER_PROMPT = """ +### objective +Your task is to analyze the given workflow graph, current execution information, and candidate subtasks to decide one of the following actions: - Re-execute a previous subtask to correct errors or gather missing information (if a previous subtask's result is erroneous or incomplete). You may only re-execute each task up to {max_num_turns} times, based on the `Workflow Execution History`. - Select a subtask for iterative execution (if there is a loop or iterative context in the workflow). You may only iterate on each task up to {max_num_turns} times, also determined by the `Workflow Execution History`. - Select a new subtask from the candidates to move the workflow forward (if the workflow should proceed without re-executing or iterating). @@ -123,7 +172,7 @@ } -OUTPUT_EXTRACTION_PROMPT = """ +WORKFLOW_OUTPUT_EXTRACTION_PROMPT = """ ### Objective Your goal is to read the Workflow Goal, the WorkFlow Information, and the WorkFlow Execution Results from the provided input. Then, based on those details, extract and present ONLY the FINAL output data that meets the Workflow Goal. diff --git a/evoagentx/utils/async_utils.py b/evoagentx/utils/async_utils.py new file mode 100644 index 00000000..e67bd58f --- /dev/null +++ b/evoagentx/utils/async_utils.py @@ -0,0 +1,55 @@ +import asyncio +import inspect +from collections.abc import Awaitable, Callable +from concurrent.futures import ThreadPoolExecutor +from typing import Any, Type, TypeVar + + +T = TypeVar("T") + + +def run_coroutine_sync(coro: Awaitable[T]) -> T: + """ + Run an awaitable from synchronous code, including when the caller is already + inside a running event loop. + """ + try: + asyncio.get_running_loop() + except RuntimeError: + return asyncio.run(coro) + + with ThreadPoolExecutor(max_workers=1) as pool: + return pool.submit(asyncio.run, coro).result() + + +def is_method_overridden(instance: Any, base_cls: Type[Any], method_name: str) -> bool: + """ + Return whether ``method_name`` is implemented on ``instance``'s class rather + than inherited unchanged from ``base_cls``. + """ + base_method = _normalize_method(getattr(base_cls, method_name, None)) + instance_dict = getattr(instance, "__dict__", {}) + + if method_name in instance_dict: + instance_method = _normalize_method(inspect.unwrap(instance_dict[method_name])) + return instance_method is not None and instance_method is not base_method + + instance_method = _normalize_method(getattr(type(instance), method_name, None)) + return instance_method is not None and instance_method is not base_method + + +def _normalize_method(method: Any) -> Any: + return getattr(method, "__func__", method) + + +async def call_maybe_async(func: Callable[..., T], *args: Any, **kwargs: Any) -> T: + """ + Await async callables directly and run sync callables in a worker thread. + """ + if inspect.iscoroutinefunction(func): + return await func(*args, **kwargs) + + result = await asyncio.to_thread(func, *args, **kwargs) + if inspect.isawaitable(result): + return await result + return result diff --git a/evoagentx/utils/utils.py b/evoagentx/utils/utils.py index e44cf7e8..94e44be7 100644 --- a/evoagentx/utils/utils.py +++ b/evoagentx/utils/utils.py @@ -282,7 +282,12 @@ def validate_param( actual_params_name: str, ): """ - Checks if `actual_param` has the same type, required, description and json_schema value as `required_param`. + Checks if `actual_param` is compatible with `required_param`. + + Only the attributes that affect runtime behavior are strictly enforced: `type` and + `required`. `description` is free-form text and is not compared. If the required + parameter provides a `json_schema`, the actual parameter must provide the same + schema so structured contracts cannot be dropped downstream. """ def format_error_msg( @@ -306,11 +311,11 @@ def format_error_msg( if required_param.required != actual_param.required: raise ValueError(format_error_msg("required", required_param.required, actual_param.required)) - if required_param.description != actual_param.description: - raise ValueError(format_error_msg("description", required_param.description, actual_param.description)) - - if required_param.json_schema != actual_param.json_schema: - raise ValueError(format_error_msg("json_schema", required_param.json_schema, actual_param.json_schema)) + # `json_schema` is optional for parameters in general, but once an upstream + # contract provides one, downstream params must preserve it. + if required_param.json_schema is not None: + if required_param.json_schema != actual_param.json_schema: + raise ValueError(format_error_msg("json_schema", required_param.json_schema, actual_param.json_schema)) def format_validation_error(error: ValidationError) -> str: @@ -354,51 +359,6 @@ def params_to_json(params: List[Parameter], ignore: List[str] = []) -> str: return params_json -def fix_property_name(object: Any, json_schema: Dict) -> Any: - """ - Recursively fixes the property names of `object` to match the provided JSON schema. - """ - if object is None: - return object - - if json_schema["type"] == "array" and json_schema["items"]["type"] == "object": - return [fix_property_name(item, json_schema["items"]) for item in object] - - elif json_schema["type"] == "object": - fixed_object = dict() - properties = json_schema.get("properties") - - if properties is None: - return object - - for property_name, property_schema in properties.items(): - - if property_schema["type"] == "array": - property = object.get(property_name, None) - if property is not None: - fixed_object[property_name] = [fix_property_name(item, property_schema["items"]) for item in property] - - elif property_schema["type"] == "object": - property = object.get(property_name, None) - if property is not None: - fixed_object[property_name] = fix_property_name(property, property_schema) - - else: - object_properties_lower = {name.lower(): name for name in object} - schema_properties_lower = {name.lower(): name for name in properties} - - for name in object_properties_lower: - if name in schema_properties_lower: - fixed_object[schema_properties_lower[name]] = object[object_properties_lower[name]] - else: - fixed_object[object_properties_lower[name]] = object[object_properties_lower[name]] - - return fixed_object - - else: - return object - - def resolve_json_schema_ref(json_schema: Any, root_schema: Optional[Dict] = None) -> Any: """ Recursively resolve all $ref in a JSON schema. @@ -683,6 +643,39 @@ def wrapped_fn(*args, **kwargs): "list": list, } +# Common aliases LLMs emit for parameter types, mapped to canonical names. +_param_type_aliases = { + "text": "string", + "char": "string", + "double": "number", + "long": "integer", + "json": "object", + "dictionary": "object", + "map": "object", + "tuple": "array", + "set": "array", +} + + +def normalize_param_type(type_str: str) -> Optional[str]: + """Best-effort normalization of a parameter `type` string into a canonical type. + + Handles casing/whitespace, common aliases, and parametrized forms such as + ``List[str]`` or ``Dict[str, int]``. Returns ``None`` when the type cannot be + recognized, so the caller can decide how to handle it (e.g. raise an error). + """ + if not isinstance(type_str, str): + return None + normalized = type_str.strip().lower() + # Strip parametrization, e.g. "list[str]" -> "list", "dict[str, int]" -> "dict". + base = normalized.split("[", 1)[0].strip() + if base in string_to_python_type: + return base + if base in _param_type_aliases: + return _param_type_aliases[base] + return None + + json_to_python_type = { "string": str, "integer": int, diff --git a/evoagentx/workflow/__init__.py b/evoagentx/workflow/__init__.py index 691324ed..e32d5746 100644 --- a/evoagentx/workflow/__init__.py +++ b/evoagentx/workflow/__init__.py @@ -6,13 +6,14 @@ # from .controller import * from .workflow_generator import WorkFlowGenerator from .workflow_graph import WorkFlowGraph, SequentialWorkFlowGraph, SEWWorkFlowGraph -from .workflow import WorkFlow +from .workflow import WorkFlow, WorkflowResult from .action_graph import ActionGraph, QAActionGraph __all__ = [ "WorkFlowGenerator", "WorkFlowGraph", "WorkFlow", + "WorkflowResult", "ActionGraph", "QAActionGraph", "SequentialWorkFlowGraph", diff --git a/evoagentx/workflow/environment.py b/evoagentx/workflow/environment.py index d80123df..6064614a 100644 --- a/evoagentx/workflow/environment.py +++ b/evoagentx/workflow/environment.py @@ -1,6 +1,6 @@ from enum import Enum from pydantic import Field -from typing import Union, Optional, List +from typing import Union, Optional, List, Dict from ..core.module import BaseModule from ..core.message import Message, MessageType from ..models.base_model import LLMOutputParser @@ -30,6 +30,15 @@ class Environment(BaseModule): task_execution_history: List[str] = Field(default_factory=list) execution_data: dict = Field(default_factory=dict) + def reset(self): + """ + Clear all intermediate execution state so the environment can be reused for a + fresh workflow run without leaking data from a previous execution. + """ + self.trajectory = [] + self.task_execution_history = [] + self.execution_data = {} + def update(self, message: Message, state: TrajectoryState = None, error: str = None, **kwargs): """ Add a message to the shared memory and optionally to a specific task's message list. @@ -99,13 +108,12 @@ def get_last_executed_task(self) -> str: def get_all_execution_data(self) -> dict: return self.execution_data - def get_execution_data(self, params: Union[str, List[str]]) -> dict: - if isinstance(params, str): - params = [params] + def get_execution_data(self, params: Dict[str, bool]) -> dict: data = {} - for param in params: - if param not in self.execution_data: + for param, required in params.items(): + if param in self.execution_data: + data[param] = self.execution_data[param] + elif required: raise KeyError(f"Couldn't find execution data with key '{param}'. Available execution data: {list(self.execution_data.keys())}") - data[param] = self.execution_data[param] return data diff --git a/evoagentx/workflow/model_selector.py b/evoagentx/workflow/model_selector.py new file mode 100644 index 00000000..3e68542b --- /dev/null +++ b/evoagentx/workflow/model_selector.py @@ -0,0 +1,180 @@ +from abc import ABC, abstractmethod +from typing import Dict, Optional, Union + +from ..agents import Agent +from ..models import LLMConfig + + +class ModelSelector(ABC): + + def validate_agent(self, agent: Union[Agent, Dict]) -> None: + if not isinstance(agent, Agent) and not isinstance(agent, dict): + raise TypeError(f"Unsupported agent type: {type(agent)}") + + @abstractmethod + def get_model(self, agent: Union[Agent, Dict]) -> LLMConfig: + pass + + +class DefaultModelSelector(ModelSelector): + """ + The default model selector, which will return the same LLMConfig for all agents. + """ + + def __init__(self, llm_config: LLMConfig, override: bool = True): + """ + Args: + llm_config: The LLMConfig to use for all agents. + override: Whether to override the existingLLMConfig of agents. + """ + self.llm_config = llm_config + self.override = override + + def get_model(self, agent: Union[Agent, Dict]) -> LLMConfig: + self.validate_agent(agent) + + if self.override: + return self.llm_config + else: + if isinstance(agent, Agent): + llm_config = getattr(agent, "llm_config", None) + if llm_config: + return llm_config + return self.llm_config + + elif isinstance(agent, dict): + llm_config = agent.get("llm_config") + if llm_config: + if isinstance(llm_config, dict): + return LLMConfig.from_dict(llm_config) + return llm_config + return self.llm_config + + else: + raise TypeError(f"Unsupported agent type: {type(agent)}") + + +class SimpleModelSelector(ModelSelector): + """ + The simple model selector provides one LLMConfig for agents without tools and another for agents with tools. + """ + + def __init__( + self, + llm_config_no_tools: LLMConfig, + llm_config_with_tools: LLMConfig, + override: bool = True + ): + """ + Args: + llm_config_no_tools: The LLMConfig to use for agents without tools. + llm_config_with_tools: The LLMConfig to use for agents with tools. + override: Whether to override the existing LLMConfig of agents. + """ + self.llm_config_no_tools = llm_config_no_tools + self.llm_config_with_tools = llm_config_with_tools + self.override = override + + def get_model(self, agent: Union[Agent, Dict]) -> LLMConfig: + self.validate_agent(agent) + + if isinstance(agent, Agent): + llm_config = getattr(agent, "llm_config", None) + if llm_config and not self.override: + return llm_config + + tools = getattr(agent, "tools", None) + if tools: + return self.llm_config_with_tools + return self.llm_config_no_tools + + elif isinstance(agent, dict): + llm_config = agent.get("llm_config") + if llm_config and not self.override: + if isinstance(llm_config, dict): + return LLMConfig.from_dict(llm_config) + return llm_config + + tools = agent.get("tools") + tool_names = agent.get("tool_names") + if tools or tool_names: + return self.llm_config_with_tools + return self.llm_config_no_tools + + else: + raise TypeError(f"Unsupported agent type: {type(agent)}") + + +class ToolBasedModelSelector(SimpleModelSelector): + """ + The tool-based model selector provides specific LLMConfigs based on the tools an agent possesses. + """ + + def __init__( + self, + llm_config_no_tools: LLMConfig, + llm_config_with_tools: LLMConfig, + tool_to_llm_config: Optional[Dict[str, LLMConfig]] = None, + override: bool = True + ): + """ + Args: + llm_config_no_tools: The LLMConfig to use for agents without tools. + llm_config_with_tools: The LLMConfig to use for agents with tools. + tool_to_llm_config: A dictionary mapping tool names to LLMConfigs. + override: Whether to override the existing LLMConfig of agents. + """ + super().__init__(llm_config_no_tools, llm_config_with_tools, override) + self.tool_to_llm_config = tool_to_llm_config + + + def get_model(self, agent: Union[Agent, Dict]) -> LLMConfig: + self.validate_agent(agent) + + if self.tool_to_llm_config is None: + return super().get_model(agent) + + if isinstance(agent, Agent): + llm_config = getattr(agent, "llm_config", None) + if llm_config and not self.override: + return llm_config + + tools = getattr(agent, "tools", None) + if tools: + for tool in tools: + if tool.name in self.tool_to_llm_config: + return self.tool_to_llm_config[tool.name] + return self.llm_config_with_tools + return self.llm_config_no_tools + + elif isinstance(agent, dict): + llm_config = agent.get("llm_config") + if llm_config and not self.override: + if isinstance(llm_config, dict): + return LLMConfig.from_dict(llm_config) + return llm_config + + tools = agent.get("tools") + tool_names = agent.get("tool_names") + + if tools: + for tool in tools: + if isinstance(tool, dict): + tool_name = tool.get("name") + else: + tool_name = getattr(tool, "name", None) + + if tool_name in self.tool_to_llm_config: + return self.tool_to_llm_config[tool_name] + return self.llm_config_with_tools + + elif tool_names: + for tool_name in tool_names: + if tool_name in self.tool_to_llm_config: + return self.tool_to_llm_config[tool_name] + return self.llm_config_with_tools + + return self.llm_config_no_tools + + else: + raise TypeError(f"Unsupported agent type: {type(agent)}") \ No newline at end of file diff --git a/evoagentx/workflow/workflow.py b/evoagentx/workflow/workflow.py index 5b6ceaba..4552edec 100644 --- a/evoagentx/workflow/workflow.py +++ b/evoagentx/workflow/workflow.py @@ -1,24 +1,34 @@ import inspect -import asyncio +import traceback from copy import deepcopy -from pydantic import Field, create_model -from typing import Optional, List +from pydantic import Field, ValidationError, create_model +from typing import Literal, Optional, List, Union from ..core.logging import logger +from ..core.exception import DisplayableException, InputValidationError from ..core.module import BaseModule from ..core.message import Message, MessageType from ..core.module_utils import generate_id from ..models.base_model import BaseLLM from ..agents.agent import Agent from ..agents.agent_manager import AgentManager, AgentState +from ..agents.customize_agent import CustomizeAgent from ..storages.base import StorageHandler from .environment import Environment, TrajectoryState from .workflow_manager import WorkFlowManager, NextAction from .workflow_graph import WorkFlowNode, WorkFlowGraph from .action_graph import ActionGraph from ..hitl import HITLManager, HITLBaseAgent -from ..utils.utils import generate_dynamic_class_name +from ..utils.async_utils import call_maybe_async, is_method_overridden, run_coroutine_sync +from ..utils.utils import generate_dynamic_class_name, format_validation_error from ..actions import ActionInput, ActionOutput + +class WorkflowResult(BaseModule): + status: Literal["success", "failed"] + result: Optional[Union[dict, str]] = None + error_msg: Optional[str] = Field(default=None, description="Contains error message and traceback") + displayable_error: Optional[str] = Field(default=None, description="This is the error message that can be displayed to the user") + class WorkFlow(BaseModule): graph: WorkFlowGraph @@ -37,54 +47,130 @@ def init_module(self): if self.llm is None: raise ValueError("Must provide `llm` when `workflow_manager` is None") self.workflow_manager = WorkFlowManager(llm=self.llm) + self._validate_agent_manager() + + self.graph.validate_workflow_graph() + self.output_names = {output.name: output.required for output in self.graph.workflow_outputs} + + def _validate_agent_manager(self): + """ + Validate that ``agent_manager`` can satisfy the workflow's agent-based nodes. + + A node executed by an ``action_graph`` needs no agents, but any other node is + scheduled through ``agent_manager.get_agent()`` at runtime (see + ``WorkFlowManager._prepare_action_execution``). For those nodes we require, at + init time, that an ``agent_manager`` is present and contains every referenced + agent, so that a misconfiguration fails fast instead of crashing mid-execution. + """ + # Collect the agents required by nodes that are not pure ActionGraph nodes. + required_agents = {} + for node in self.graph.nodes: + if node.action_graph is not None: + continue + agent_names = node.get_agents() + if not agent_names: + # A node with neither an action_graph nor agents is an invalid node; + # this is reported by graph.validate_workflow_graph(). + continue + for agent_name in agent_names: + required_agents.setdefault(agent_name, node.name) + + if not required_agents: + # Pure ActionGraph workflow: no agent_manager needed. + return + if self.agent_manager is None: - logger.warning("agent_manager is NoneType when initializing a WorkFlow instance") + raise ValueError( + "agent_manager is required because the workflow contains agent-based " + f"node(s): {sorted(required_agents.values())}. The following agents " + f"must be provided: {sorted(required_agents)}." + ) - def execute(self, inputs: dict = {}, **kwargs) -> str: + missing_agents = [name for name in required_agents if not self.agent_manager.has_agent(name)] + if missing_agents: + raise ValueError( + "agent_manager is missing agent(s) required by the workflow: " + f"{sorted(missing_agents)}. Available agents: {sorted(self.agent_manager.list_agents())}." + ) + + def execute(self, inputs: Optional[dict] = None, extract_output: bool = False, **kwargs) -> WorkflowResult: """ - Synchronous wrapper for async_execute. Creates a new event loop and runs the async method. - + Synchronous wrapper for async_execute. + Args: inputs: Dictionary of inputs for workflow execution + extract_output: Use LLM to extract the workflow output **kwargs (Any): Additional keyword arguments - + Returns: - str: The output of the workflow execution + WorkflowResult: The result of the workflow execution + """ + return run_coroutine_sync(self.async_execute(inputs, extract_output, **kwargs)) + + async def async_execute(self, inputs: Optional[dict] = None, extract_output: bool = False, **kwargs) -> WorkflowResult: + """ + Asynchronously execute the workflow. + + Args: + inputs: Dictionary of inputs for workflow execution + extract_output: Use LLM to extract the final workflow output + **kwargs (Any): Additional keyword arguments + + Returns: + WorkflowResult: The result of the workflow execution """ - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) try: - return loop.run_until_complete(self.async_execute(inputs, **kwargs)) - finally: - loop.close() + result = await self._execute_workflow(inputs, extract_output, **kwargs) + return WorkflowResult(status="success", result=result) + + except DisplayableException as e: + logger.exception(e) + tb = traceback.format_exc() + return WorkflowResult(status="failed", error_msg=tb, displayable_error=str(e)) - async def async_execute(self, inputs: dict = {}, **kwargs) -> str: + except Exception as e: + logger.exception(e) + tb = traceback.format_exc() + return WorkflowResult( + status="failed", + error_msg=tb, + displayable_error="An unexpected error occurred. Please try again. If the problem persists, please contact support.", + ) + + async def _execute_workflow(self, inputs: Optional[dict] = None, extract_output: bool = False, **kwargs) -> Union[dict, str]: """ Asynchronously execute the workflow. - + Args: inputs: Dictionary of inputs for workflow execution + extract_output: Use LLM to extract the final workflow output **kwargs (Any): Additional keyword arguments - + Returns: - str: The output of the workflow execution + Union[dict, str]: The output of the workflow execution """ goal = self.graph.goal + inputs = dict(inputs or {}) + # Reset node statuses and environment state so reusing the same WorkFlow instance + # does not leak the previous run's trajectory/execution data into this one. This + # runs unconditionally, so it also recovers from a previous failed execution. + self.graph.reset_graph() + self.environment.reset() # inputs.update({"goal": goal}) inputs = self._prepare_inputs(inputs) + self._validate_inputs(inputs) # prepare for hitl functionalities if hasattr(self, "hitl_manager") and (self.hitl_manager is not None): self._prepare_hitl() - # check the inputs and outputs of the task + # check the inputs and outputs of the task self._validate_workflow_structure(inputs=inputs, **kwargs) inp_message = Message(content=inputs, msg_type=MessageType.INPUT, wf_goal=goal) self.environment.update(message=inp_message, state=TrajectoryState.COMPLETED) + task = None - failed = False - error_message = None - while not self.graph.is_complete and not failed: + while not self.graph.is_complete: try: task: WorkFlowNode = await self.get_next_task() if task is None: @@ -92,26 +178,42 @@ async def async_execute(self, inputs: dict = {}, **kwargs) -> str: logger.info(f"Executing subtask: {task.name}") await self.execute_task(task=task) except Exception as e: - failed = True + task_name = getattr(task, "name", "Unknown") error_message = Message( - content=f"An Error occurs when executing the workflow: {e}", - msg_type=MessageType.ERROR, + content=f"An Error occurs when executing task {task_name}: {e}", + msg_type=MessageType.ERROR, wf_goal=goal ) self.environment.update(message=error_message, state=TrajectoryState.FAILED, error=str(e)) - - if failed: - logger.error(error_message.content) - return "Workflow Execution Failed" - + raise + logger.info("Extracting WorkFlow Output ...") - output: str = await self.workflow_manager.extract_output(graph=self.graph, env=self.environment) + + if extract_output: + output: str = await self.workflow_manager.extract_output(graph=self.graph, env=self.environment) + else: + output: dict = self.environment.get_execution_data(self.output_names) + + self.graph.reset_graph() + logger.info("Workflow execution completed successfully") return output - - def _prepare_inputs(self, inputs: dict) -> dict: + + def _validate_inputs(self, inputs: dict): + workflow_inputs = [param.to_dict(ignore=["class_name"]) for param in self.graph.workflow_inputs] + input_validator = CustomizeAgent.create_action_input(workflow_inputs, "workflow_inputs") + try: + input_validator(**inputs) + except ValidationError as e: + error_msg = format_validation_error(e) + raise InputValidationError(f"Invalid inputs: {error_msg}") from e + except Exception: + raise + + def _prepare_inputs(self, inputs: Optional[dict] = None) -> dict: """ Prepare the inputs for the workflow execution. Mainly determine whether the goal should be added to the inputs. """ + inputs = dict(inputs or {}) initial_node_names = self.graph.find_initial_nodes() initial_node_required_inputs = set() for initial_node_name in initial_node_names: @@ -123,12 +225,15 @@ def _prepare_inputs(self, inputs: dict) -> dict: return inputs - async def get_next_task(self) -> WorkFlowNode: + async def get_next_task(self) -> Optional[WorkFlowNode]: task_execution_history = " -> ".join(self.environment.task_execution_history) if not task_execution_history: task_execution_history = "None" logger.info(f"Task Execution Trajectory: {task_execution_history}. Scheduling next subtask ...") - task: WorkFlowNode = await self.workflow_manager.schedule_next_task(graph=self.graph, env=self.environment) + task: Optional[WorkFlowNode] = await self.workflow_manager.schedule_next_task(graph=self.graph, env=self.environment) + if task is None: + logger.info("No next subtask could be scheduled (the scheduler returned None).") + return None logger.info(f"The next subtask to be executed is: {task.name}") return task @@ -162,13 +267,14 @@ async def _async_execute_task_by_action_graph(self, task: WorkFlowNode, next_act next_action: The next action to perform with its action graph """ action_graph: ActionGraph = next_action.action_graph - async_execute_source = inspect.getsource(action_graph.async_execute) - if "NotImplementedError" in async_execute_source: + if is_method_overridden(action_graph, ActionGraph, "async_execute"): + execute_function = action_graph.async_execute + elif is_method_overridden(action_graph, ActionGraph, "execute"): execute_function = action_graph.execute - async_execute = False else: - execute_function = action_graph.async_execute - async_execute = True + raise NotImplementedError( + f"The action graph '{type(action_graph).__name__}' must implement `execute` or `async_execute`." + ) # execute_signature = inspect.signature(type(action_graph).async_execute) execute_signature = inspect.signature(execute_function) execute_params = {} @@ -186,10 +292,7 @@ async def _async_execute_task_by_action_graph(self, task: WorkFlowNode, next_act # action_input_data = self.environment.get_all_execution_data() # execute_inputs = {param: action_input_data.get(param, "") for param in execute_params} # action_graph_output: dict = await action_graph.async_execute(**execute_inputs) - if async_execute: - action_graph_output: dict = await action_graph.async_execute(**execute_params) - else: - action_graph_output: dict = action_graph.execute(**execute_params) + action_graph_output: dict = await call_maybe_async(execute_function, **execute_params) message = Message( content=action_graph_output, action=action_graph.name, msg_type=MessageType.RESPONSE, @@ -327,6 +430,8 @@ def _validate_workflow_structure(self, inputs: dict, **kwargs): ) for node in self.graph.nodes: + if not node.agents: + continue for agent in node.agents: if hasattr(agent, "forbidden_in_workflow") and (agent.forbidden_in_workflow): raise ValueError(f"The Agent of class {agent.__class__} is forbidden to be used in the workflow.") @@ -432,4 +537,4 @@ def _prepare_hitl(self): for agent, node in zip(hitl_agents, node_with_hitl_agents): self._prepare_single_hitl_agent(agent, node) - return \ No newline at end of file + return diff --git a/evoagentx/workflow/workflow_graph.py b/evoagentx/workflow/workflow_graph.py index ce2459b6..62ebfd37 100644 --- a/evoagentx/workflow/workflow_graph.py +++ b/evoagentx/workflow/workflow_graph.py @@ -1,24 +1,33 @@ -import json import inspect +import json import threading -from enum import Enum -import networkx as nx +from collections import defaultdict from copy import deepcopy +from enum import Enum +from functools import wraps +from typing import Dict, List, Literal, Optional, Tuple, Union + +import networkx as nx from networkx import MultiDiGraph -from collections import defaultdict from pydantic import Field, field_validator, model_validator -from typing import Union, Optional, Tuple, Callable, Dict, List -from functools import wraps +from ..agents import Agent, CustomizeAgent +from ..core.base_config import Parameter from ..core.logging import logger from ..core.module import BaseModule -from ..core.base_config import Parameter -from .action_graph import ActionGraph -from ..agents.agent import Agent -from ..utils.utils import generate_dynamic_class_name, make_parent_folder -from ..prompts.workflow.sew_workflow import SEW_WORKFLOW +from ..core.module_utils import recursive_to_dict from ..prompts.utils import DEFAULT_SYSTEM_PROMPT -# from ..tools.tool import Toolkit, Tool +from ..prompts.workflow.sew_workflow import SEW_WORKFLOW +from ..utils.async_utils import is_method_overridden +from ..utils.utils import ( + generate_dynamic_class_name, + make_parent_folder, + pydantic_to_parameters, + recursive_remove, + remove_none, + validate_param, +) +from .action_graph import ActionGraph class WorkFlowNodeState(str, Enum): @@ -58,16 +67,44 @@ class WorkFlowNode(BaseModule): name: str # A short name of the task. Should be unique in a single workflow description: str # A detailed description of the task - inputs: List[Parameter] # inputs for the task - outputs: List[Parameter] # outputs of the task + inputs: List[Parameter] = Field(default_factory=list) # may be empty for nodes with no external input + outputs: List[Parameter] = Field(default_factory=list) reason: Optional[str] = None - agents: Optional[List[Union[str, dict]]] = None + agents: Optional[List] = None action_graph: Optional[ActionGraph] = None status: Optional[WorkFlowNodeState] = WorkFlowNodeState.PENDING + @staticmethod + def _warn_agent_dict_class_name(agent: dict, node_name: Optional[str] = None) -> None: + if not isinstance(agent, dict) or "class_name" not in agent: + return + + agent_name = agent.get("name", "") + node_context = f" in node '{node_name}'" if node_name else "" + logger.warning( + "Agent dict '{}'{} contains class_name='{}'. BaseModule will convert " + "this dict into an Agent instance during construction/loading. If you " + "intended to keep it as a plain agent config dict for AgentManager, " + "remove or rename the top-level 'class_name' key.", + agent_name, + node_context, + agent.get("class_name"), + ) + + @classmethod + def _warn_agent_dicts_with_class_name(cls, agents, node_name: Optional[str] = None) -> None: + if not agents: + return + for agent in agents: + cls._warn_agent_dict_class_name(agent, node_name=node_name) + + def __init__(self, **kwargs): + self._warn_agent_dicts_with_class_name(kwargs.get("agents"), node_name=kwargs.get("name")) + super().__init__(**kwargs) + @field_validator('agents', mode="before") @classmethod - def check_agent_format(cls, agents: List[Union[str, dict, Agent]]): + def check_agent_format(cls, agents: List[Union[str, Dict, Agent]]): if agents is None: return None @@ -76,40 +113,42 @@ def check_agent_format(cls, agents: List[Union[str, dict, Agent]]): if isinstance(agent, str): validated_agents.append(agent) elif isinstance(agent, Agent): - validated_agents.append(agent.get_config()) + validated_agents.append(agent) elif isinstance(agent, dict): + cls._warn_agent_dict_class_name(agent) assert "name" in agent and "description" in agent, \ "must provide the name and description of an agent when specifying an agent with a dict." validated_agents.append(agent) + else: + raise TypeError(f"'{type(agent)}' is an unknown agent type!") return validated_agents @model_validator(mode="after") - def check_action_graph(self) -> "WorkFlowNode": + def check_action_graph(self): """ Validates that: 1. All required parameters of execute/async_execute methods are included in inputs 2. The execute/async_execute methods return dictionaries 3. All output parameters are present in the returned dictionaries """ - - # alias to minimise diff from the pre-Pydantic-2.12 version - instance = self - - if instance.action_graph is None: - return instance + if self.action_graph is None: + return self # Get input parameter names from the node's input parameters - input_param_names = {param.name for param in instance.inputs if param.required} - output_param_names = {param.name for param in instance.outputs if param.required} + input_param_names = {param.name for param in self.inputs if param.required} + output_param_names = {param.name for param in self.outputs if param.required} - def check_method_signature(method, method_name): + implemented_methods = [] + + def check_method_signature(method_name): """Helper function to check method signature against input parameters""" - method_source = inspect.getsource(method) - if "NotImplementedError" in method_source: + if not is_method_overridden(self.action_graph, ActionGraph, method_name): return - + implemented_methods.append(method_name) + # Get method signature + method = getattr(self.action_graph, method_name) method_sig = inspect.signature(method) # Only consider parameters other than self, *args, and **kwargs as required @@ -125,14 +164,16 @@ def check_method_signature(method, method_name): raise ValueError(f"`{method_name}` method requires parameters that are not in `inputs`: {missing_inputs}") # Check execute method - check_method_signature(instance.action_graph.execute, "execute") + check_method_signature("execute") # Check async_execute method if it exists - check_method_signature(instance.action_graph.async_execute, "async_execute") + check_method_signature("async_execute") + + if not implemented_methods: + raise ValueError( + f"ActionGraph '{type(self.action_graph).__name__}' must implement `execute` or `async_execute`." + ) # Monkey-patch execute and async_execute to check returns at runtime - original_execute = instance.action_graph.execute - original_async_execute = instance.action_graph.async_execute - def check_method_return(method_name, result): if not isinstance(result, dict): raise TypeError(f"{method_name} must return a dictionary, got {type(result)}") @@ -143,30 +184,61 @@ def check_method_return(method_name, result): raise ValueError(f"{method_name} return value is missing required outputs: {missing_outputs}") return result - - @wraps(original_execute) - def patched_execute(*args, **kwargs): - result = original_execute(*args, **kwargs) - return check_method_return("execute", result) - - @wraps(original_async_execute) - async def patched_async_execute(*args, **kwargs): - result = await original_async_execute(*args, **kwargs) - return check_method_return("async_execute", result) - + # Replace the methods with our patched versions - instance.action_graph.execute = patched_execute - instance.action_graph.async_execute = patched_async_execute + if "execute" in implemented_methods: + original_execute = self.action_graph.execute + + @wraps(original_execute) + def patched_execute(*args, **kwargs): + result = original_execute(*args, **kwargs) + return check_method_return("execute", result) + + self.action_graph.execute = patched_execute + + if "async_execute" in implemented_methods: + original_async_execute = self.action_graph.async_execute + + @wraps(original_async_execute) + async def patched_async_execute(*args, **kwargs): + result = original_async_execute(*args, **kwargs) + if inspect.isawaitable(result): + result = await result + return check_method_return("async_execute", result) + + self.action_graph.async_execute = patched_async_execute - return instance + return self def to_dict(self, exclude_none: bool = True, ignore: List[str] = [], **kwargs) -> dict: + + agents_dict: List[Union[str, dict]] = [] + if self.agents: + for agent in self.agents: + if isinstance(agent, str): + agents_dict.append(agent) + elif isinstance(agent, Agent): + agents_dict.append(agent.get_config()) + elif isinstance(agent, dict): + agent_dict = recursive_to_dict(agent) + # for CustomizeAgent: a callable parse_func is not serializable, store its name + if "parse_func" in agent_dict and callable(agent_dict["parse_func"]): + agent_dict["parse_func"] = agent_dict["parse_func"].__name__ + agents_dict.append(agent_dict) + else: + raise TypeError(f"'{type(agent)}' is an unknown agent type!") + + if exclude_none: + agents_dict = remove_none(agents_dict) + + if ignore: + agents_dict = recursive_remove(agents_dict, ignore) + + if "agents" not in ignore: + ignore = [*ignore, "agents"] data = super().to_dict(exclude_none=exclude_none, ignore=ignore, **kwargs) - for agent in data.get("agents", []): - # for CustomizeAgent - if isinstance(agent, dict) and "parse_func" in agent and isinstance(agent["parse_func"], Callable): - agent["parse_func"] = agent["parse_func"].__name__ + data["agents"] = agents_dict return data def get_agents(self) -> List[str]: @@ -182,11 +254,13 @@ def get_agents(self) -> List[str]: agent_names.append(agent) elif isinstance(agent, dict): agent_names.append(agent["name"]) + elif isinstance(agent, Agent): + agent_names.append(agent.name) else: raise TypeError(f"{type(agent)} is an unknown agent type!") return agent_names - def set_agents(self, agents: List[Union[str, dict]]): + def set_agents(self, agents: List[Union[str, Dict, Agent]]): self.agents = agents def get_status(self) -> WorkFlowNodeState: @@ -226,6 +300,227 @@ def get_output_names(self, required: bool = False) -> List[str]: else: return [param.name for param in self.outputs] + def check_agents(self): + """ + Checks if any agent assigned to this node accept the node inputs and if any agent outputs the node outputs. + """ + if not self.agents: + # A node may be executed either by agents or by an action_graph. When an + # action_graph is provided, the node is satisfied without any agents, and + # its inputs/outputs are validated against the action_graph elsewhere. + if self.action_graph is not None: + return + raise ValueError(f"No agents assigned to node '{self.name}'") + + node_inputs_dict = {} + inputs_in_agents = {} + for node_input in self.inputs: + node_inputs_dict[node_input.name] = node_input + inputs_in_agents[node_input.name] = False + + node_outputs_dict = {} + outputs_in_agents = {} + for node_output in self.outputs: + node_outputs_dict[node_output.name] = node_output + outputs_in_agents[node_output.name] = False + + in_agents = {"inputs": inputs_in_agents, "outputs": outputs_in_agents} + node_inputs_outputs = {"inputs": node_inputs_dict, "outputs": node_outputs_dict} + + + def _check_agent_dict(agent_dict: dict, inputs_or_outputs: Literal["inputs", "outputs"]) -> dict: + + # "input" or "output" + input_or_output = inputs_or_outputs[:-1] + + for agent_input_or_output in agent_dict[inputs_or_outputs]: + if agent_input_or_output["name"] in node_inputs_outputs[inputs_or_outputs]: + in_agents[inputs_or_outputs][agent_input_or_output["name"]] = True + agent_input_or_output_param = Parameter(**agent_input_or_output) + validate_param( + node_inputs_outputs[inputs_or_outputs][agent_input_or_output["name"]], + agent_input_or_output_param, + f"node '{self.name}' {input_or_output}", + f"agent '{agent_dict['name']}' {input_or_output}", + ) + + return agent_dict + + + def _check_agent(agent: Agent, inputs_or_outputs: Literal["inputs", "outputs"]) -> Agent: + + # "input" or "output" + input_or_output = inputs_or_outputs[:-1] + + for i, agent_actions in enumerate(agent.actions): + if agent_actions.name == "ContextExtraction": + continue + + action_format_name = "inputs_format" if inputs_or_outputs == "inputs" else "outputs_format" + action_format = getattr(agent_actions, action_format_name) + + ignore = ["class_name"] + + if not action_format._is_content_defined_in_subclass(): + ignore.append("content") + + action_params = pydantic_to_parameters(action_format, ignore=ignore) + + for param in action_params: + if param.name in node_inputs_outputs[inputs_or_outputs]: + in_agents[inputs_or_outputs][param.name] = True + validate_param( + node_inputs_outputs[inputs_or_outputs][param.name], + param, + f"node '{self.name}' {input_or_output}", + f"agent action '{agent_actions.name}' {input_or_output}", + ) + + action_params = [param.to_dict(ignore=["class_name"]) for param in action_params] + if inputs_or_outputs == "inputs": + required_action_format = CustomizeAgent.create_action_input(action_params, agent_actions.name) + else: + required_action_format = CustomizeAgent.create_action_output(action_params, agent_actions.name) + setattr(agent.actions[i], action_format_name, required_action_format) + return agent + + + def _check_customize_agent(agent: CustomizeAgent, inputs_or_outputs: Literal["inputs", "outputs"]) -> CustomizeAgent: + # input or output + input_or_output = inputs_or_outputs[:-1] + + for agent_input_or_output in getattr(agent, inputs_or_outputs): + param_name = agent_input_or_output.name + if param_name in node_inputs_outputs[inputs_or_outputs]: + in_agents[inputs_or_outputs][param_name] = True + validate_param( + node_inputs_outputs[inputs_or_outputs][param_name], + agent_input_or_output, + f"node '{self.name}' {input_or_output}", + f"agent '{agent.name}' {input_or_output}", + ) + + return agent + + + # String agents are references to agents registered elsewhere (e.g. in an + # AgentManager) and are not resolved here, so their inputs/outputs cannot be + # inspected for structural validation. + has_unresolved_agents = any(isinstance(agent, str) for agent in self.agents) + + for i, agent in enumerate(self.agents): + if isinstance(agent, str): + continue + elif isinstance(agent, Agent): + if isinstance(agent, CustomizeAgent): + self.agents[i] = _check_customize_agent(agent, "inputs") + self.agents[i] = _check_customize_agent(agent, "outputs") + else: + self.agents[i] = _check_agent(agent, "inputs") + self.agents[i] = _check_agent(agent, "outputs") + elif isinstance(agent, dict): + self.agents[i] = _check_agent_dict(agent, "inputs") + self.agents[i] = _check_agent_dict(agent, "outputs") + else: + raise TypeError(f"{type(agent)} is an unknown agent type!") + + # When the node has unresolved (string) agent references, their inputs/outputs + # are unknown here, so coverage of the node's inputs/outputs cannot be verified. + if has_unresolved_agents: + return + + if not all(in_agents["inputs"].values()): + missing_inputs = [input_name for input_name, in_agent in in_agents["inputs"].items() if not in_agent] + raise ValueError(f"Not all inputs of node '{self.name}' are used by agents: {missing_inputs}") + if not all(in_agents["outputs"].values()): + missing_outputs = [output_name for output_name, in_agent in in_agents["outputs"].items() if not in_agent] + raise ValueError(f"Not all outputs of node '{self.name}' can be found in agents: {missing_outputs}") + + def update_inputs(self, inputs: List[Parameter]): + self.inputs = inputs + self.update_agent_inputs() + + def update_outputs(self, outputs: List[Parameter]): + self.outputs = outputs + self.update_agent_outputs() + + def update_agent_inputs(self): + """Update all agent inputs to match the node's inputs""" + self._update_agent_params("inputs") + + def update_agent_outputs(self): + """Update all agent outputs to match the node's outputs""" + self._update_agent_params("outputs") + + @staticmethod + def _get_agent_param_names(agent, param_type: Literal["inputs", "outputs"]) -> set: + if isinstance(agent, Agent) and hasattr(agent, param_type): + return {p.name for p in getattr(agent, param_type)} + elif isinstance(agent, dict): + return {p["name"] for p in agent.get(param_type, [])} + return set() + + def _update_agent_params(self, param_type: Literal["inputs", "outputs"]): + if not self.agents: + logger.warning(f"Node '{self.name}' has no agents; {param_type} not updated.") + return + + node_params_map = { + param.name: param + for param in getattr(self, param_type) + } + agents = self.agents + complement_type = "outputs" if param_type == "inputs" else "inputs" + + for i, agent in enumerate(agents): + if len(agents) == 1: + # Single-agent: agent must exactly mirror the node's params + if isinstance(agent, Agent) and hasattr(agent, param_type): + setattr(self.agents[i], param_type, list(node_params_map.values())) + elif isinstance(agent, dict): + self.agents[i][param_type] = [p.to_dict(ignore=["class_name"]) for p in node_params_map.values()] + continue + + # Multi-agent: collect inter-agent param names from other agents' + # complementary params (e.g. other agents' outputs when syncing inputs) + inter_agent_names = set() + for j, other in enumerate(agents): + if j != i: + inter_agent_names |= self._get_agent_param_names(other, complement_type) + + if isinstance(agent, Agent) and hasattr(agent, param_type): + new_params = [] + for param in getattr(agent, param_type): + name = param.name + if name in node_params_map: + new_params.append(node_params_map[name]) + elif name in inter_agent_names: + new_params.append(param) + # else: stale node param — drop it + setattr(self.agents[i], param_type, new_params) + + elif isinstance(agent, dict): + new_params = [] + for param in agent.get(param_type, []): + name = param["name"] + if name in node_params_map: + new_params.append(node_params_map[name].to_dict(ignore=["class_name"])) + elif name in inter_agent_names: + new_params.append(param) + self.agents[i][param_type] = new_params + + if len(agents) > 1: + covered = set() + for agent in agents: + covered |= self._get_agent_param_names(agent, param_type) + missing = set(node_params_map.keys()) - covered + if missing: + logger.warning( + f"Node '{self.name}': the following {param_type} are not handled by any agent " + f"and will not be propagated: {sorted(missing)}" + ) + + class WorkFlowEdge(BaseModule): """ Represents a directed edge in a workflow graph. @@ -303,66 +598,258 @@ class WorkFlowGraph(BaseModule): nodes: List of WorkFlowNode instances representing tasks edges: List of WorkFlowEdge instances representing dependencies graph: Internal NetworkX MultiDiGraph or another WorkFlowGraph + workflow_inputs: List of inputs that the workflow accepts. If not provided, inputs from initial nodes are used. + workflow_outputs: The final outputs of the workflow. If not provided, outputs from end nodes are used. """ goal: str nodes: Optional[List[WorkFlowNode]] = [] edges: Optional[List[WorkFlowEdge]] = [] graph: Optional[Union[MultiDiGraph, "WorkFlowGraph"]] = Field(default=None, exclude=True) + workflow_inputs: Optional[List[Parameter]] = None + workflow_outputs: Optional[List[Parameter]] = None def init_module(self): self._lock = threading.Lock() if not self.graph: - self._init_from_nodes_and_edges(self.nodes, self.edges) + self._init_from_nodes(self.nodes, explicit_edges=self.edges) elif isinstance(self.graph, MultiDiGraph): - self._init_from_multidigraph(self.graph, self.nodes, self.edges) + self._init_from_multidigraph(self.graph, self.nodes) elif isinstance(self.graph, WorkFlowGraph): - self._init_from_workflowgraph(self.graph, self.nodes, self.edges) + self._init_from_workflowgraph(self.graph, self.nodes) else: raise TypeError(f"{type(self.graph)} is an unknown type for graph. Supported types: [MultiDiGraph, WorkFlowGraph]") + + def _dedup_params(params: List[Parameter]) -> List[Parameter]: + # Multiple initial/end nodes may share a parameter name (e.g. a common workflow + # input). Keep the first occurrence so the derived list passes the uniqueness check. + seen = set() + deduped = [] + for param in params: + if param.name in seen: + continue + seen.add(param.name) + deduped.append(param) + return deduped + + # If `workflow_inputs` is not provided, set it to the inputs of initial nodes + if self.workflow_inputs is None: + initial_nodes = [node for node, in_degree in self.graph.in_degree() if in_degree==0] + workflow_inputs = [] + for node_name in initial_nodes: + workflow_inputs.extend(self.get_node(node_name).inputs) + self.workflow_inputs = _dedup_params(workflow_inputs) + + # If `workflow_outputs` is not provided, set it to the outputs of end nodes + if self.workflow_outputs is None: + end_nodes = [node for node, out_degree in self.graph.out_degree() if out_degree==0] + workflow_outputs = [] + for node_name in end_nodes: + workflow_outputs.extend(self.get_node(node_name).outputs) + self.workflow_outputs = _dedup_params(workflow_outputs) + + self.workflow_inputs_dict = {param.name: param for param in self.workflow_inputs} + self.workflow_outputs_dict = {param.name: param for param in self.workflow_outputs} + self._validate_workflow_structure() + self._check_workflow_inputs_outputs() + # NOTE: skip _check_agents during initialization because in workflow generator + # the graph needs to be constructed before agents are generated self.update_graph() def update_graph(self): # call this function when modifying nodes or edges! self._loops = self._find_all_loops() - def _init_from_nodes_and_edges(self, nodes: List[WorkFlowNode] = [], edges: List[WorkFlowEdge] = []): + def _infer_edges_from_nodes(self, nodes: List[WorkFlowNode]) -> List[WorkFlowEdge]: + """ + Infer edges from the provided nodes by checking whether any output of a node + matches an input of another node. An edge (source -> target) is created whenever + at least one output name of `source` appears as an input name of `target`. + """ + edges = [] + node_inputs = {node.name: set(param.name for param in node.inputs) for node in nodes} + node_outputs = {node.name: set(param.name for param in node.outputs) for node in nodes} + for source in nodes: + for target in nodes: + if source.name == target.name: + continue + if node_inputs[target.name] & node_outputs[source.name]: + edge = WorkFlowEdge(source=source.name, target=target.name) + if edge not in edges: + edges.append(edge) + return edges + + def _init_from_nodes(self, nodes: List[WorkFlowNode] = [], explicit_edges: Optional[List[WorkFlowEdge]] = None): """ - Initialize the WorkFlowGraph from a set of nodes and edges. + Initialize the WorkFlowGraph from a set of nodes. Data-flow edges are inferred + automatically: an edge from node A to node B is created whenever an output of A shares + a name with an input of B. + + Any user-provided `explicit_edges` are merged *in addition to* the inferred edges + (deduplicated by `(source, target)`). If an explicit edge has the same `(source, target)` + as an inferred edge, the explicit edge replaces the inferred edge's metadata (e.g. + `priority`) while preserving the inferred data-flow topology. This lets users express + ordering dependencies that carry no shared data, while wrong/incomplete explicit edges can + no longer silently break the graph. Explicit edges referencing unknown nodes raise; + explicit edges with no matching input/output are kept but emit a warning (see `add_edge`). """ - - if edges and not nodes: - raise ValueError("edges cannot be passed without nodes or a graph") - self.nodes = [] self.edges = [] self.graph = MultiDiGraph() self.add_nodes(*nodes, update_graph=False) - self.add_edges(*edges, update_graph=False) + inferred_edges = self._infer_edges_from_nodes(self.nodes) + self.add_edges(*inferred_edges, update_graph=False) + + if explicit_edges: + seen_pairs = {(edge.source, edge.target) for edge in inferred_edges} + extra_edges = [] + for edge in explicit_edges: + pair = (edge.source, edge.target) + if pair in seen_pairs: + self._replace_edge_by_pair(edge) + continue + seen_pairs.add(pair) + extra_edges.append(edge) + self.add_edges(*extra_edges, update_graph=False) - def _init_from_multidigraph(self, graph: MultiDiGraph, nodes: List[WorkFlowNode] = [], edges: List[WorkFlowEdge] = []): + def _replace_edge_by_pair(self, edge: WorkFlowEdge) -> bool: + """ + Replace an existing edge with the same source/target pair, preserving one edge + in both `self.edges` and the underlying NetworkX graph. + """ + if not isinstance(edge, WorkFlowEdge): + raise ValueError(f"{edge} is not a valid WorkFlowEdge instance!") + for i, existing_edge in enumerate(self.edges): + if existing_edge.source != edge.source or existing_edge.target != edge.target: + continue + + self.edges[i] = edge + edge_data = self.graph.get_edge_data(edge.source, edge.target, default={}) + for attrs in edge_data.values(): + ref = attrs.get("ref") + if isinstance(ref, WorkFlowEdge) and ref.source == edge.source and ref.target == edge.target: + attrs["ref"] = edge + return True + return True + return False + + def _init_from_multidigraph(self, graph: MultiDiGraph, nodes: List[WorkFlowNode] = []): graph_nodes = [deepcopy(node_attrs["ref"]) for _, node_attrs in graph.nodes(data=True)] graph_edges = [deepcopy(edge_attrs["ref"]) for *_, edge_attrs in graph.edges(data=True)] graph_nodes = self.merge_nodes(graph_nodes, nodes) - graph_edges = self.merge_edges(graph_edges, edges) - self._init_from_nodes_and_edges(nodes=graph_nodes, edges=graph_edges) - - def _init_from_workflowgraph(self, graph: "WorkFlowGraph", nodes: List[WorkFlowNode] = [], edges: List[WorkFlowEdge] = []): + self._init_from_nodes(nodes=graph_nodes, explicit_edges=graph_edges) + def _init_from_workflowgraph(self, graph: "WorkFlowGraph", nodes: List[WorkFlowNode] = []): graph_nodes = deepcopy(graph.nodes) graph_edges = deepcopy(graph.edges) graph_nodes = self.merge_nodes(graph_nodes, nodes) - graph_edges = self.merge_edges(graph_edges, edges) - self._init_from_nodes_and_edges(nodes=graph_nodes, edges=graph_edges) - - def _validate_workflow_structure(self): + self._init_from_nodes(nodes=graph_nodes, explicit_edges=graph_edges) + def _check_isolated_nodes(self): + """ + If there are isolated nodes, check if their inputs and outputs are workflow inputs and outputs. + If not, raise error as their inputs and outputs are not connected to the workflow graph. + """ isolated_nodes = list(nx.isolates(self.graph)) - if len(self.graph.nodes) > 1 and isolated_nodes: - logger.warning(f"The workflow contains isolated nodes: {isolated_nodes}") + if len(self.graph.nodes) > 1 and len(isolated_nodes) > 0: + + for node_name in isolated_nodes: + node = self.get_node(node_name) + + for node_input in node.inputs: + if node_input.name not in self.workflow_inputs_dict: + error_message = f"Node '{node_name}' is an isolated node and has an input '{node_input.name}' that is not in the workflow inputs: {list(self.workflow_inputs_dict.keys())}" + logger.error(error_message) + raise ValueError(error_message) + + for node_output in node.outputs: + if node_output.name not in self.workflow_outputs_dict: + error_message = f"Node '{node_name}' is an isolated node and has an output '{node_output.name}' that is not in the workflow outputs: {list(self.workflow_outputs_dict.keys())}" + logger.error(error_message) + raise ValueError(error_message) + + @staticmethod + def check_io_duplicates(inputs: List[Parameter], outputs: List[Parameter], context: str) -> None: + """Helper to ensure input/output names are unique and do not overlap.""" + def check_group(params: List[Parameter], group_type: str): + seen = set() + for param in params: + if param.name in seen: + error_message = f"{context} has duplicate {group_type} name: '{param.name}'" + logger.error(error_message) + raise ValueError(error_message) + seen.add(param.name) + return seen + + seen_inputs = check_group(inputs, "input") + seen_outputs = check_group(outputs, "output") + + overlap = seen_inputs & seen_outputs + if overlap: + error_message = f"{context} inputs and outputs share the following name(s), which is not allowed: {list(overlap)}" + logger.error(error_message) + raise ValueError(error_message) + + def _check_workflow_io_duplicates(self): + """Workflow inputs and outputs must be unique and not share any names.""" + WorkFlowGraph.check_io_duplicates(self.workflow_inputs, self.workflow_outputs, "Workflow") + + def _check_node_io_duplicates(self): + """Within a single node, inputs and outputs must be unique and not share any names.""" + for node in self.nodes: + WorkFlowGraph.check_io_duplicates(node.inputs, node.outputs, f"Node '{node.name}'") + + def _check_node_output_uniqueness(self): + """Each output name must appear as a node output in at most one node.""" + output_to_nodes: Dict[str, List[str]] = defaultdict(list) + + for node in self.nodes: + for param in node.outputs: + output_to_nodes[param.name].append(node.name) + + conflicts = { + name: nodes + for name, nodes in output_to_nodes.items() + if len(nodes) > 1 + } + + if conflicts: + details = "\n".join( + f"'{name}' produced by {nodes}" for name, nodes in conflicts.items() + ) + error_message = f"Each node output name must be unique across all nodes. Found conflicts:\n{details}" + logger.error(error_message) + raise ValueError(error_message) + + def _check_node_inputs(self): + """Validate that every node input is sourced from workflow inputs or another node's output.""" + + workflow_input_names = set(self.workflow_inputs_dict.keys()) + all_node_outputs = set() + + for node in self.nodes: + for param in node.outputs: + all_node_outputs.add(param.name) + + for node in self.nodes: + node_outputs = {param.name for param in node.outputs} + valid_sources = workflow_input_names.union(all_node_outputs - node_outputs) + + for param in node.inputs: + if param.name not in valid_sources: + error_message = f"Node '{node.name}' input '{param.name}' is not a workflow input or from another node's output." + logger.error(error_message) + raise ValueError(error_message) + + def _validate_workflow_structure(self): + self._check_workflow_io_duplicates() + self._check_node_io_duplicates() + self._check_node_output_uniqueness() + self._check_node_inputs() + self._check_isolated_nodes() initial_nodes = self.find_initial_nodes() if len(self.graph.nodes) > 1 and not initial_nodes: @@ -375,12 +862,34 @@ def _validate_workflow_structure(self): logger.warning("There are no end nodes in the workflow") def find_initial_nodes(self) -> List[str]: - initial_nodes = [node for node, in_degree in self.graph.in_degree() if in_degree==0] + # initial_nodes = [node for node, in_degree in self.graph.in_degree() if in_degree==0] + + # initial nodes are nodes that only require workflow inputs + workflow_input_names = set(self.workflow_inputs_dict.keys()) + initial_nodes = [] + + for node in self.nodes: + required_inputs = set(node.get_input_names(required=True)) + + if required_inputs.issubset(workflow_input_names): + initial_nodes.append(node.name) + return initial_nodes - + def find_end_nodes(self) -> List[str]: - end_nodes = [node for node, out_degree in self.graph.out_degree() if out_degree==0] - return end_nodes + # end_nodes = [node for node, out_degree in self.graph.out_degree() if out_degree==0] + + # end nodes are nodes that produce workflow outputs + workflow_output_names = set(self.workflow_outputs_dict.keys()) + end_node_names = [] + + for node in self.nodes: + node_output_names = set(node.get_output_names()) + + if not node_output_names.isdisjoint(workflow_output_names): + end_node_names.append(node.name) + + return end_node_names def _find_loops(self, start_node: Union[str, WorkFlowNode]) -> Dict[str, list]: @@ -637,14 +1146,9 @@ def get_node_status(self, node: Union[str, WorkFlowNode]) -> WorkFlowNodeState: @property def is_complete(self): - # node_complete_list = [node.is_complete for node in self.nodes] - leaf_nodes = [self.get_node(name) for name in self.find_end_nodes()] - node_complete_list = [node.is_complete for node in leaf_nodes] - if len(node_complete_list) == 0: - return True - if all(node_complete_list): - return True - return False + output_node_names = self.find_end_nodes() + target_nodes = [self.get_node(name) for name in output_node_names] if output_node_names else self.nodes + return all(node.is_complete for node in target_nodes) def reset_graph(self): """ @@ -865,11 +1369,17 @@ def filter_nodes_with_uncompleted_predecessors(self, nodes: List[Union[str, Work def get_next_candidate_nodes(self) -> List[str]: + # `find_initial_nodes` classifies a node as initial purely from data readiness + # (its required inputs are a subset of the workflow inputs). A node can satisfy that + # while still having incoming edges — an explicit control edge, or an inferred edge + # feeding one of its optional inputs. Those nodes must not start before their + # predecessors, so filter the initial candidates by predecessor completion as well. uncomplete_initial_nodes = self.get_uncomplete_initial_nodes() - if len(uncomplete_initial_nodes) > 0: - return uncomplete_initial_nodes - - # find the last completed nodes in all paths starting from initial nodes. + ready_initial_nodes = self.filter_nodes_with_uncompleted_predecessors(uncomplete_initial_nodes) + if len(ready_initial_nodes) > 0: + return ready_initial_nodes + + # find the last completed nodes in all paths starting from initial nodes. completed_leaf_nodes = self.find_completed_leaf_nodes_start_from_initial_nodes() # obtain children nodes of last completed nodes which are uncompleted (consider previous completed tasks if there exists loops) @@ -1048,22 +1558,67 @@ def format_parameters(params: List[Parameter]) -> str: subtask_texts.append(text) workflow_desc = "\n\n".join(subtask_texts) return workflow_desc - - def _infer_edges_from_nodes(self, nodes: List[WorkFlowNode]) -> List[WorkFlowEdge]: - if not nodes: - return [] - edges: List[WorkFlowEdge] = [] - for node in nodes: - for another_node in nodes: - if node.name == another_node.name: - continue - node_output_params = [param.name for param in node.outputs] - another_node_input_params = [param.name for param in another_node.inputs] - if any([param in another_node_input_params for param in node_output_params]): - edges.append(WorkFlowEdge(edge_tuple=(node.name, another_node.name))) - return edges - + + def _check_workflow_inputs_outputs(self): + """Checks if the workflow inputs and outputs can be satisfied by the nodes in the workflow graph. + Raises error If any workflow input is not used by a node, or if any workflow output cannot be found from a node's output. + """ + workflow_inputs_outputs = {"inputs": self.workflow_inputs_dict, "outputs": self.workflow_outputs_dict} + workflow_input_in_nodes = {workflow_input_name: False for workflow_input_name in self.workflow_inputs_dict} + workflow_output_in_nodes = {workflow_output_name: False for workflow_output_name in self.workflow_outputs_dict} + in_nodes = {"inputs": workflow_input_in_nodes, "outputs": workflow_output_in_nodes} + + def _check_node(node: WorkFlowNode, inputs_or_outputs: Literal["inputs", "outputs"]) -> WorkFlowNode: + + # "input" or "output" + input_or_output = inputs_or_outputs[:-1] + node_inputs_or_outputs = getattr(node, inputs_or_outputs) + + for node_input_or_output in node_inputs_or_outputs: + if node_input_or_output.name in in_nodes[inputs_or_outputs]: + in_nodes[inputs_or_outputs][node_input_or_output.name] = True + validate_param( + workflow_inputs_outputs[inputs_or_outputs][node_input_or_output.name], + node_input_or_output, + f"workflow {input_or_output}", + f"node '{node.name}' {input_or_output}", + ) + + return node + + + for i, node in enumerate(self.nodes): + self.nodes[i] = _check_node(node, "inputs") + self.nodes[i] = _check_node(node, "outputs") + + if not all(in_nodes["inputs"].values()): + missing_inputs = [input_name for input_name, in_node in in_nodes["inputs"].items() if not in_node] + raise ValueError(f"Not all workflow inputs are used by nodes: {missing_inputs}") + + if not all(in_nodes["outputs"].values()): + missing_outputs = [output_name for output_name, in_node in in_nodes["outputs"].items() if not in_node] + raise ValueError(f"Not all workflow outputs are found in nodes: {missing_outputs}") + + + def _check_agents(self): + """Checks if each node's inputs and outputs can be satisfied by the agents assigned to the node.""" + for node in self.nodes: + node.check_agents() + + + def validate_workflow_graph(self): + """ + Validates the workflow graph by checking: + - The workflow structure + - Workflow inputs and outputs can be derived from nodes + - Nodes' inputs and outputs can be derived from agents + """ + self._validate_workflow_structure() + self._check_workflow_inputs_outputs() + self._check_agents() + + def get_config(self) -> dict: """ Get a dictionary containing all necessary configuration to recreate this workflow graph. @@ -1077,6 +1632,33 @@ def get_config(self) -> dict: return config + @classmethod + def from_dict(cls, data: Dict, **kwargs) -> 'WorkFlowGraph': + """ + Create a WorkFlowGraph instance from a dictionary. + + Agent entries follow the same BaseModule revival rules as the rest of the + workflow data: string agents stay as references, dict agents without a + top-level `class_name` stay as config dicts, and dict agents with a top-level + `class_name` are converted into live `Agent` instances during construction. + Use a top-level `class_name` only when that eager revival is intended; otherwise + leave it out so `AgentManager.add_agents_from_workflow` can materialize the + agent later with the desired `llm_config`/`tools`. + + Args: + data (Dict): The serialized workflow graph. + **kwargs: Additional keyword arguments forwarded to `BaseModule.from_dict`. + """ + # Delegate the actual data loading (class_name dispatch, nested _process_data, + # construction) to BaseModule. + workflow_graph: WorkFlowGraph = super().from_dict(data, **kwargs) + + # Validate structure and node/agent parameter compatibility at load time. This is + # the one behavior BaseModule.from_dict does not provide. + workflow_graph.validate_workflow_graph() + return workflow_graph + + class SequentialWorkFlowGraph(WorkFlowGraph): """ @@ -1096,8 +1678,8 @@ class SequentialWorkFlowGraph(WorkFlowGraph): "output_parser" (optional): Type[ActionOutput], "parse_mode" (optional): str, default is "str" "parse_func" (optional): Callable, - "parse_title" (optional): str , - "tool_names" (optional): List[str] + "title_format" (optional): str , + "tool_names" (optional): List[str] } """ @@ -1131,7 +1713,9 @@ def _infer_node_from_task(self, task: dict) -> WorkFlowNode: agent_output_parser = task.get("output_parser", None) agent_parse_mode = task.get("parse_mode", "str") agent_parse_func = task.get("parse_func", None) - agent_parse_title = task.get("parse_title", None) + # `parse_title` is the legacy key name; `title_format` is what CustomizeAgent expects. + # Accept both, preferring `title_format`, so old saved configs keep working. + agent_title_format = task.get("title_format", task.get("parse_title", None)) tool_names = task.get("tool_names", None) # tools = task.get("tools", []) # tool_names = [] @@ -1162,7 +1746,7 @@ def _infer_node_from_task(self, task: dict) -> WorkFlowNode: "output_parser": agent_output_parser, "parse_mode": agent_parse_mode, "parse_func": agent_parse_func, - "parse_title": agent_parse_title, + "title_format": agent_title_format, "tool_names": tool_names } ], @@ -1188,7 +1772,7 @@ def get_graph_info(self, **kwargs) -> dict: "system_prompt": node.agents[0].get("system_prompt", None), "parse_mode": node.agents[0].get("parse_mode", "str"), "parse_func": node.agents[0].get("parse_func", None).__name__ if node.agents[0].get("parse_func", None) else None, - "parse_title": node.agents[0].get("parse_title", None), + "title_format": node.agents[0].get("title_format", None), "tool_names": node.agents[0].get("tool_names", None) } for node in self.nodes @@ -1218,6 +1802,10 @@ def get_config(self) -> Dict: with the same properties as this one. """ return self.get_graph_info() + + @classmethod + def from_dict(cls, data: Dict, **kwargs) -> 'SequentialWorkFlowGraph': + return cls._create_instance(data) class SEWWorkFlowGraph(SequentialWorkFlowGraph): @@ -1225,4 +1813,4 @@ class SEWWorkFlowGraph(SequentialWorkFlowGraph): def __init__(self, **kwargs): goal = kwargs.pop("goal", SEW_WORKFLOW["goal"]) tasks = kwargs.pop("tasks", SEW_WORKFLOW["tasks"]) - super().__init__(goal=goal, tasks=tasks, **kwargs) \ No newline at end of file + super().__init__(goal=goal, tasks=tasks, **kwargs) diff --git a/evoagentx/workflow/workflow_manager.py b/evoagentx/workflow/workflow_manager.py index 193c516d..e4421a89 100644 --- a/evoagentx/workflow/workflow_manager.py +++ b/evoagentx/workflow/workflow_manager.py @@ -16,7 +16,7 @@ from ..prompts.workflow.workflow_manager import ( DEFAULT_TASK_SCHEDULER, DEFAULT_ACTION_SCHEDULER, - OUTPUT_EXTRACTION_PROMPT + WORKFLOW_OUTPUT_EXTRACTION_PROMPT ) @@ -320,7 +320,7 @@ def _prepare_action_execution( # prepare task and execution information task_info = task.get_task_info() - task_input_names = [param.name for param in task.inputs] + task_input_names = {param.name: param.required for param in task.inputs} task_input_data: dict = env.get_execution_data(task_input_names) task_input_data_info = self.format_task_input_data(data=task_input_data) task_execution_history = "\n\n".join([str(msg) for msg in task_execution_messages]) @@ -480,7 +480,7 @@ async def extract_output(self, graph: WorkFlowGraph, env: Environment, **kwargs) candidate_msgs_with_output.extend(env.get_task_messages(tasks=task, n=1)) candidate_msgs_with_output = Message.sort_by_timestamp(messages=candidate_msgs_with_output) - prompt = OUTPUT_EXTRACTION_PROMPT.format( + prompt = WORKFLOW_OUTPUT_EXTRACTION_PROMPT.format( goal=graph.goal, workflow_graph_representation=graph.get_workflow_description(), workflow_execution_results="\n\n".join([str(msg) for msg in candidate_msgs_with_output]), diff --git a/examples/hitl/hitl_example.py b/examples/hitl/hitl_example.py index 9d7f5f3b..2ce5068c 100644 --- a/examples/hitl/hitl_example.py +++ b/examples/hitl/hitl_example.py @@ -165,9 +165,12 @@ async def main(): try: print("\n📋 start to execute the workflow") result = await workflow.async_execute(inputs=inputs) - print(f"\n✅ workflow executed successfully!") - print(f"result: {result}") - + if result.status == "success": + print(f"\n✅ workflow executed successfully!") + print(f"result: {result.result}") + else: + print(f"\n❌ workflow execution failed: {result.displayable_error}") + except Exception as e: print(f"\n❌ workflow execution failed: {e}") diff --git a/examples/hitl/hitl_example2.py b/examples/hitl/hitl_example2.py index dc800946..691da1b6 100644 --- a/examples/hitl/hitl_example2.py +++ b/examples/hitl/hitl_example2.py @@ -171,10 +171,15 @@ async def main(): ) print("\n" + "="*60) - print("🎉 workflow executed successfully!") - print("="*60) - print("final result:\n") - print(result) + if result.status == "success": + print("🎉 workflow executed successfully!") + print("="*60) + print("final result:\n") + print(result.result) + else: + print("❌ workflow execution failed!") + print("="*60) + print(result.displayable_error) except Exception as e: print(f"workflow execution failed: {e}") finally: diff --git a/examples/models/workflow_demo_lite_azure.py b/examples/models/workflow_demo_lite_azure.py index bd50d06c..b2116dcd 100644 --- a/examples/models/workflow_demo_lite_azure.py +++ b/examples/models/workflow_demo_lite_azure.py @@ -177,7 +177,12 @@ def execute_workflow(llm: LiteLLM, graph: WorkFlowGraph, goal: str, target_dir: mgr = AgentManager() mgr.add_agents_from_workflow(graph, llm_config=cfg) workflow = WorkFlow(graph=graph, agent_manager=mgr, llm=llm) - output = workflow.execute() + result = workflow.execute(extract_output=True) + if result.status != "success": + raise RuntimeError(f"Workflow failed: {result.displayable_error or result.error_msg}") + output = result.result + if not isinstance(output, str): + output = str(output) print("Workflow execution completed") return output @@ -230,4 +235,4 @@ def main(): print(f"\nComplete Tetris game has been generated to directory: {target_dir}") if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/examples/sequential_workflow.py b/examples/sequential_workflow.py index 37768c02..bd10810a 100644 --- a/examples/sequential_workflow.py +++ b/examples/sequential_workflow.py @@ -77,14 +77,17 @@ def build_sequential_workflow(): # create a workflow instance for execution workflow = WorkFlow(graph=graph, agent_manager=agent_manager, llm=llm) - output = workflow.execute( + result = workflow.execute( inputs = { "problem": "Write a function to find the longest palindromic substring in a given string. Save the code to local file: ./debug/test.py" } ) - - print("Workflow completed!") - print("Workflow output:\n", output) + + if result.status == "success": + print("Workflow completed!") + print("Workflow output:\n", result.result) + else: + print("Workflow failed:\n", result.displayable_error) diff --git a/examples/workflow/arxiv_workflow.py b/examples/workflow/arxiv_workflow.py index 420d16fe..3a547ee8 100644 --- a/examples/workflow/arxiv_workflow.py +++ b/examples/workflow/arxiv_workflow.py @@ -66,7 +66,14 @@ def main(): agent_manager.add_agents_from_workflow(workflow_graph, llm_config=openrouter_config, tools=tools) workflow = WorkFlow(graph=workflow_graph, agent_manager=agent_manager, llm=llm) - output = workflow.execute() + result = workflow.execute() + + if result.status != "success": + raise RuntimeError(f"Workflow failed: {result.displayable_error or result.error_msg}") + + output = result.result + if not isinstance(output, str): + output = str(output) with open(result_path, "w", encoding="utf-8") as f: f.write(output) diff --git a/examples/workflow/invest/stock_analysis.py b/examples/workflow/invest/stock_analysis.py index 64d3404e..77d2e7a7 100644 --- a/examples/workflow/invest/stock_analysis.py +++ b/examples/workflow/invest/stock_analysis.py @@ -186,7 +186,11 @@ def execute_workflow(stock_code, data_dir, report_dir, timestamp): Please read ALL files in the data folder and generate a comprehensive trading decision report in Chinese based on real data. Return the complete content. """ - output = workflow.execute({"goal": goal}) + result = workflow.execute({"goal": goal}, extract_output=True) + if result.status != "success": + print(f"Error executing workflow: {result.displayable_error}") + return + output = result.result try: with open(output_file, "w", encoding="utf-8") as f: f.write(output) diff --git a/examples/workflow/workflow_demo.py b/examples/workflow/workflow_demo.py index 0d0da18f..fa7ebb99 100644 --- a/examples/workflow/workflow_demo.py +++ b/examples/workflow/workflow_demo.py @@ -34,7 +34,12 @@ def main(): agent_manager.add_agents_from_workflow(workflow_graph, llm_config=openrouter_config) workflow = WorkFlow(graph=workflow_graph, agent_manager=agent_manager, llm=llm) - output = workflow.execute() + result = workflow.execute(extract_output=True) + if result.status != "success": + raise RuntimeError(f"Workflow failed: {result.displayable_error or result.error_msg}") + output = result.result + if not isinstance(output, str): + output = str(output) # verify the code code_verifier = CodeVerification() @@ -79,4 +84,4 @@ def main(): if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/examples/workflow/workflow_direction.py b/examples/workflow/workflow_direction.py index 6fce2558..24699676 100644 --- a/examples/workflow/workflow_direction.py +++ b/examples/workflow/workflow_direction.py @@ -59,7 +59,12 @@ def main(goal=None): agent_manager.add_agents_from_workflow(workflow_graph, llm_config=openrouter_config, tools=tools) workflow = WorkFlow(graph=workflow_graph, agent_manager=agent_manager, llm=llm) - output = workflow.execute() + result = workflow.execute(extract_output=True) + if result.status != "success": + raise RuntimeError(f"Workflow failed: {result.displayable_error or result.error_msg}") + output = result.result + if not isinstance(output, str): + output = str(output) ## _______________ Save Output _______________ diff --git a/examples/workflow_demo_with_tools.py b/examples/workflow_demo_with_tools.py index f991d0f0..0ae89628 100644 --- a/examples/workflow_demo_with_tools.py +++ b/examples/workflow_demo_with_tools.py @@ -52,10 +52,15 @@ def demo_basic_workflow(): agent_manager = AgentManager() agent_manager.add_agents_from_workflow(workflow_graph, llm_config=openai_config) - # Create and execute workflow - workflow = WorkFlow(graph=workflow_graph, agent_manager=agent_manager, llm=llm) - print("\nExecuting workflow...") - output = workflow.execute() + # Create and execute workflow + workflow = WorkFlow(graph=workflow_graph, agent_manager=agent_manager, llm=llm) + print("\nExecuting workflow...") + result = workflow.execute(extract_output=True) + if result.status != "success": + raise RuntimeError(f"Workflow failed: {result.displayable_error or result.error_msg}") + output = result.result + if not isinstance(output, str): + output = str(output) print("Basic workflow completed successfully") print(f"\nOutput (first 500 chars):\n{str(output)[:500]}...") @@ -96,10 +101,15 @@ def demo_toolkit_workflow(): agent_manager = AgentManager(tools=tools) agent_manager.add_agents_from_workflow(workflow_graph, llm_config=openai_config) - # Create and execute workflow - workflow = WorkFlow(graph=workflow_graph, agent_manager=agent_manager, llm=llm) - print("\nExecuting workflow with CMDToolkit...") - output = workflow.execute() + # Create and execute workflow + workflow = WorkFlow(graph=workflow_graph, agent_manager=agent_manager, llm=llm) + print("\nExecuting workflow with CMDToolkit...") + result = workflow.execute(extract_output=True) + if result.status != "success": + raise RuntimeError(f"Workflow failed: {result.displayable_error or result.error_msg}") + output = result.result + if not isinstance(output, str): + output = str(output) print("Toolkit workflow completed successfully") print(f"\nOutput (first 800 chars):\n{str(output)[:800]}...") @@ -203,4 +213,4 @@ def main(): return 1 if __name__ == "__main__": - exit(main()) \ No newline at end of file + exit(main()) diff --git a/tests/src/agents/test_agent.py b/tests/src/agents/test_agent.py index b4b9dd1e..ab36ce77 100644 --- a/tests/src/agents/test_agent.py +++ b/tests/src/agents/test_agent.py @@ -1,9 +1,25 @@ import os import unittest +from unittest.mock import patch from evoagentx.models.litellm_model import LiteLLM from evoagentx.models.model_configs import LiteLLMConfig from evoagentx.agents.agent import Agent -from evoagentx.actions.action import Action +from evoagentx.actions.action import Action, ActionOutput + + +class EchoOutput(ActionOutput): + result: str + + +class SyncOnlyTestAction(Action): + def __init__(self): + super().__init__(name="SyncOnlyTestAction", description="Echoes a value synchronously.") + + def execute(self, llm=None, inputs=None, sys_msg=None, return_prompt=False, **kwargs): + output = EchoOutput(result=inputs["value"]) + if return_prompt: + return output, "sync prompt" + return output class TestModule(unittest.TestCase): @@ -107,5 +123,24 @@ def tearDown(self): if os.path.exists(self.save_file): os.remove(self.save_file) + +class TestAgentAsyncExecution(unittest.IsolatedAsyncioTestCase): + async def test_async_execute_supports_sync_only_action_without_source_inspection(self): + agent = Agent( + name="SyncAgent", + description="Runs sync-only actions.", + actions=[SyncOnlyTestAction()], + is_human=True, + ) + + with patch("inspect.getsource", side_effect=OSError("source unavailable")): + message = await agent.async_execute( + action_name="SyncOnlyTestAction", + action_input_data={"value": "echo"}, + ) + + self.assertEqual(message.content.result, "echo") + self.assertEqual(message.prompt, "sync prompt") + if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main() diff --git a/tests/src/utils/test_validate_param.py b/tests/src/utils/test_validate_param.py new file mode 100644 index 00000000..5455940b --- /dev/null +++ b/tests/src/utils/test_validate_param.py @@ -0,0 +1,68 @@ +import unittest + +from evoagentx.core.base_config import Parameter +from evoagentx.utils.utils import validate_param + + +class TestValidateParam(unittest.TestCase): + + def _validate(self, required: Parameter, actual: Parameter): + validate_param(required, actual, "node", "agent") + + def test_differing_description_is_allowed(self): + """Description is free-form text and must not cause a validation failure.""" + required = Parameter(name="p", type="string", description="node-side wording") + actual = Parameter(name="p", type="string", description="agent-side wording") + # Should not raise. + self._validate(required, actual) + + def test_type_mismatch_raises(self): + required = Parameter(name="p", type="string", description="d") + actual = Parameter(name="p", type="integer", description="d") + with self.assertRaises(ValueError): + self._validate(required, actual) + + def test_required_mismatch_raises(self): + required = Parameter(name="p", type="string", description="d", required=True) + actual = Parameter(name="p", type="string", description="d", required=False) + with self.assertRaises(ValueError): + self._validate(required, actual) + + def test_json_schema_required_when_required_param_has_schema(self): + """A downstream param cannot drop a schema declared by the required param.""" + schema = {"type": "object", "properties": {"a": {"type": "string"}}} + required = Parameter(name="p", type="object", description="d", json_schema=schema) + actual = Parameter(name="p", type="object", description="d") # no json_schema + with self.assertRaises(ValueError): + self._validate(required, actual) + + def test_extra_actual_json_schema_allowed_when_required_param_has_none(self): + """A more specific actual param is allowed when the required param has no schema.""" + schema = {"type": "object", "properties": {"a": {"type": "string"}}} + required = Parameter(name="p", type="object", description="d") # no json_schema + actual = Parameter(name="p", type="object", description="d", json_schema=schema) + # Should not raise. + self._validate(required, actual) + + def test_json_schema_enforced_when_both_provided(self): + required = Parameter( + name="p", type="object", description="d", + json_schema={"type": "object", "properties": {"a": {"type": "string"}}}, + ) + actual = Parameter( + name="p", type="object", description="d", + json_schema={"type": "object", "properties": {"a": {"type": "integer"}}}, + ) + with self.assertRaises(ValueError): + self._validate(required, actual) + + def test_json_schema_match_passes(self): + schema = {"type": "object", "properties": {"a": {"type": "string"}}} + required = Parameter(name="p", type="object", description="d", json_schema=dict(schema)) + actual = Parameter(name="p", type="object", description="d", json_schema=dict(schema)) + # Should not raise. + self._validate(required, actual) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/src/workflow/test_action_graph.py b/tests/src/workflow/test_action_graph.py index 929f468d..2e95a596 100644 --- a/tests/src/workflow/test_action_graph.py +++ b/tests/src/workflow/test_action_graph.py @@ -1,12 +1,51 @@ import unittest import os -import pytest -from unittest.mock import patch +from typing import Optional +from unittest.mock import Mock, patch + +from evoagentx.core.base_config import Parameter from evoagentx.models import OpenAILLMConfig +from evoagentx.models.base_model import BaseLLM +from evoagentx.models.model_configs import LLMConfig from evoagentx.workflow.action_graph import ActionGraph, QAActionGraph +from evoagentx.workflow.workflow import WorkFlow +from evoagentx.workflow.workflow_graph import WorkFlowGraph, WorkFlowNode + + +class EchoActionGraph(ActionGraph): + llm_config: Optional[LLMConfig] = None + + def init_module(self): + pass + + def execute(self, value: str) -> dict: + return {"result": value} + + async def async_execute(self, value: str) -> dict: + return {"result": value} + + +class SyncOnlyActionGraph(ActionGraph): + llm_config: Optional[LLMConfig] = None + + def init_module(self): + pass + def execute(self, value: str) -> dict: + return {"result": value} -class TestModule(unittest.TestCase): + +class GoalEchoActionGraph(ActionGraph): + llm_config: Optional[LLMConfig] = None + + def init_module(self): + pass + + async def async_execute(self, goal: str) -> dict: + return {"result": goal} + + +class TestModule(unittest.IsolatedAsyncioTestCase): def setUp(self): self.llm_config = OpenAILLMConfig(model="gpt-4o-mini", openai_key="XXX") @@ -30,7 +69,6 @@ def test_execute(self, mock_sc_ensemble, mock_answer_generate): # Verify the result contains both answer and score self.assertEqual(result["answer"], "final answer") - @pytest.mark.asyncio @patch('evoagentx.workflow.operators.AnswerGenerate.async_execute') @patch('evoagentx.workflow.operators.QAScEnsemble.async_execute') async def test_async_execute(self, mock_sc_ensemble, mock_answer_generate): @@ -78,4 +116,131 @@ def test_save_and_load(self): def tearDown(self): if os.path.exists("tests/src/workflow/saved_qa_action_graph.json"): - os.remove("tests/src/workflow/saved_qa_action_graph.json") \ No newline at end of file + os.remove("tests/src/workflow/saved_qa_action_graph.json") + + +class TestActionGraphWorkflow(unittest.IsolatedAsyncioTestCase): + + def setUp(self): + self.workflow = WorkFlow( + graph=WorkFlowGraph( + goal="Echo input value", + nodes=[ + WorkFlowNode( + name="EchoTask", + description="Echo the provided input value.", + inputs=[Parameter(name="value", type="string", description="Input value")], + outputs=[Parameter(name="result", type="string", description="Echo result")], + action_graph=EchoActionGraph( + name="EchoActionGraph", + description="Echoes the input value.", + ), + ) + ], + ), + llm=Mock(spec=BaseLLM), + ) + + async def test_reusing_workflow_resets_environment_between_action_graph_runs(self): + first_result = await self.workflow.async_execute(inputs={"value": "first"}) + + self.assertEqual(first_result.status, "success") + self.assertEqual(first_result.result, {"result": "first"}) + self.assertGreater(len(self.workflow.environment.trajectory), 0) + self.assertEqual(self.workflow.environment.execution_data["value"], "first") + + self.workflow.environment.execution_data["stale"] = "leaked" + self.workflow.environment.task_execution_history.append("StaleTask") + + second_result = await self.workflow.async_execute(inputs={"value": "second"}) + + self.assertEqual(second_result.status, "success") + self.assertEqual(second_result.result, {"result": "second"}) + self.assertEqual(self.workflow.environment.execution_data["value"], "second") + self.assertEqual(self.workflow.environment.execution_data["result"], "second") + self.assertNotIn("stale", self.workflow.environment.execution_data) + self.assertNotIn("StaleTask", self.workflow.environment.task_execution_history) + + async def test_sync_execute_can_run_inside_running_event_loop(self): + result = self.workflow.execute(inputs={"value": "inside-loop"}) + + self.assertEqual(result.status, "success") + self.assertEqual(result.result, {"result": "inside-loop"}) + + async def test_workflow_executes_sync_only_action_graph(self): + workflow = WorkFlow( + graph=WorkFlowGraph( + goal="Echo input value", + nodes=[ + WorkFlowNode( + name="EchoTask", + description="Echo the provided input value.", + inputs=[Parameter(name="value", type="string", description="Input value")], + outputs=[Parameter(name="result", type="string", description="Echo result")], + action_graph=SyncOnlyActionGraph( + name="SyncOnlyActionGraph", + description="Echoes the input value synchronously.", + ), + ) + ], + ), + llm=Mock(spec=BaseLLM), + ) + + result = await workflow.async_execute(inputs={"value": "sync-only"}) + + self.assertEqual(result.status, "success") + self.assertEqual(result.result, {"result": "sync-only"}) + + async def test_action_graph_workflow_does_not_require_source_code_inspection(self): + with patch("inspect.getsource", side_effect=OSError("source unavailable")): + workflow = WorkFlow( + graph=WorkFlowGraph( + goal="Echo input value", + nodes=[ + WorkFlowNode( + name="EchoTask", + description="Echo the provided input value.", + inputs=[Parameter(name="value", type="string", description="Input value")], + outputs=[Parameter(name="result", type="string", description="Echo result")], + action_graph=EchoActionGraph( + name="EchoActionGraphNoSource", + description="Echoes the input value.", + ), + ) + ], + ), + llm=Mock(spec=BaseLLM), + ) + + result = await workflow.async_execute(inputs={"value": "no-source"}) + + self.assertEqual(result.status, "success") + self.assertEqual(result.result, {"result": "no-source"}) + + async def test_goal_injection_does_not_mutate_input_dict(self): + workflow = WorkFlow( + graph=WorkFlowGraph( + goal="Injected goal", + nodes=[ + WorkFlowNode( + name="GoalTask", + description="Echo the workflow goal.", + inputs=[Parameter(name="goal", type="string", description="Workflow goal")], + outputs=[Parameter(name="result", type="string", description="Echoed goal")], + action_graph=GoalEchoActionGraph( + name="GoalEchoActionGraph", + description="Echoes the workflow goal.", + ), + ) + ], + ), + llm=Mock(spec=BaseLLM), + ) + caller_inputs = {} + + result = await workflow.async_execute(inputs=caller_inputs) + + self.assertEqual(result.status, "success") + self.assertEqual(result.result, {"result": "Injected goal"}) + self.assertEqual(caller_inputs, {}) diff --git a/tests/src/workflow/test_workflow_graph.py b/tests/src/workflow/test_workflow_graph.py index d511a599..3f351890 100644 --- a/tests/src/workflow/test_workflow_graph.py +++ b/tests/src/workflow/test_workflow_graph.py @@ -1,6 +1,13 @@ import unittest +import pytest + from evoagentx.core.base_config import Parameter -from evoagentx.workflow.workflow_graph import WorkFlowNode, WorkFlowGraph, WorkFlowEdge, WorkFlowNodeState +from evoagentx.workflow.workflow_graph import ( + WorkFlowEdge, + WorkFlowGraph, + WorkFlowNode, + WorkFlowNodeState, +) class TestWorkFlowGraph(unittest.TestCase): @@ -30,30 +37,47 @@ def setUp(self): name="Task3", description="Third task", inputs=[Parameter(name="output2", type="string", description="Output from Task2")], - outputs=[Parameter(name="final_output", type="string", description="Final output")], + outputs=[Parameter(name="output3", type="string", description="Output from Task3")], agents=["TestAgent"], status=WorkFlowNodeState.PENDING ) - - # Create a fork-join workflow structure - # Task1 - # / - # Task2 -- Task3 - # \ / - # Task4 + self.task4 = WorkFlowNode( name="Task4", description="Fourth task (join)", inputs=[ Parameter(name="output2", type="string", description="Output from Task2"), - Parameter(name="final_output", type="string", description="Output from Task3") + Parameter(name="output3", type="string", description="Output from Task3") ], - outputs=[Parameter(name="result", type="string", description="Final result")], + outputs=[Parameter(name="output4", type="string", description="Output from Task4")], + agents=["TestAgent"], + status=WorkFlowNodeState.PENDING + ) + + self.task5 = WorkFlowNode( + name="Task5", + description="Fifth task (first node in loop)", + inputs=[ + Parameter(name="output1", type="string", description="Output from Task1"), + Parameter(name="output6", type="string", description="Output from Task6", required=False) + ], + outputs=[Parameter(name="output5", type="string", description="Output from Task5")], + agents=["TestAgent"], + status=WorkFlowNodeState.PENDING + ) + + self.task6 = WorkFlowNode( + name="Task6", + description="Sixth task (second node in loop)", + inputs=[ + Parameter(name="output5", type="string", description="Output from Task5") + ], + outputs=[Parameter(name="output6", type="string", description="Output from Task6")], agents=["TestAgent"], status=WorkFlowNodeState.PENDING ) - # Create a simple linear workflow + # Create a simple linear workflow: Task1 -> Task2 -> Task3 self.linear_graph = WorkFlowGraph( goal="Simple Linear Workflow", nodes=[self.task1, self.task2, self.task3], @@ -64,6 +88,11 @@ def setUp(self): ) # Create a fork-join workflow + # Task1 + # / + # Task2 -- Task3 + # \ / + # Task4 self.fork_join_graph = WorkFlowGraph( goal="Fork-Join Workflow", nodes=[self.task1, self.task2, self.task3, self.task4], @@ -76,14 +105,10 @@ def setUp(self): ) # Create a workflow with a cycle + # Task1 -> Task5 -> Task6 -> Task5 self.cycle_graph = WorkFlowGraph( goal="Workflow with Cycle", - nodes=[self.task1, self.task2, self.task3], - edges=[ - WorkFlowEdge(source="Task1", target="Task2"), - WorkFlowEdge(source="Task2", target="Task3"), - WorkFlowEdge(source="Task3", target="Task2") # Creates a cycle - ] + nodes=[self.task1, self.task5, self.task6] ) def test_graph_initialization(self): @@ -189,16 +214,138 @@ def test_fork_join_execution(self): self.fork_join_graph.set_node_status("Task4", WorkFlowNodeState.COMPLETED) next_tasks = self.fork_join_graph.next() self.assertEqual(0, len(next_tasks)) + + def test_workflow_completes_when_workflow_output_nodes_complete(self): + """Nodes that do not produce workflow outputs should not keep the workflow open.""" + node_a = WorkFlowNode( + name="A", + description="initial task", + inputs=[Parameter(name="input1", type="string", description="workflow input")], + outputs=[ + Parameter(name="target_input", type="string", description="feeds target output"), + Parameter(name="side_input", type="string", description="feeds side branch"), + ], + agents=["TestAgent"], + ) + node_b = WorkFlowNode( + name="B", + description="workflow output task", + inputs=[Parameter(name="target_input", type="string", description="from A")], + outputs=[Parameter(name="target_output", type="string", description="workflow output")], + agents=["TestAgent"], + ) + node_c = WorkFlowNode( + name="C", + description="side branch task", + inputs=[Parameter(name="side_input", type="string", description="from A")], + outputs=[Parameter(name="side_output", type="string", description="not a workflow output")], + agents=["TestAgent"], + ) + graph = WorkFlowGraph( + goal="Complete on workflow output", + nodes=[node_a, node_b, node_c], + workflow_inputs=[Parameter(name="input1", type="string", description="workflow input")], + workflow_outputs=[Parameter(name="target_output", type="string", description="workflow output")], + ) + + graph.set_node_status("A", WorkFlowNodeState.COMPLETED) + self.assertFalse(graph.is_complete) + + graph.set_node_status("B", WorkFlowNodeState.COMPLETED) + self.assertTrue(graph.is_complete) + self.assertEqual([], graph.next()) + def test_control_edge_not_executed_in_parallel(self): + """An explicit control edge (A -> B with no shared data) must be respected even + when B's required inputs are all workflow inputs and therefore B is data-initial.""" + node_a = WorkFlowNode( + name="A", + description="control source", + inputs=[Parameter(name="input1", type="string", description="workflow input")], + outputs=[Parameter(name="outputA", type="string", description="output A")], + agents=["TestAgent"], + ) + node_b = WorkFlowNode( + name="B", + description="control target", + inputs=[Parameter(name="input1", type="string", description="workflow input")], + outputs=[Parameter(name="outputB", type="string", description="output B")], + agents=["TestAgent"], + ) + graph = WorkFlowGraph( + goal="Control Edge Workflow", + nodes=[node_a, node_b], + edges=[WorkFlowEdge(source="A", target="B")], + workflow_inputs=[Parameter(name="input1", type="string", description="workflow input")], + workflow_outputs=[ + Parameter(name="outputA", type="string", description="output A"), + Parameter(name="outputB", type="string", description="output B"), + ], + ) + + # Both A and B are data-initial, but only A may run first. + self.assertEqual({"A", "B"}, set(graph.find_initial_nodes())) + next_tasks = graph.next() + self.assertEqual(1, len(next_tasks)) + self.assertEqual("A", next_tasks[0].name) + + graph.set_node_status("A", WorkFlowNodeState.COMPLETED) + next_tasks = graph.next() + self.assertEqual(1, len(next_tasks)) + self.assertEqual("B", next_tasks[0].name) + + def test_optional_input_edge_respects_dependency(self): + """An inferred edge feeding an optional input must be respected even though the + target's required inputs are all workflow inputs (so it is data-initial).""" + node_a = WorkFlowNode( + name="A", + description="optional source", + inputs=[Parameter(name="input1", type="string", description="workflow input")], + outputs=[Parameter(name="outA", type="string", description="output A")], + agents=["TestAgent"], + ) + node_b = WorkFlowNode( + name="B", + description="optional target", + inputs=[ + Parameter(name="input1", type="string", description="workflow input"), + Parameter(name="outA", type="string", description="optional from A", required=False), + ], + outputs=[Parameter(name="outputB", type="string", description="output B")], + agents=["TestAgent"], + ) + graph = WorkFlowGraph( + goal="Optional Input Workflow", + nodes=[node_a, node_b], + workflow_inputs=[Parameter(name="input1", type="string", description="workflow input")], + workflow_outputs=[ + Parameter(name="outA", type="string", description="output A"), + Parameter(name="outputB", type="string", description="output B"), + ], + ) + + # The A -> B edge is inferred from the shared `outA` name (B's optional input). + edge_pairs = [(edge.source, edge.target) for edge in graph.edges] + self.assertIn(("A", "B"), edge_pairs) + + # B is data-initial (only required input is the workflow input) but must wait for A. + self.assertEqual({"A", "B"}, set(graph.find_initial_nodes())) + next_tasks = graph.next() + self.assertEqual(1, len(next_tasks)) + self.assertEqual("A", next_tasks[0].name) + + graph.set_node_status("A", WorkFlowNodeState.COMPLETED) + next_tasks = graph.next() + self.assertEqual(1, len(next_tasks)) + self.assertEqual("B", next_tasks[0].name) + def test_cycle_detection(self): """Test cycle detection in a workflow.""" # The cycle graph should identify a loop loops = self.cycle_graph._find_all_loops() self.assertTrue(loops) # Should contain at least one loop - - # Check if Task1 is identified as both a loop start and loop end - self.assertTrue(self.cycle_graph.is_loop_start("Task2")) - self.assertTrue(self.cycle_graph.is_loop_end("Task3")) + self.assertTrue(self.cycle_graph.is_loop_start("Task5")) + self.assertTrue(self.cycle_graph.is_loop_end("Task6")) def test_node_status_management(self): """Test node status management.""" @@ -258,6 +405,355 @@ def test_graph_dependency_checking(self): self.fork_join_graph.set_node_status("Task3", WorkFlowNodeState.COMPLETED) self.assertTrue(self.fork_join_graph.are_dependencies_complete("Task4")) + def test_workflow_io_duplicates(self): + """Test that workflow inputs and outputs cannot have duplicate names.""" + with pytest.raises(ValueError, match=r"Workflow inputs and outputs share the following name\(s\), which is not allowed: \['shared'\]"): + WorkFlowGraph( + goal="Duplicate IO", + nodes=[self.task1], + workflow_inputs=[Parameter(name="shared", type="string", description="desc")], + workflow_outputs=[Parameter(name="shared", type="string", description="desc")] + ) + + def test_node_io_duplicates(self): + """Test that node inputs and outputs cannot have internal duplicates or overlap.""" + # Duplicate input name + node_dup_in = WorkFlowNode( + name="DupIn", + description="test", + inputs=[ + Parameter(name="in1", type="string", description="desc"), + Parameter(name="in1", type="string", description="desc") + ], + outputs=[Parameter(name="out1", type="string", description="desc")], + agents=["TestAgent"] + ) + with pytest.raises(ValueError, match="Node 'DupIn' has duplicate input name: 'in1'"): + WorkFlowGraph( + goal="test", + nodes=[node_dup_in], + workflow_inputs=[Parameter(name="workflow_in", type="string", description="desc")], + workflow_outputs=[Parameter(name="workflow_out", type="string", description="desc")] + ) + + # Duplicate output name + node_dup_out = WorkFlowNode( + name="DupOut", + description="test", + inputs=[Parameter(name="in1", type="string", description="desc")], + outputs=[ + Parameter(name="out1", type="string", description="desc"), + Parameter(name="out1", type="string", description="desc") + ], + agents=["TestAgent"] + ) + with pytest.raises(ValueError, match="Node 'DupOut' has duplicate output name: 'out1'"): + WorkFlowGraph( + goal="test", + nodes=[node_dup_out], + workflow_inputs=[Parameter(name="workflow_in", type="string", description="desc")], + workflow_outputs=[Parameter(name="workflow_out", type="string", description="desc")] + ) + + # Overlap between inputs and outputs + node_overlap = WorkFlowNode( + name="Overlap", + description="test", + inputs=[Parameter(name="shared", type="string", description="desc")], + outputs=[Parameter(name="shared", type="string", description="desc")], + agents=["TestAgent"] + ) + with pytest.raises(ValueError, match=r"Node 'Overlap' inputs and outputs share the following name\(s\), which is not allowed: \['shared'\]"): + WorkFlowGraph( + goal="test", + nodes=[node_overlap], + workflow_inputs=[Parameter(name="workflow_in", type="string", description="desc")], + workflow_outputs=[Parameter(name="workflow_out", type="string", description="desc")] + ) + + def test_node_output_uniqueness(self): + """Test that each output name must be unique across all nodes.""" + node1 = WorkFlowNode( + name="Node1", + description="test", + inputs=[Parameter(name="in1", type="string", description="desc")], + outputs=[Parameter(name="out_shared", type="string", description="desc")], + agents=["TestAgent"] + ) + node2 = WorkFlowNode( + name="Node2", + description="test", + inputs=[Parameter(name="in2", type="string", description="desc")], + outputs=[Parameter(name="out_shared", type="string", description="desc")], + agents=["TestAgent"] + ) + expected_msg = ( + r"Each node output name must be unique across all nodes\. " + r"Found conflicts:\n'out_shared' produced by \['Node1', 'Node2'\]" + ) + with pytest.raises(ValueError, match=expected_msg): + WorkFlowGraph( + goal="test", + nodes=[node1, node2], + workflow_inputs=[Parameter(name="workflow_in", type="string", description="desc")], + workflow_outputs=[Parameter(name="workflow_out", type="string", description="desc")] + ) + + def test_mismatched_params_raise(self): + """A node parameter that mismatches the workflow parameter (type/required) must raise.""" + mismatched_node = WorkFlowNode( + name="MismatchedNode", + description="test", + inputs=[Parameter(name="input", type="number", description="desc", required=False)], # Should be string, required=True + outputs=[Parameter(name="output", type="boolean", description="desc")], + agents=["TestAgent"] + ) + + with pytest.raises(ValueError): + WorkFlowGraph( + goal="Test mismatch", + nodes=[mismatched_node], + workflow_inputs=[Parameter(name="input", type="string", description="desc", required=True)], + workflow_outputs=[Parameter(name="output", type="string", description="desc")], + ) + + def test_json_schema_cannot_be_dropped_downstream(self): + """A node/agent cannot omit a schema declared by the workflow output.""" + schema = { + "type": "object", + "properties": {"answer": {"type": "string"}}, + "required": ["answer"], + } + graph_dict = { + "goal": "Schema contract", + "nodes": [{ + "name": "SchemaNode", + "description": "test", + "inputs": [{"name": "input", "type": "string", "description": "desc"}], + "outputs": [{"name": "output", "type": "object", "description": "desc"}], + "agents": [{ + "name": "SchemaAgent", + "description": "test", + "inputs": [{"name": "input", "type": "string", "description": "desc"}], + "outputs": [{"name": "output", "type": "object", "description": "desc"}], + }], + }], + "workflow_inputs": [{"name": "input", "type": "string", "description": "desc"}], + "workflow_outputs": [{ + "name": "output", + "type": "object", + "description": "desc", + "json_schema": schema, + }], + } + + with pytest.raises(Exception, match="json_schema"): + WorkFlowGraph.from_dict(graph_dict) + + def test_object_params_without_any_json_schema_are_allowed(self): + """object params remain valid when no workflow/node/agent layer declares a schema.""" + graph_dict = { + "goal": "Schema optional", + "nodes": [{ + "name": "SchemaNode", + "description": "test", + "inputs": [{"name": "input", "type": "string", "description": "desc"}], + "outputs": [{"name": "output", "type": "object", "description": "desc"}], + "agents": [{ + "name": "SchemaAgent", + "description": "test", + "inputs": [{"name": "input", "type": "string", "description": "desc"}], + "outputs": [{"name": "output", "type": "object", "description": "desc"}], + }], + }], + "workflow_inputs": [{"name": "input", "type": "string", "description": "desc"}], + "workflow_outputs": [{"name": "output", "type": "object", "description": "desc"}], + } + + graph = WorkFlowGraph.from_dict(graph_dict) + self.assertIsNone(graph.get_node("SchemaNode").outputs[0].json_schema) + self.assertNotIn("json_schema", graph.get_node("SchemaNode").agents[0]["outputs"][0]) + + def test_from_dict_preserves_explicit_edges(self): + """Explicit control edges (no shared input/output) must survive a from_dict round-trip; + they cannot be re-inferred from data flow.""" + + def make_agent(name, in_name, out_name): + return { + "name": name, + "description": "test", + "inputs": [{"name": in_name, "type": "string", "description": "desc"}], + "outputs": [{"name": out_name, "type": "string", "description": "desc"}], + "prompt_template": {"class_name": "ChatTemplate", "instruction": "instruction"}, + } + + graph_dict = { + "goal": "Test Explicit Edges", + "nodes": [ + { + "name": "A", + "description": "control source", + "inputs": [{"name": "wf_in", "type": "string", "description": "desc"}], + "outputs": [{"name": "outA", "type": "string", "description": "desc"}], + "agents": [make_agent("AgentA", "wf_in", "outA")], + }, + { + "name": "B", + "description": "control target", + "inputs": [{"name": "wf_in", "type": "string", "description": "desc"}], + "outputs": [{"name": "outB", "type": "string", "description": "desc"}], + "agents": [make_agent("AgentB", "wf_in", "outB")], + }, + ], + # A -> B shares no input/output, so it cannot be inferred from data flow. + "edges": [{"source": "A", "target": "B"}], + "workflow_inputs": [{"name": "wf_in", "type": "string", "description": "desc"}], + "workflow_outputs": [ + {"name": "outA", "type": "string", "description": "desc"}, + {"name": "outB", "type": "string", "description": "desc"}, + ], + } + + graph = WorkFlowGraph.from_dict(graph_dict) + + edge_pairs = {(edge.source, edge.target) for edge in graph.edges} + self.assertIn(("A", "B"), edge_pairs) + + # The restored control edge must still gate execution: B waits for A. + next_tasks = graph.next() + self.assertEqual(["A"], [task.name for task in next_tasks]) + + def test_explicit_edge_priority_overrides_inferred_edge_priority(self): + """When an explicit edge matches an inferred data-flow edge, preserve the explicit metadata.""" + graph_dict = { + "goal": "Test Explicit Edge Priority", + "nodes": [ + { + "name": "A", + "description": "source", + "inputs": [{"name": "wf_in", "type": "string", "description": "desc"}], + "outputs": [{"name": "outA", "type": "string", "description": "desc"}], + "agents": ["AgentA"], + }, + { + "name": "B", + "description": "target", + "inputs": [{"name": "outA", "type": "string", "description": "desc"}], + "outputs": [{"name": "outB", "type": "string", "description": "desc"}], + "agents": ["AgentB"], + }, + ], + # A -> B is also inferred from outA, but the explicit priority must win. + "edges": [{"source": "A", "target": "B", "priority": 7}], + "workflow_inputs": [{"name": "wf_in", "type": "string", "description": "desc"}], + "workflow_outputs": [{"name": "outB", "type": "string", "description": "desc"}], + } + + graph = WorkFlowGraph.from_dict(graph_dict) + + self.assertEqual([("A", "B", 7)], [(edge.source, edge.target, edge.priority) for edge in graph.edges]) + graph_edge_refs = [ + attrs["ref"] + for source, target, attrs in graph.graph.edges(data=True) + if source == "A" and target == "B" + ] + self.assertEqual([7], [edge.priority for edge in graph_edge_refs]) + + def test_to_dict_supports_string_and_dict_agents(self): + """get_config()/to_dict() must support string agents and convert a callable + parse_func in a dict agent to its function name (JSON-serializable).""" + import json + + def my_parser(x): + return x + + node_a = WorkFlowNode( + name="A", + description="d", + inputs=[Parameter(name="i", type="string", description="x")], + outputs=[Parameter(name="o", type="string", description="x")], + agents=["StrAgent"], + ) + node_b = WorkFlowNode( + name="B", + description="d", + inputs=[Parameter(name="o", type="string", description="x")], + outputs=[Parameter(name="o2", type="string", description="x")], + agents=[{"name": "DAgent", "description": "d", "parse_func": my_parser}], + ) + graph = WorkFlowGraph( + goal="g", + nodes=[node_a, node_b], + workflow_inputs=[Parameter(name="i", type="string", description="x")], + workflow_outputs=[Parameter(name="o2", type="string", description="x")], + ) + + config = graph.get_config() + + # string agent is preserved as-is + self.assertEqual(["StrAgent"], config["nodes"][0]["agents"]) + # callable parse_func is converted to its name + self.assertEqual("my_parser", config["nodes"][1]["agents"][0]["parse_func"]) + # whole config is JSON-serializable + json.dumps(config) + + def test_derived_workflow_inputs_deduplicated(self): + """When workflow_inputs is not provided, two initial nodes sharing the same input + name must not produce a duplicate (which the uniqueness check would reject).""" + node_a = WorkFlowNode( + name="A", + description="d", + inputs=[Parameter(name="shared_in", type="string", description="x")], + outputs=[Parameter(name="outA", type="string", description="x")], + agents=["TestAgent"], + ) + node_b = WorkFlowNode( + name="B", + description="d", + inputs=[Parameter(name="shared_in", type="string", description="x")], + outputs=[Parameter(name="outB", type="string", description="x")], + agents=["TestAgent"], + ) + # No workflow_inputs/outputs provided -> they are derived from initial/end nodes. + graph = WorkFlowGraph(goal="g", nodes=[node_a, node_b]) + + input_names = [param.name for param in graph.workflow_inputs] + self.assertEqual(["shared_in"], input_names) + self.assertIn("shared_in", graph.workflow_inputs_dict) + + def test_node_input_unknown_source_raises(self): + """A node input that is neither a workflow input nor any other node's output must fail.""" + workflow_input = Parameter(name="wf_in", type="string", description="workflow input", required=True) + workflow_output = Parameter(name="wf_out", type="string", description="workflow output") + + nodeA = WorkFlowNode( + name="NodeA", + description="NodeA", + inputs=[Parameter(name="wf_in", type="string", description="workflow input", required=True)], + outputs=[Parameter(name="nodeA_out", type="string", description="node A output")] + ) + + nodeB = WorkFlowNode( + name="NodeB", + description="NodeB", + inputs=[ + Parameter(name="ghost_input", type="string", description="ghost_input", required=True), + Parameter(name="nodeA_out", type="string", description="node A output", required=True) + ], + outputs=[Parameter(name="wf_out", type="string", description="workflow output")] + ) + + with pytest.raises( + ValueError, + match="Node 'NodeB' input 'ghost_input' is not a workflow input or from another node's output.", + ): + WorkFlowGraph( + goal="test", + nodes=[nodeA, nodeB], + workflow_inputs=[workflow_input], + workflow_outputs=[workflow_output], + ) + if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main() diff --git a/tests/src/workflow/test_workflow_manager.py b/tests/src/workflow/test_workflow_manager.py index 73ccc103..25d0a505 100644 --- a/tests/src/workflow/test_workflow_manager.py +++ b/tests/src/workflow/test_workflow_manager.py @@ -1,5 +1,4 @@ import unittest -import pytest from unittest.mock import Mock, patch, AsyncMock from evoagentx.core.message import Message, MessageType @@ -17,7 +16,7 @@ def to_str(self, **kwargs) -> str: return self.content -class TestWorkFlowManager(unittest.TestCase): +class TestWorkFlowManager(unittest.IsolatedAsyncioTestCase): def setUp(self): # Create mock LLM @@ -98,7 +97,6 @@ def test_workflow_initialization(self): self.assertIsNotNone(self.workflow_manager.task_scheduler) self.assertIsNotNone(self.workflow_manager.action_scheduler) - @pytest.mark.asyncio @patch('evoagentx.workflow.workflow_manager.TaskScheduler.async_execute') async def test_sync_task_scheduling_with_single_task(self, mock_task_scheduler_execute): """Test that the task scheduler correctly handles the case of a single candidate task""" @@ -108,8 +106,10 @@ async def test_sync_task_scheduling_with_single_task(self, mock_task_scheduler_e task_name="Task2", reason="Only one candidate task is available" ) - mock_task_scheduler_execute.return_value = single_task_output - + # schedule_next_task calls async_execute with return_prompt=True and unpacks a + # (scheduled_task, prompt) tuple, so the mock must return a tuple. + mock_task_scheduler_execute.return_value = (single_task_output, "mock prompt") + # Mark Task1 as completed to make Task2 the only next candidate self.workflow.set_node_status("Task1", WorkFlowNodeState.COMPLETED) @@ -127,7 +127,6 @@ async def test_sync_task_scheduling_with_single_task(self, mock_task_scheduler_e self.assertEqual("Task2", message.content.task_name) self.assertEqual(MessageType.COMMAND, message.msg_type) - @pytest.mark.asyncio @patch('evoagentx.workflow.workflow_manager.ActionScheduler.async_execute') async def test_action_scheduling(self, mock_action_scheduler_execute): """Test scheduling the next action for a task""" @@ -160,57 +159,67 @@ async def test_action_scheduling(self, mock_action_scheduler_execute): self.assertEqual("TestAction", message.content.action) self.assertEqual(MessageType.COMMAND, message.msg_type) - @pytest.mark.asyncio async def test_async_task_scheduling(self): - """Test async task scheduling with multiple candidate tasks""" - # Set up the llm.async_generate to return a task + """Test async task scheduling: with a single candidate task the scheduler + forwards to it directly without consulting the LLM.""" + # Set up the llm.async_generate to return a task (it should NOT be used here) self.mock_llm.async_generate.return_value = self.task_output - - # Run the test + + # With all nodes pending, the only entry/candidate task is Task1 task = await self.workflow_manager.schedule_next_task(graph=self.workflow, env=self.env) - + # Check results self.assertIsNotNone(task) - self.assertEqual("Task2", task.name) - + self.assertEqual("Task1", task.name) + # The single-candidate edge case short-circuits without an LLM call + self.mock_llm.async_generate.assert_not_called() + # Verify the environment was updated self.assertEqual(1, len(self.env.trajectory)) message = self.env.trajectory[0].message - self.assertEqual(self.task_output, message.content) + self.assertIsInstance(message.content, TaskSchedulerOutput) + self.assertEqual("Task1", message.content.task_name) self.assertEqual(TrajectoryState.COMPLETED, self.env.trajectory[0].status) - @pytest.mark.asyncio async def test_async_action_scheduling(self): - """Test async action scheduling""" - # Set up the llm.async_generate to return an action + """Test async action scheduling: a task with a single agent that has a single + action resolves to that action directly without consulting the LLM.""" + # Set up the llm.async_generate to return an action (it should NOT be used here) self.mock_llm.async_generate.return_value = self.action_output - - # Get the first task node + + # Get the first task node (it references the single agent 'TestAgent') task = self.workflow.get_node("Task1") - - # Create a mock agent manager + + # Mock agent manager that resolves 'TestAgent' to an agent exposing one action + mock_action = Mock() + mock_action.name = "TestAction" + mock_agent = Mock() + mock_agent.name = "TestAgent" + mock_agent.get_all_actions.return_value = [mock_action] mock_agent_manager = Mock() - + mock_agent_manager.get_agent.return_value = mock_agent + # Run the test action = await self.workflow_manager.schedule_next_action( - goal="Test Goal", - task=task, - agent_manager=mock_agent_manager, + goal="Test Goal", + task=task, + agent_manager=mock_agent_manager, env=self.env ) - + # Check results self.assertIsNotNone(action) self.assertEqual("TestAgent", action.agent) self.assertEqual("TestAction", action.action) - + # The single-agent/single-action case short-circuits without an LLM call + self.mock_llm.async_generate.assert_not_called() + # Verify the environment was updated self.assertEqual(1, len(self.env.trajectory)) message = self.env.trajectory[0].message - self.assertEqual(self.action_output, message.content) + self.assertEqual(action, message.content) self.assertEqual(TrajectoryState.COMPLETED, self.env.trajectory[0].status) - @pytest.mark.asyncio async def test_output_extraction(self): """Test extracting the output from the workflow execution""" # Set up the llm.async_generate to return an output @@ -247,7 +256,6 @@ async def test_output_extraction(self): # Verify the LLM was called self.mock_llm.async_generate.assert_called_once() - @pytest.mark.asyncio async def test_edge_case_handling(self): """Test edge case handling in workflow management""" # Test case: No tasks available for scheduling