diff --git a/docs/pyagentspec/source/agentspec/json_spec/agentspec_json_spec_26_1_2.json b/docs/pyagentspec/source/agentspec/json_spec/agentspec_json_spec_26_1_2.json index ea2b97ebe..3839913ea 100644 --- a/docs/pyagentspec/source/agentspec/json_spec/agentspec_json_spec_26_1_2.json +++ b/docs/pyagentspec/source/agentspec/json_spec/agentspec_json_spec_26_1_2.json @@ -2847,7 +2847,7 @@ }, "BaseFlowNode": { "additionalProperties": false, - "description": "The flow node executes a subflow as part of a flow.\n\n- **Inputs**\n Inferred from the inner structure. It's the sets of inputs\n required by the StartNode of the inner flow.\n- **Outputs**\n Inferred from the inner structure. It's the union of the\n sets of outputs exposed by the EndNodes of the inner flow.\n- **Branches**\n Inferred from the inner flow, one per each different value of the attribute\n ``branch_name`` of the nodes of type EndNode in the inner flow.\n\nExample\n-------\nThe ``FlowNode`` is particularly suitable when subflows can be reused inside a project.\nLet's see an example with a flow that estimates numerical value\nusing the \"wisdowm of the crowd\" effect:\n\n>>> from pyagentspec.property import Property\n>>> from pyagentspec.flows.flow import Flow\n>>> from pyagentspec.flows.edges import ControlFlowEdge, DataFlowEdge\n>>> from pyagentspec.flows.nodes import MapNode, LlmNode, ToolNode, StartNode, EndNode\n>>> from pyagentspec.tools import ServerTool\n>>> duplication_tool = ServerTool(\n... name=\"duplication_tool\",\n... description=\"\",\n... inputs=[\n... Property(\n... json_schema={\"title\": \"element\", \"description\": \"\", \"type\": \"string\"}\n... ),\n... Property(\n... json_schema={\"title\": \"n\", \"description\": \"\", \"type\": \"integer\"}\n... ),\n... ],\n... outputs=[\n... Property(\n... json_schema={\n... \"title\": \"flow_iterable_queries\",\n... \"type\": \"array\",\n... \"items\": {\"type\": \"string\"}\n... },\n... )\n... ],\n... )\n>>> reduce_tool = ServerTool(\n... name=\"reduce_tool\",\n... description=\"\",\n... inputs=[\n... Property(\n... json_schema={\"title\": \"elements\", \"type\": \"array\", \"items\": {\"type\": \"string\"}}\n... ),\n... ],\n... outputs=[Property(json_schema={\"title\": \"flow_processed_query\", \"type\": \"string\"})],\n... )\n>>> # Defining a simple prompt\n>>> REASONING_PROMPT_TEMPLATE = '''Provide your best numerical estimate for: {{user_input}}\n... Your answer should be a single number.\n... Do not include any units, reasoning, or extra text.'''\n>>> # Defining the subflow for the map step\n>>> user_input_property = Property(\n... json_schema={\"title\": \"user_input\", \"type\": \"string\"}\n... )\n>>> flow_processed_query_property = Property(\n... json_schema={\"title\": \"flow_processed_query\", \"type\": \"string\"}\n... )\n>>> start_node = StartNode(name=\"start\", inputs=[user_input_property])\n>>> end_node = EndNode(name=\"end\", outputs=[flow_processed_query_property])\n>>> llm_node = LlmNode(\n... name=\"reasoning llm node\",\n... llm_config=llm_config,\n... prompt_template=REASONING_PROMPT_TEMPLATE,\n... inputs=[user_input_property],\n... outputs=[flow_processed_query_property],\n... )\n>>> inner_map_flow = Flow(\n... name=\"Map flow\",\n... start_node=start_node,\n... nodes=[start_node, llm_node, end_node],\n... control_flow_connections=[\n... ControlFlowEdge(name=\"start_to_llm\", from_node=start_node, to_node=llm_node),\n... ControlFlowEdge(name=\"llm_to_end\", from_node=llm_node, to_node=end_node),\n... ],\n... data_flow_connections=[\n... DataFlowEdge(\n... name=\"query_edge\",\n... source_node=start_node,\n... source_output=\"user_input\",\n... destination_node=llm_node,\n... destination_input=\"user_input\",\n... ),\n... DataFlowEdge(\n... name=\"search_results_edge\",\n... source_node=llm_node,\n... source_output=\"flow_processed_query\",\n... destination_node=end_node,\n... destination_input=\"flow_processed_query\"\n... ),\n... ],\n... )\n>>> user_query_property = Property(\n... json_schema={\"title\": \"user_query\", \"type\": \"string\"}\n... )\n>>> n_repeat_property = Property(\n... json_schema={\"title\": \"n_repeat\", \"type\": \"integer\"}\n... )\n>>> flow_iterable_queries_property = Property(\n... json_schema={\n... \"title\": \"iterated_user_input\",\n... \"type\": \"array\",\n... \"items\": {\"type\": \"string\"},\n... }\n... )\n>>> flow_processed_queries_property = Property(\n... json_schema={\n... \"title\": \"collected_flow_processed_query\",\n... \"type\": \"array\",\n... \"items\": {\"type\": \"string\"},\n... }\n... )\n>>> start_node = StartNode(name=\"start\", inputs=[user_query_property, n_repeat_property])\n>>> end_node = EndNode(name=\"end\", outputs=[flow_processed_query_property])\n>>> duplication_node = ToolNode(\n... name=\"duplication_tool node\",\n... tool=duplication_tool,\n... )\n>>> reduce_node = ToolNode(\n... name=\"reduce_tool node\",\n... tool=reduce_tool,\n... )\n>>> map_node = MapNode(\n... name=\"map node\",\n... subflow=inner_map_flow,\n... inputs=[flow_iterable_queries_property],\n... outputs=[flow_processed_queries_property],\n... )\n>>> mapreduce_flow = Flow(\n... name=\"Map-reduce flow\",\n... start_node=start_node,\n... nodes=[start_node, duplication_node, map_node, reduce_node, end_node],\n... control_flow_connections=[\n... ControlFlowEdge(\n... name=\"start_to_duplication\", from_node=start_node, to_node=duplication_node\n... ),\n... ControlFlowEdge(\n... name=\"duplication_to_map\", from_node=duplication_node, to_node=map_node\n... ),\n... ControlFlowEdge(name=\"map_to_reduce\", from_node=map_node, to_node=reduce_node),\n... ControlFlowEdge(name=\"reduce_to_end\", from_node=reduce_node, to_node=end_node),\n... ],\n... data_flow_connections=[\n... DataFlowEdge(\n... name=\"query_edge\",\n... source_node=start_node,\n... source_output=\"user_query\",\n... destination_node=duplication_node,\n... destination_input=\"element\",\n... ),\n... DataFlowEdge(\n... name=\"n_repeat_edge\",\n... source_node=start_node,\n... source_output=\"n_repeat\",\n... destination_node=duplication_node,\n... destination_input=\"n\",\n... ),\n... DataFlowEdge(\n... name=\"flow_iterables_edge\",\n... source_node=duplication_node,\n... source_output=\"flow_iterable_queries\",\n... destination_node=map_node,\n... destination_input=\"iterated_user_input\",\n... ),\n... DataFlowEdge(\n... name=\"flow_processed_queries_edge\",\n... source_node=map_node,\n... source_output=\"collected_flow_processed_query\",\n... destination_node=reduce_node,\n... destination_input=\"elements\",\n... ),\n... DataFlowEdge(\n... name=\"flow_processed_query_edge\",\n... source_node=reduce_node,\n... source_output=\"flow_processed_query\",\n... destination_node=end_node,\n... destination_input=\"flow_processed_query\",\n... ),\n... ],\n... )\n\nOnce the subflow is created we can simply integrate it with the ``FlowNode``:\n\n>>> from pyagentspec.flows.nodes import FlowNode, AgentNode\n>>> from pyagentspec.agent import Agent\n>>> start_node = StartNode(name=\"start\")\n>>> end_node = EndNode(name=\"end\", outputs=[flow_processed_query_property])\n>>> flow_node = FlowNode(name=\"flow node\", subflow=mapreduce_flow)\n>>> agent = Agent(\n... name=\"User interaction agent\",\n... llm_config=llm_config,\n... system_prompt=(\n... \"Your task is to gather from the user the query and the number of times \"\n... \"it should be asked to an LLM. Once you have this information, submit and exit.\"\n... ),\n... outputs=[user_query_property, n_repeat_property],\n... )\n>>> agent_node = AgentNode(name=\"flow node\", agent=agent)\n>>> flow = Flow(\n... name=\"Map flow\",\n... start_node=start_node,\n... nodes=[start_node, agent_node, flow_node, end_node],\n... control_flow_connections=[\n... ControlFlowEdge(name=\"start_to_agent\", from_node=start_node, to_node=agent_node),\n... ControlFlowEdge(name=\"agent_to_flow\", from_node=agent_node, to_node=flow_node),\n... ControlFlowEdge(name=\"flow_to_end\", from_node=flow_node, to_node=end_node),\n... ],\n... data_flow_connections=[\n... DataFlowEdge(\n... name=\"query_edge\",\n... source_node=agent_node,\n... source_output=\"user_query\",\n... destination_node=flow_node,\n... destination_input=\"user_query\",\n... ),\n... DataFlowEdge(\n... name=\"n_rep_edge\",\n... source_node=agent_node,\n... source_output=\"n_repeat\",\n... destination_node=flow_node,\n... destination_input=\"n_repeat\"\n... ),\n... DataFlowEdge(\n... name=\"n_rep_edge\",\n... source_node=flow_node,\n... source_output=\"flow_processed_query\",\n... destination_node=end_node,\n... destination_input=\"flow_processed_query\"\n... ),\n... ],\n... )", + "description": "The flow node executes a subflow as part of a flow.\n\n- **Inputs**\n Inferred from the inner structure. It's the sets of inputs\n required by the StartNode of the inner flow.\n- **Outputs**\n Inferred from the inner structure. It's the union of the\n sets of outputs exposed by the EndNodes of the inner flow.\n- **Branches**\n Inferred from the inner flow, one per each different value of the attribute\n ``branch_name`` of the nodes of type EndNode in the inner flow.\n- **Pending input**\n If the subflow execution pauses to request input from a user or external caller,\n the pending input state is propagated through the ``FlowNode`` by default. This\n preserves the inline semantics of the subflow and lets the parent flow surface\n the child prompt and resume only after the child input is supplied.\n\nExample\n-------\nThe ``FlowNode`` is particularly suitable when subflows can be reused inside a project.\nLet's see an example with a flow that estimates numerical value\nusing the \"wisdowm of the crowd\" effect:\n\n>>> from pyagentspec.property import Property\n>>> from pyagentspec.flows.flow import Flow\n>>> from pyagentspec.flows.edges import ControlFlowEdge, DataFlowEdge\n>>> from pyagentspec.flows.nodes import MapNode, LlmNode, ToolNode, StartNode, EndNode\n>>> from pyagentspec.tools import ServerTool\n>>> duplication_tool = ServerTool(\n... name=\"duplication_tool\",\n... description=\"\",\n... inputs=[\n... Property(\n... json_schema={\"title\": \"element\", \"description\": \"\", \"type\": \"string\"}\n... ),\n... Property(\n... json_schema={\"title\": \"n\", \"description\": \"\", \"type\": \"integer\"}\n... ),\n... ],\n... outputs=[\n... Property(\n... json_schema={\n... \"title\": \"flow_iterable_queries\",\n... \"type\": \"array\",\n... \"items\": {\"type\": \"string\"}\n... },\n... )\n... ],\n... )\n>>> reduce_tool = ServerTool(\n... name=\"reduce_tool\",\n... description=\"\",\n... inputs=[\n... Property(\n... json_schema={\"title\": \"elements\", \"type\": \"array\", \"items\": {\"type\": \"string\"}}\n... ),\n... ],\n... outputs=[Property(json_schema={\"title\": \"flow_processed_query\", \"type\": \"string\"})],\n... )\n>>> # Defining a simple prompt\n>>> REASONING_PROMPT_TEMPLATE = '''Provide your best numerical estimate for: {{user_input}}\n... Your answer should be a single number.\n... Do not include any units, reasoning, or extra text.'''\n>>> # Defining the subflow for the map step\n>>> user_input_property = Property(\n... json_schema={\"title\": \"user_input\", \"type\": \"string\"}\n... )\n>>> flow_processed_query_property = Property(\n... json_schema={\"title\": \"flow_processed_query\", \"type\": \"string\"}\n... )\n>>> start_node = StartNode(name=\"start\", inputs=[user_input_property])\n>>> end_node = EndNode(name=\"end\", outputs=[flow_processed_query_property])\n>>> llm_node = LlmNode(\n... name=\"reasoning llm node\",\n... llm_config=llm_config,\n... prompt_template=REASONING_PROMPT_TEMPLATE,\n... inputs=[user_input_property],\n... outputs=[flow_processed_query_property],\n... )\n>>> inner_map_flow = Flow(\n... name=\"Map flow\",\n... start_node=start_node,\n... nodes=[start_node, llm_node, end_node],\n... control_flow_connections=[\n... ControlFlowEdge(name=\"start_to_llm\", from_node=start_node, to_node=llm_node),\n... ControlFlowEdge(name=\"llm_to_end\", from_node=llm_node, to_node=end_node),\n... ],\n... data_flow_connections=[\n... DataFlowEdge(\n... name=\"query_edge\",\n... source_node=start_node,\n... source_output=\"user_input\",\n... destination_node=llm_node,\n... destination_input=\"user_input\",\n... ),\n... DataFlowEdge(\n... name=\"search_results_edge\",\n... source_node=llm_node,\n... source_output=\"flow_processed_query\",\n... destination_node=end_node,\n... destination_input=\"flow_processed_query\"\n... ),\n... ],\n... )\n>>> user_query_property = Property(\n... json_schema={\"title\": \"user_query\", \"type\": \"string\"}\n... )\n>>> n_repeat_property = Property(\n... json_schema={\"title\": \"n_repeat\", \"type\": \"integer\"}\n... )\n>>> flow_iterable_queries_property = Property(\n... json_schema={\n... \"title\": \"iterated_user_input\",\n... \"type\": \"array\",\n... \"items\": {\"type\": \"string\"},\n... }\n... )\n>>> flow_processed_queries_property = Property(\n... json_schema={\n... \"title\": \"collected_flow_processed_query\",\n... \"type\": \"array\",\n... \"items\": {\"type\": \"string\"},\n... }\n... )\n>>> start_node = StartNode(name=\"start\", inputs=[user_query_property, n_repeat_property])\n>>> end_node = EndNode(name=\"end\", outputs=[flow_processed_query_property])\n>>> duplication_node = ToolNode(\n... name=\"duplication_tool node\",\n... tool=duplication_tool,\n... )\n>>> reduce_node = ToolNode(\n... name=\"reduce_tool node\",\n... tool=reduce_tool,\n... )\n>>> map_node = MapNode(\n... name=\"map node\",\n... subflow=inner_map_flow,\n... inputs=[flow_iterable_queries_property],\n... outputs=[flow_processed_queries_property],\n... )\n>>> mapreduce_flow = Flow(\n... name=\"Map-reduce flow\",\n... start_node=start_node,\n... nodes=[start_node, duplication_node, map_node, reduce_node, end_node],\n... control_flow_connections=[\n... ControlFlowEdge(\n... name=\"start_to_duplication\", from_node=start_node, to_node=duplication_node\n... ),\n... ControlFlowEdge(\n... name=\"duplication_to_map\", from_node=duplication_node, to_node=map_node\n... ),\n... ControlFlowEdge(name=\"map_to_reduce\", from_node=map_node, to_node=reduce_node),\n... ControlFlowEdge(name=\"reduce_to_end\", from_node=reduce_node, to_node=end_node),\n... ],\n... data_flow_connections=[\n... DataFlowEdge(\n... name=\"query_edge\",\n... source_node=start_node,\n... source_output=\"user_query\",\n... destination_node=duplication_node,\n... destination_input=\"element\",\n... ),\n... DataFlowEdge(\n... name=\"n_repeat_edge\",\n... source_node=start_node,\n... source_output=\"n_repeat\",\n... destination_node=duplication_node,\n... destination_input=\"n\",\n... ),\n... DataFlowEdge(\n... name=\"flow_iterables_edge\",\n... source_node=duplication_node,\n... source_output=\"flow_iterable_queries\",\n... destination_node=map_node,\n... destination_input=\"iterated_user_input\",\n... ),\n... DataFlowEdge(\n... name=\"flow_processed_queries_edge\",\n... source_node=map_node,\n... source_output=\"collected_flow_processed_query\",\n... destination_node=reduce_node,\n... destination_input=\"elements\",\n... ),\n... DataFlowEdge(\n... name=\"flow_processed_query_edge\",\n... source_node=reduce_node,\n... source_output=\"flow_processed_query\",\n... destination_node=end_node,\n... destination_input=\"flow_processed_query\",\n... ),\n... ],\n... )\n\nOnce the subflow is created we can simply integrate it with the ``FlowNode``:\n\n>>> from pyagentspec.flows.nodes import FlowNode, AgentNode\n>>> from pyagentspec.agent import Agent\n>>> start_node = StartNode(name=\"start\")\n>>> end_node = EndNode(name=\"end\", outputs=[flow_processed_query_property])\n>>> flow_node = FlowNode(name=\"flow node\", subflow=mapreduce_flow)\n>>> agent = Agent(\n... name=\"User interaction agent\",\n... llm_config=llm_config,\n... system_prompt=(\n... \"Your task is to gather from the user the query and the number of times \"\n... \"it should be asked to an LLM. Once you have this information, submit and exit.\"\n... ),\n... outputs=[user_query_property, n_repeat_property],\n... )\n>>> agent_node = AgentNode(name=\"flow node\", agent=agent)\n>>> flow = Flow(\n... name=\"Map flow\",\n... start_node=start_node,\n... nodes=[start_node, agent_node, flow_node, end_node],\n... control_flow_connections=[\n... ControlFlowEdge(name=\"start_to_agent\", from_node=start_node, to_node=agent_node),\n... ControlFlowEdge(name=\"agent_to_flow\", from_node=agent_node, to_node=flow_node),\n... ControlFlowEdge(name=\"flow_to_end\", from_node=flow_node, to_node=end_node),\n... ],\n... data_flow_connections=[\n... DataFlowEdge(\n... name=\"query_edge\",\n... source_node=agent_node,\n... source_output=\"user_query\",\n... destination_node=flow_node,\n... destination_input=\"user_query\",\n... ),\n... DataFlowEdge(\n... name=\"n_rep_edge\",\n... source_node=agent_node,\n... source_output=\"n_repeat\",\n... destination_node=flow_node,\n... destination_input=\"n_repeat\"\n... ),\n... DataFlowEdge(\n... name=\"n_rep_edge\",\n... source_node=flow_node,\n... source_output=\"flow_processed_query\",\n... destination_node=end_node,\n... destination_input=\"flow_processed_query\"\n... ),\n... ],\n... )", "properties": { "id": { "title": "Id", @@ -2921,6 +2921,11 @@ "subflow": { "$ref": "#/$defs/Flow" }, + "propagate_pending_input": { + "default": true, + "title": "Propagate Pending Input", + "type": "boolean" + }, "$referenced_components": { "$ref": "#/$defs/ReferencedComponents" }, diff --git a/docs/pyagentspec/source/agentspec/json_spec/agentspec_json_spec_26_2_0.json b/docs/pyagentspec/source/agentspec/json_spec/agentspec_json_spec_26_2_0.json index ea2b97ebe..3839913ea 100644 --- a/docs/pyagentspec/source/agentspec/json_spec/agentspec_json_spec_26_2_0.json +++ b/docs/pyagentspec/source/agentspec/json_spec/agentspec_json_spec_26_2_0.json @@ -2847,7 +2847,7 @@ }, "BaseFlowNode": { "additionalProperties": false, - "description": "The flow node executes a subflow as part of a flow.\n\n- **Inputs**\n Inferred from the inner structure. It's the sets of inputs\n required by the StartNode of the inner flow.\n- **Outputs**\n Inferred from the inner structure. It's the union of the\n sets of outputs exposed by the EndNodes of the inner flow.\n- **Branches**\n Inferred from the inner flow, one per each different value of the attribute\n ``branch_name`` of the nodes of type EndNode in the inner flow.\n\nExample\n-------\nThe ``FlowNode`` is particularly suitable when subflows can be reused inside a project.\nLet's see an example with a flow that estimates numerical value\nusing the \"wisdowm of the crowd\" effect:\n\n>>> from pyagentspec.property import Property\n>>> from pyagentspec.flows.flow import Flow\n>>> from pyagentspec.flows.edges import ControlFlowEdge, DataFlowEdge\n>>> from pyagentspec.flows.nodes import MapNode, LlmNode, ToolNode, StartNode, EndNode\n>>> from pyagentspec.tools import ServerTool\n>>> duplication_tool = ServerTool(\n... name=\"duplication_tool\",\n... description=\"\",\n... inputs=[\n... Property(\n... json_schema={\"title\": \"element\", \"description\": \"\", \"type\": \"string\"}\n... ),\n... Property(\n... json_schema={\"title\": \"n\", \"description\": \"\", \"type\": \"integer\"}\n... ),\n... ],\n... outputs=[\n... Property(\n... json_schema={\n... \"title\": \"flow_iterable_queries\",\n... \"type\": \"array\",\n... \"items\": {\"type\": \"string\"}\n... },\n... )\n... ],\n... )\n>>> reduce_tool = ServerTool(\n... name=\"reduce_tool\",\n... description=\"\",\n... inputs=[\n... Property(\n... json_schema={\"title\": \"elements\", \"type\": \"array\", \"items\": {\"type\": \"string\"}}\n... ),\n... ],\n... outputs=[Property(json_schema={\"title\": \"flow_processed_query\", \"type\": \"string\"})],\n... )\n>>> # Defining a simple prompt\n>>> REASONING_PROMPT_TEMPLATE = '''Provide your best numerical estimate for: {{user_input}}\n... Your answer should be a single number.\n... Do not include any units, reasoning, or extra text.'''\n>>> # Defining the subflow for the map step\n>>> user_input_property = Property(\n... json_schema={\"title\": \"user_input\", \"type\": \"string\"}\n... )\n>>> flow_processed_query_property = Property(\n... json_schema={\"title\": \"flow_processed_query\", \"type\": \"string\"}\n... )\n>>> start_node = StartNode(name=\"start\", inputs=[user_input_property])\n>>> end_node = EndNode(name=\"end\", outputs=[flow_processed_query_property])\n>>> llm_node = LlmNode(\n... name=\"reasoning llm node\",\n... llm_config=llm_config,\n... prompt_template=REASONING_PROMPT_TEMPLATE,\n... inputs=[user_input_property],\n... outputs=[flow_processed_query_property],\n... )\n>>> inner_map_flow = Flow(\n... name=\"Map flow\",\n... start_node=start_node,\n... nodes=[start_node, llm_node, end_node],\n... control_flow_connections=[\n... ControlFlowEdge(name=\"start_to_llm\", from_node=start_node, to_node=llm_node),\n... ControlFlowEdge(name=\"llm_to_end\", from_node=llm_node, to_node=end_node),\n... ],\n... data_flow_connections=[\n... DataFlowEdge(\n... name=\"query_edge\",\n... source_node=start_node,\n... source_output=\"user_input\",\n... destination_node=llm_node,\n... destination_input=\"user_input\",\n... ),\n... DataFlowEdge(\n... name=\"search_results_edge\",\n... source_node=llm_node,\n... source_output=\"flow_processed_query\",\n... destination_node=end_node,\n... destination_input=\"flow_processed_query\"\n... ),\n... ],\n... )\n>>> user_query_property = Property(\n... json_schema={\"title\": \"user_query\", \"type\": \"string\"}\n... )\n>>> n_repeat_property = Property(\n... json_schema={\"title\": \"n_repeat\", \"type\": \"integer\"}\n... )\n>>> flow_iterable_queries_property = Property(\n... json_schema={\n... \"title\": \"iterated_user_input\",\n... \"type\": \"array\",\n... \"items\": {\"type\": \"string\"},\n... }\n... )\n>>> flow_processed_queries_property = Property(\n... json_schema={\n... \"title\": \"collected_flow_processed_query\",\n... \"type\": \"array\",\n... \"items\": {\"type\": \"string\"},\n... }\n... )\n>>> start_node = StartNode(name=\"start\", inputs=[user_query_property, n_repeat_property])\n>>> end_node = EndNode(name=\"end\", outputs=[flow_processed_query_property])\n>>> duplication_node = ToolNode(\n... name=\"duplication_tool node\",\n... tool=duplication_tool,\n... )\n>>> reduce_node = ToolNode(\n... name=\"reduce_tool node\",\n... tool=reduce_tool,\n... )\n>>> map_node = MapNode(\n... name=\"map node\",\n... subflow=inner_map_flow,\n... inputs=[flow_iterable_queries_property],\n... outputs=[flow_processed_queries_property],\n... )\n>>> mapreduce_flow = Flow(\n... name=\"Map-reduce flow\",\n... start_node=start_node,\n... nodes=[start_node, duplication_node, map_node, reduce_node, end_node],\n... control_flow_connections=[\n... ControlFlowEdge(\n... name=\"start_to_duplication\", from_node=start_node, to_node=duplication_node\n... ),\n... ControlFlowEdge(\n... name=\"duplication_to_map\", from_node=duplication_node, to_node=map_node\n... ),\n... ControlFlowEdge(name=\"map_to_reduce\", from_node=map_node, to_node=reduce_node),\n... ControlFlowEdge(name=\"reduce_to_end\", from_node=reduce_node, to_node=end_node),\n... ],\n... data_flow_connections=[\n... DataFlowEdge(\n... name=\"query_edge\",\n... source_node=start_node,\n... source_output=\"user_query\",\n... destination_node=duplication_node,\n... destination_input=\"element\",\n... ),\n... DataFlowEdge(\n... name=\"n_repeat_edge\",\n... source_node=start_node,\n... source_output=\"n_repeat\",\n... destination_node=duplication_node,\n... destination_input=\"n\",\n... ),\n... DataFlowEdge(\n... name=\"flow_iterables_edge\",\n... source_node=duplication_node,\n... source_output=\"flow_iterable_queries\",\n... destination_node=map_node,\n... destination_input=\"iterated_user_input\",\n... ),\n... DataFlowEdge(\n... name=\"flow_processed_queries_edge\",\n... source_node=map_node,\n... source_output=\"collected_flow_processed_query\",\n... destination_node=reduce_node,\n... destination_input=\"elements\",\n... ),\n... DataFlowEdge(\n... name=\"flow_processed_query_edge\",\n... source_node=reduce_node,\n... source_output=\"flow_processed_query\",\n... destination_node=end_node,\n... destination_input=\"flow_processed_query\",\n... ),\n... ],\n... )\n\nOnce the subflow is created we can simply integrate it with the ``FlowNode``:\n\n>>> from pyagentspec.flows.nodes import FlowNode, AgentNode\n>>> from pyagentspec.agent import Agent\n>>> start_node = StartNode(name=\"start\")\n>>> end_node = EndNode(name=\"end\", outputs=[flow_processed_query_property])\n>>> flow_node = FlowNode(name=\"flow node\", subflow=mapreduce_flow)\n>>> agent = Agent(\n... name=\"User interaction agent\",\n... llm_config=llm_config,\n... system_prompt=(\n... \"Your task is to gather from the user the query and the number of times \"\n... \"it should be asked to an LLM. Once you have this information, submit and exit.\"\n... ),\n... outputs=[user_query_property, n_repeat_property],\n... )\n>>> agent_node = AgentNode(name=\"flow node\", agent=agent)\n>>> flow = Flow(\n... name=\"Map flow\",\n... start_node=start_node,\n... nodes=[start_node, agent_node, flow_node, end_node],\n... control_flow_connections=[\n... ControlFlowEdge(name=\"start_to_agent\", from_node=start_node, to_node=agent_node),\n... ControlFlowEdge(name=\"agent_to_flow\", from_node=agent_node, to_node=flow_node),\n... ControlFlowEdge(name=\"flow_to_end\", from_node=flow_node, to_node=end_node),\n... ],\n... data_flow_connections=[\n... DataFlowEdge(\n... name=\"query_edge\",\n... source_node=agent_node,\n... source_output=\"user_query\",\n... destination_node=flow_node,\n... destination_input=\"user_query\",\n... ),\n... DataFlowEdge(\n... name=\"n_rep_edge\",\n... source_node=agent_node,\n... source_output=\"n_repeat\",\n... destination_node=flow_node,\n... destination_input=\"n_repeat\"\n... ),\n... DataFlowEdge(\n... name=\"n_rep_edge\",\n... source_node=flow_node,\n... source_output=\"flow_processed_query\",\n... destination_node=end_node,\n... destination_input=\"flow_processed_query\"\n... ),\n... ],\n... )", + "description": "The flow node executes a subflow as part of a flow.\n\n- **Inputs**\n Inferred from the inner structure. It's the sets of inputs\n required by the StartNode of the inner flow.\n- **Outputs**\n Inferred from the inner structure. It's the union of the\n sets of outputs exposed by the EndNodes of the inner flow.\n- **Branches**\n Inferred from the inner flow, one per each different value of the attribute\n ``branch_name`` of the nodes of type EndNode in the inner flow.\n- **Pending input**\n If the subflow execution pauses to request input from a user or external caller,\n the pending input state is propagated through the ``FlowNode`` by default. This\n preserves the inline semantics of the subflow and lets the parent flow surface\n the child prompt and resume only after the child input is supplied.\n\nExample\n-------\nThe ``FlowNode`` is particularly suitable when subflows can be reused inside a project.\nLet's see an example with a flow that estimates numerical value\nusing the \"wisdowm of the crowd\" effect:\n\n>>> from pyagentspec.property import Property\n>>> from pyagentspec.flows.flow import Flow\n>>> from pyagentspec.flows.edges import ControlFlowEdge, DataFlowEdge\n>>> from pyagentspec.flows.nodes import MapNode, LlmNode, ToolNode, StartNode, EndNode\n>>> from pyagentspec.tools import ServerTool\n>>> duplication_tool = ServerTool(\n... name=\"duplication_tool\",\n... description=\"\",\n... inputs=[\n... Property(\n... json_schema={\"title\": \"element\", \"description\": \"\", \"type\": \"string\"}\n... ),\n... Property(\n... json_schema={\"title\": \"n\", \"description\": \"\", \"type\": \"integer\"}\n... ),\n... ],\n... outputs=[\n... Property(\n... json_schema={\n... \"title\": \"flow_iterable_queries\",\n... \"type\": \"array\",\n... \"items\": {\"type\": \"string\"}\n... },\n... )\n... ],\n... )\n>>> reduce_tool = ServerTool(\n... name=\"reduce_tool\",\n... description=\"\",\n... inputs=[\n... Property(\n... json_schema={\"title\": \"elements\", \"type\": \"array\", \"items\": {\"type\": \"string\"}}\n... ),\n... ],\n... outputs=[Property(json_schema={\"title\": \"flow_processed_query\", \"type\": \"string\"})],\n... )\n>>> # Defining a simple prompt\n>>> REASONING_PROMPT_TEMPLATE = '''Provide your best numerical estimate for: {{user_input}}\n... Your answer should be a single number.\n... Do not include any units, reasoning, or extra text.'''\n>>> # Defining the subflow for the map step\n>>> user_input_property = Property(\n... json_schema={\"title\": \"user_input\", \"type\": \"string\"}\n... )\n>>> flow_processed_query_property = Property(\n... json_schema={\"title\": \"flow_processed_query\", \"type\": \"string\"}\n... )\n>>> start_node = StartNode(name=\"start\", inputs=[user_input_property])\n>>> end_node = EndNode(name=\"end\", outputs=[flow_processed_query_property])\n>>> llm_node = LlmNode(\n... name=\"reasoning llm node\",\n... llm_config=llm_config,\n... prompt_template=REASONING_PROMPT_TEMPLATE,\n... inputs=[user_input_property],\n... outputs=[flow_processed_query_property],\n... )\n>>> inner_map_flow = Flow(\n... name=\"Map flow\",\n... start_node=start_node,\n... nodes=[start_node, llm_node, end_node],\n... control_flow_connections=[\n... ControlFlowEdge(name=\"start_to_llm\", from_node=start_node, to_node=llm_node),\n... ControlFlowEdge(name=\"llm_to_end\", from_node=llm_node, to_node=end_node),\n... ],\n... data_flow_connections=[\n... DataFlowEdge(\n... name=\"query_edge\",\n... source_node=start_node,\n... source_output=\"user_input\",\n... destination_node=llm_node,\n... destination_input=\"user_input\",\n... ),\n... DataFlowEdge(\n... name=\"search_results_edge\",\n... source_node=llm_node,\n... source_output=\"flow_processed_query\",\n... destination_node=end_node,\n... destination_input=\"flow_processed_query\"\n... ),\n... ],\n... )\n>>> user_query_property = Property(\n... json_schema={\"title\": \"user_query\", \"type\": \"string\"}\n... )\n>>> n_repeat_property = Property(\n... json_schema={\"title\": \"n_repeat\", \"type\": \"integer\"}\n... )\n>>> flow_iterable_queries_property = Property(\n... json_schema={\n... \"title\": \"iterated_user_input\",\n... \"type\": \"array\",\n... \"items\": {\"type\": \"string\"},\n... }\n... )\n>>> flow_processed_queries_property = Property(\n... json_schema={\n... \"title\": \"collected_flow_processed_query\",\n... \"type\": \"array\",\n... \"items\": {\"type\": \"string\"},\n... }\n... )\n>>> start_node = StartNode(name=\"start\", inputs=[user_query_property, n_repeat_property])\n>>> end_node = EndNode(name=\"end\", outputs=[flow_processed_query_property])\n>>> duplication_node = ToolNode(\n... name=\"duplication_tool node\",\n... tool=duplication_tool,\n... )\n>>> reduce_node = ToolNode(\n... name=\"reduce_tool node\",\n... tool=reduce_tool,\n... )\n>>> map_node = MapNode(\n... name=\"map node\",\n... subflow=inner_map_flow,\n... inputs=[flow_iterable_queries_property],\n... outputs=[flow_processed_queries_property],\n... )\n>>> mapreduce_flow = Flow(\n... name=\"Map-reduce flow\",\n... start_node=start_node,\n... nodes=[start_node, duplication_node, map_node, reduce_node, end_node],\n... control_flow_connections=[\n... ControlFlowEdge(\n... name=\"start_to_duplication\", from_node=start_node, to_node=duplication_node\n... ),\n... ControlFlowEdge(\n... name=\"duplication_to_map\", from_node=duplication_node, to_node=map_node\n... ),\n... ControlFlowEdge(name=\"map_to_reduce\", from_node=map_node, to_node=reduce_node),\n... ControlFlowEdge(name=\"reduce_to_end\", from_node=reduce_node, to_node=end_node),\n... ],\n... data_flow_connections=[\n... DataFlowEdge(\n... name=\"query_edge\",\n... source_node=start_node,\n... source_output=\"user_query\",\n... destination_node=duplication_node,\n... destination_input=\"element\",\n... ),\n... DataFlowEdge(\n... name=\"n_repeat_edge\",\n... source_node=start_node,\n... source_output=\"n_repeat\",\n... destination_node=duplication_node,\n... destination_input=\"n\",\n... ),\n... DataFlowEdge(\n... name=\"flow_iterables_edge\",\n... source_node=duplication_node,\n... source_output=\"flow_iterable_queries\",\n... destination_node=map_node,\n... destination_input=\"iterated_user_input\",\n... ),\n... DataFlowEdge(\n... name=\"flow_processed_queries_edge\",\n... source_node=map_node,\n... source_output=\"collected_flow_processed_query\",\n... destination_node=reduce_node,\n... destination_input=\"elements\",\n... ),\n... DataFlowEdge(\n... name=\"flow_processed_query_edge\",\n... source_node=reduce_node,\n... source_output=\"flow_processed_query\",\n... destination_node=end_node,\n... destination_input=\"flow_processed_query\",\n... ),\n... ],\n... )\n\nOnce the subflow is created we can simply integrate it with the ``FlowNode``:\n\n>>> from pyagentspec.flows.nodes import FlowNode, AgentNode\n>>> from pyagentspec.agent import Agent\n>>> start_node = StartNode(name=\"start\")\n>>> end_node = EndNode(name=\"end\", outputs=[flow_processed_query_property])\n>>> flow_node = FlowNode(name=\"flow node\", subflow=mapreduce_flow)\n>>> agent = Agent(\n... name=\"User interaction agent\",\n... llm_config=llm_config,\n... system_prompt=(\n... \"Your task is to gather from the user the query and the number of times \"\n... \"it should be asked to an LLM. Once you have this information, submit and exit.\"\n... ),\n... outputs=[user_query_property, n_repeat_property],\n... )\n>>> agent_node = AgentNode(name=\"flow node\", agent=agent)\n>>> flow = Flow(\n... name=\"Map flow\",\n... start_node=start_node,\n... nodes=[start_node, agent_node, flow_node, end_node],\n... control_flow_connections=[\n... ControlFlowEdge(name=\"start_to_agent\", from_node=start_node, to_node=agent_node),\n... ControlFlowEdge(name=\"agent_to_flow\", from_node=agent_node, to_node=flow_node),\n... ControlFlowEdge(name=\"flow_to_end\", from_node=flow_node, to_node=end_node),\n... ],\n... data_flow_connections=[\n... DataFlowEdge(\n... name=\"query_edge\",\n... source_node=agent_node,\n... source_output=\"user_query\",\n... destination_node=flow_node,\n... destination_input=\"user_query\",\n... ),\n... DataFlowEdge(\n... name=\"n_rep_edge\",\n... source_node=agent_node,\n... source_output=\"n_repeat\",\n... destination_node=flow_node,\n... destination_input=\"n_repeat\"\n... ),\n... DataFlowEdge(\n... name=\"n_rep_edge\",\n... source_node=flow_node,\n... source_output=\"flow_processed_query\",\n... destination_node=end_node,\n... destination_input=\"flow_processed_query\"\n... ),\n... ],\n... )", "properties": { "id": { "title": "Id", @@ -2921,6 +2921,11 @@ "subflow": { "$ref": "#/$defs/Flow" }, + "propagate_pending_input": { + "default": true, + "title": "Propagate Pending Input", + "type": "boolean" + }, "$referenced_components": { "$ref": "#/$defs/ReferencedComponents" }, diff --git a/docs/pyagentspec/source/agentspec/language_spec_nightly.rst b/docs/pyagentspec/source/agentspec/language_spec_nightly.rst index 2fb4ef912..b7e607bda 100644 --- a/docs/pyagentspec/source/agentspec/language_spec_nightly.rst +++ b/docs/pyagentspec/source/agentspec/language_spec_nightly.rst @@ -1742,6 +1742,9 @@ A more detailed description of each node follows. * The flow is started giving the specified inputs * The flow provides the specified outputs * The flow is run as it was inlined with the overall flow + * If the child flow enters an ``input-required`` state, that pending input + is propagated through the ``FlowNode`` by default so the parent flow also + waits for the same input before continuing. - .. list-table:: :header-rows: 1 :widths: 20 35 15 15 15 @@ -1757,6 +1760,15 @@ A more detailed description of each node follows. - Flow - Yes - - + * - propagate_pending_input + - Whether pending input requests raised by the subflow propagate through the + ``FlowNode``. When ``true``, the parent flow must surface the child prompt + and resume only after the child input is supplied. When ``false``, a runtime + may keep the pending input local to the child execution, but must fail + explicitly if it cannot preserve that isolation. + - bool + - No + - true - Inferred from the inner structure. It's the sets of inputs required by the StartNode of the inner flow diff --git a/docs/pyagentspec/source/changelog.rst b/docs/pyagentspec/source/changelog.rst index cb4e7eac8..2f648a807 100644 --- a/docs/pyagentspec/source/changelog.rst +++ b/docs/pyagentspec/source/changelog.rst @@ -7,6 +7,13 @@ Agent Spec |release| Improvements ^^^^^^^^^^^^ +* **FlowNode pending input propagation** + + ``FlowNode`` now exposes ``propagate_pending_input`` to make nested + ``input-required`` behavior explicit. By default, pending input raised inside + a child flow is surfaced through the parent ``FlowNode`` so parent flows do not + continue or hang while a child is waiting for input. + * **LangGraph adapter timeout and retry improvements** The LangGraph adapter now applies ``RetryPolicy.request_timeout`` and diff --git a/pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py b/pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py index dab0a0c77..3003a6103 100644 --- a/pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py +++ b/pyagentspec/src/pyagentspec/adapters/langgraph/_node_execution.py @@ -21,6 +21,7 @@ BaseChatModel, BaseMessage, Checkpointer, + Command, CompiledStateGraph, ExecuteOutput, FlowStateSchema, @@ -768,18 +769,76 @@ def __init__( self.subflow = subflow self.config = config + def _disabled_pending_input_error(self) -> RuntimeError: + return RuntimeError( + f"FlowNode `{self.node.name}` received a pending input request from its " + "subflow, but propagate_pending_input is disabled." + ) + + def _invoke_subflow(self, inputs: Dict[str, Any], messages: Messages) -> Dict[str, Any]: + from langgraph.errors import GraphInterrupt + + try: + return self.subflow.invoke({"messages": messages, "inputs": inputs}, self.config) + except GraphInterrupt as e: + if self.node.propagate_pending_input: + raise + raise self._disabled_pending_input_error() from e + + async def _ainvoke_subflow(self, inputs: Dict[str, Any], messages: Messages) -> Dict[str, Any]: + from langgraph.errors import GraphInterrupt + + try: + return await self.subflow.ainvoke( + {"messages": messages, "inputs": inputs}, self.config + ) + except GraphInterrupt as e: + if self.node.propagate_pending_input: + raise + raise self._disabled_pending_input_error() from e + + def _resume_or_raise_for_nested_interrupt(self, flow_output: Dict[str, Any]) -> Any: + if not self.node.propagate_pending_input: + raise self._disabled_pending_input_error() + nested_interrupts = flow_output["__interrupt__"] + if ( + isinstance(nested_interrupts, (list, tuple)) + and len(nested_interrupts) == 1 + and hasattr(nested_interrupts[0], "value") + ): + return interrupt(nested_interrupts[0].value) + return interrupt(nested_interrupts) + + def _get_generated_messages( + self, initial_messages: Messages, flow_output: Dict[str, Any] + ) -> Messages: + flow_messages = flow_output.get("messages", []) + if not isinstance(flow_messages, list): + return [] + if isinstance(initial_messages, list) and flow_messages[: len(initial_messages)] == list( + initial_messages + ): + return flow_messages[len(initial_messages) :] + return flow_messages + def _execute(self, inputs: Dict[str, Any], messages: Messages) -> ExecuteOutput: - flow_output = self.subflow.invoke({"messages": messages, "inputs": inputs}, self.config) + flow_output = self._invoke_subflow(inputs, messages) + while "__interrupt__" in flow_output: + resume_value = self._resume_or_raise_for_nested_interrupt(flow_output) + flow_output = self.subflow.invoke(Command(resume=resume_value), self.config) return flow_output["outputs"], NodeExecutionDetails( - branch=flow_output["node_execution_details"]["branch"] + branch=flow_output["node_execution_details"]["branch"], + generated_messages=self._get_generated_messages(messages, flow_output), ) async def _aexecute(self, inputs: Dict[str, Any], messages: Messages) -> ExecuteOutput: - flow_output = await self.subflow.ainvoke( - {"messages": messages, "inputs": inputs}, self.config - ) + flow_output = await self._ainvoke_subflow(inputs, messages) + while "__interrupt__" in flow_output: + resume_value = self._resume_or_raise_for_nested_interrupt(flow_output) + flow_output = await self.subflow.ainvoke(Command(resume=resume_value), self.config) return flow_output["outputs"], NodeExecutionDetails( - branch=flow_output["node_execution_details"]["branch"] + branch=flow_output["node_execution_details"]["branch"], + generated_messages=self._get_generated_messages(messages, flow_output), ) diff --git a/pyagentspec/src/pyagentspec/adapters/langgraph/_types.py b/pyagentspec/src/pyagentspec/adapters/langgraph/_types.py index 1b44d0ad1..48c4f6c53 100644 --- a/pyagentspec/src/pyagentspec/adapters/langgraph/_types.py +++ b/pyagentspec/src/pyagentspec/adapters/langgraph/_types.py @@ -31,7 +31,7 @@ from langgraph.graph._node import StateNodeSpec from langgraph.graph.message import Messages from langgraph.graph.state import CompiledStateGraph - from langgraph.types import Checkpointer, interrupt + from langgraph.types import Checkpointer, Command, interrupt else: langgraph_swarm = LazyLoader("langgraph_swarm") @@ -42,6 +42,7 @@ BaseTool = LazyType("langchain_core.tools", "BaseTool") StructuredTool = LazyType("langchain_core.tools", "StructuredTool") Checkpointer = LazyType("langgraph.types", "Checkpointer") + Command = LazyType("langgraph.types", "Command") interrupt = LazyLoader("langgraph.types", "interrupt") StateGraph = LazyType("langgraph.graph", "StateGraph") Messages = LazyLoader("langgraph.graph.message", "Messages") @@ -137,6 +138,7 @@ class FlowOutputSchema(TypedDict): "BaseChatModel", "AgentState", "Checkpointer", + "Command", "interrupt", "RunnableConfig", "Messages", diff --git a/pyagentspec/src/pyagentspec/flows/nodes/flownode.py b/pyagentspec/src/pyagentspec/flows/nodes/flownode.py index c35127b52..c077155ab 100644 --- a/pyagentspec/src/pyagentspec/flows/nodes/flownode.py +++ b/pyagentspec/src/pyagentspec/flows/nodes/flownode.py @@ -12,6 +12,7 @@ from pyagentspec.flows.node import Node from pyagentspec.flows.nodes.endnode import EndNode from pyagentspec.property import Property +from pyagentspec.versioning import AgentSpecVersionEnum class FlowNode(Node): @@ -26,6 +27,11 @@ class FlowNode(Node): - **Branches** Inferred from the inner flow, one per each different value of the attribute ``branch_name`` of the nodes of type EndNode in the inner flow. + - **Pending input** + If the subflow execution pauses to request input from a user or external caller, + the pending input state is propagated through the ``FlowNode`` by default. This + preserves the inline semantics of the subflow and lets the parent flow surface + the child prompt and resume only after the child input is supplied. Example ------- @@ -259,6 +265,23 @@ class FlowNode(Node): subflow: Flow """The flow that should be executed""" + propagate_pending_input: bool = True + """Whether pending input requests raised inside the subflow propagate through this FlowNode.""" + + def _versioned_model_fields_to_exclude( + self, agentspec_version: AgentSpecVersionEnum + ) -> set[str]: + fields_to_exclude = super()._versioned_model_fields_to_exclude(agentspec_version) + if agentspec_version < AgentSpecVersionEnum.v26_2_0: + fields_to_exclude.add("propagate_pending_input") + return fields_to_exclude + + def _infer_min_agentspec_version_from_configuration(self) -> AgentSpecVersionEnum: + min_version = super()._infer_min_agentspec_version_from_configuration() + if not self.propagate_pending_input: + min_version = max(min_version, AgentSpecVersionEnum.v26_2_0) + return min_version + def _get_inferred_branches(self) -> List[str]: if hasattr(self, "subflow"): end_nodes = sorted( diff --git a/pyagentspec/tests/adapters/langgraph/flows/test_flownode.py b/pyagentspec/tests/adapters/langgraph/flows/test_flownode.py index 200d3ddd6..0007399a2 100644 --- a/pyagentspec/tests/adapters/langgraph/flows/test_flownode.py +++ b/pyagentspec/tests/adapters/langgraph/flows/test_flownode.py @@ -8,7 +8,7 @@ from pyagentspec.flows.edges import ControlFlowEdge, DataFlowEdge from pyagentspec.flows.flow import Flow -from pyagentspec.flows.nodes import EndNode, FlowNode, StartNode +from pyagentspec.flows.nodes import EndNode, FlowNode, InputMessageNode, StartNode from pyagentspec.property import StringProperty @@ -88,6 +88,109 @@ def test_flownode_can_be_imported_and_executed(flow_node_flow: Flow) -> None: assert outputs["custom_prop"] == "custom" +def _flow_with_nested_input_node(*, propagate_pending_input: bool = True) -> Flow: + custom_property = StringProperty(title="custom_input") + input_message_node = InputMessageNode( + name="input_message", + outputs=[custom_property], + ) + subflow_start_node = StartNode(name="subflow_start") + subflow_end_node = EndNode(name="subflow_end", outputs=[custom_property]) + subflow = Flow( + name="subflow", + start_node=subflow_start_node, + nodes=[subflow_start_node, input_message_node, subflow_end_node], + control_flow_connections=[ + ControlFlowEdge( + name="subflow_start_to_input", + from_node=subflow_start_node, + to_node=input_message_node, + ), + ControlFlowEdge( + name="input_to_subflow_end", + from_node=input_message_node, + to_node=subflow_end_node, + ), + ], + data_flow_connections=[ + DataFlowEdge( + name="input_to_subflow_end", + source_node=input_message_node, + source_output=custom_property.title, + destination_node=subflow_end_node, + destination_input=custom_property.title, + ), + ], + outputs=[custom_property], + ) + + flow_node = FlowNode( + name="flow_node", + subflow=subflow, + propagate_pending_input=propagate_pending_input, + ) + start_node = StartNode(name="start") + end_node = EndNode(name="end", outputs=[custom_property]) + return Flow( + name="flow", + start_node=start_node, + nodes=[start_node, flow_node, end_node], + control_flow_connections=[ + ControlFlowEdge(name="start_to_flow_node", from_node=start_node, to_node=flow_node), + ControlFlowEdge(name="flow_node_to_end", from_node=flow_node, to_node=end_node), + ], + data_flow_connections=[ + DataFlowEdge( + name="flow_node_to_end", + source_node=flow_node, + source_output=custom_property.title, + destination_node=end_node, + destination_input=custom_property.title, + ), + ], + outputs=[custom_property], + ) + + +def test_flownode_propagates_nested_input_required_state() -> None: + from langchain_core.messages import HumanMessage + from langchain_core.runnables import RunnableConfig + from langgraph.checkpoint.memory import MemorySaver + from langgraph.types import Command + + from pyagentspec.adapters.langgraph import AgentSpecLoader + + flow = _flow_with_nested_input_node() + agent = AgentSpecLoader(checkpointer=MemorySaver()).load_component(flow) + + config = RunnableConfig({"configurable": {"thread_id": "nested-flow-node-input"}}) + result = agent.invoke({}, config=config) + assert "__interrupt__" in result + assert result["__interrupt__"][0].value == "" + + result = agent.invoke(Command(resume="nested value"), config=config) + + assert "outputs" in result + assert result["outputs"]["custom_input"] == "nested value" + assert len(result["messages"]) == 1 + assert isinstance(result["messages"][0], HumanMessage) + assert result["messages"][0].content == "nested value" + + +def test_flownode_disabled_pending_input_propagation_raises() -> None: + from langchain_core.runnables import RunnableConfig + from langgraph.checkpoint.memory import MemorySaver + + from pyagentspec.adapters.langgraph import AgentSpecLoader + + flow = _flow_with_nested_input_node(propagate_pending_input=False) + agent = AgentSpecLoader(checkpointer=MemorySaver()).load_component(flow) + + config = RunnableConfig({"configurable": {"thread_id": "nested-flow-node-input-disabled"}}) + with pytest.raises(RuntimeError, match="propagate_pending_input is disabled"): + agent.invoke({}, config=config) + + @pytest.mark.anyio async def test_flownode_can_be_executed_async(flow_node_flow: Flow) -> None: from pyagentspec.adapters.langgraph import AgentSpecLoader diff --git a/pyagentspec/tests/serialization/test_flow_node.py b/pyagentspec/tests/serialization/test_flow_node.py new file mode 100644 index 000000000..cd7c4fe97 --- /dev/null +++ b/pyagentspec/tests/serialization/test_flow_node.py @@ -0,0 +1,67 @@ +# Copyright © 2025 Oracle and/or its affiliates. +# +# This software is under the Apache License 2.0 +# (LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0) or Universal Permissive License +# (UPL) 1.0 (LICENSE-UPL or https://oss.oracle.com/licenses/upl), at your option. + +import pytest + +from pyagentspec.flows.flow import Flow +from pyagentspec.flows.nodes.flownode import FlowNode +from pyagentspec.serialization import AgentSpecDeserializer, AgentSpecSerializer +from pyagentspec.versioning import AgentSpecVersionEnum + + +def test_flow_node_propagates_pending_input_by_default(simplest_flow: Flow) -> None: + flow_node = FlowNode(name="flow_node", subflow=simplest_flow) + + assert flow_node.propagate_pending_input is True + + +def test_flow_node_serializes_pending_input_propagation_in_26_2_0( + simplest_flow: Flow, +) -> None: + flow_node = FlowNode(name="flow_node", subflow=simplest_flow) + + serialized = AgentSpecSerializer().to_dict( + flow_node, + agentspec_version=AgentSpecVersionEnum.v26_2_0, + ) + + assert serialized["propagate_pending_input"] is True + + +def test_flow_node_excludes_default_pending_input_propagation_before_26_2_0( + simplest_flow: Flow, +) -> None: + flow_node = FlowNode(name="flow_node", subflow=simplest_flow) + + serialized = AgentSpecSerializer().to_dict( + flow_node, + agentspec_version=AgentSpecVersionEnum.v26_1_2, + ) + + assert "propagate_pending_input" not in serialized + + +def test_flow_node_can_disable_pending_input_propagation(simplest_flow: Flow) -> None: + flow_node = FlowNode( + name="flow_node", + subflow=simplest_flow, + propagate_pending_input=False, + ) + + assert flow_node.min_agentspec_version == AgentSpecVersionEnum.v26_2_0 + with pytest.raises(ValueError, match="Invalid agentspec_version"): + AgentSpecSerializer().to_dict( + flow_node, + agentspec_version=AgentSpecVersionEnum.v26_1_2, + ) + + serialized = AgentSpecSerializer().to_dict(flow_node) + assert serialized["agentspec_version"] == AgentSpecVersionEnum.v26_2_0.value + assert serialized["propagate_pending_input"] is False + + deserialized = AgentSpecDeserializer().from_dict(serialized) + assert isinstance(deserialized, FlowNode) + assert deserialized.propagate_pending_input is False diff --git a/tsagentspec/src/flows/nodes/flow-node.ts b/tsagentspec/src/flows/nodes/flow-node.ts index b6f1543f5..1fddc5686 100644 --- a/tsagentspec/src/flows/nodes/flow-node.ts +++ b/tsagentspec/src/flows/nodes/flow-node.ts @@ -10,6 +10,7 @@ import { LazyFlowRef } from "../lazy-schemas.js"; export const FlowNodeSchema = NodeBaseSchema.extend({ componentType: z.literal("FlowNode"), subflow: LazyFlowRef, + propagatePendingInput: z.boolean().default(true), }); export type FlowNode = z.infer; @@ -22,6 +23,7 @@ export function createFlowNode(opts: { metadata?: Record; inputs?: Property[]; outputs?: Property[]; + propagatePendingInput?: boolean; }): FlowNode { const subflow = opts.subflow; const inputs = diff --git a/tsagentspec/src/serialization/version-gates.ts b/tsagentspec/src/serialization/version-gates.ts index 3798c48cf..887862b63 100644 --- a/tsagentspec/src/serialization/version-gates.ts +++ b/tsagentspec/src/serialization/version-gates.ts @@ -36,6 +36,9 @@ export const VERSION_GATED_FIELDS = { CatchExceptionNode: { _self: AgentSpecVersion.V26_2_0, }, + FlowNode: { + propagatePendingInput: AgentSpecVersion.V26_2_0, + }, ParallelMapNode: { _self: AgentSpecVersion.V25_4_2, }, diff --git a/tsagentspec/tests/flows/nodes.test.ts b/tsagentspec/tests/flows/nodes.test.ts index 9d7f9f822..2f7cc5f2e 100644 --- a/tsagentspec/tests/flows/nodes.test.ts +++ b/tsagentspec/tests/flows/nodes.test.ts @@ -266,6 +266,22 @@ describe("FlowNode", () => { expect(node.outputs!.map((o) => o.title)).toContain("result"); }); + it("should propagate pending subflow input by default", () => { + const flow = makeSimpleFlow(); + const node = createFlowNode({ name: "flow-node", subflow: flow }); + expect(node.propagatePendingInput).toBe(true); + }); + + it("should allow pending subflow input propagation to be disabled", () => { + const flow = makeSimpleFlow(); + const node = createFlowNode({ + name: "flow-node", + subflow: flow, + propagatePendingInput: false, + }); + expect(node.propagatePendingInput).toBe(false); + }); + it("should infer branches from subflow EndNode branchNames", () => { const start = createStartNode({ name: "start" }); const branching = createBranchingNode({ diff --git a/tsagentspec/tests/serialization/version-gates.test.ts b/tsagentspec/tests/serialization/version-gates.test.ts index 603e5d01a..8db5bb357 100644 --- a/tsagentspec/tests/serialization/version-gates.test.ts +++ b/tsagentspec/tests/serialization/version-gates.test.ts @@ -8,6 +8,11 @@ import { createBuiltinTool, createMCPToolBox, createStdioTransport, + createControlFlowEdge, + createEndNode, + createFlow, + createFlowNode, + createStartNode, stringProperty, } from "../../src/index.js"; @@ -19,6 +24,19 @@ function makeLlmConfig() { }); } +function makeFlow() { + const start = createStartNode({ name: "start" }); + const end = createEndNode({ name: "end" }); + return createFlow({ + name: "flow", + startNode: start, + nodes: [start, end], + controlFlowConnections: [ + createControlFlowEdge({ name: "start-end", fromNode: start, toNode: end }), + ], + }); +} + describe("version-gated field serialization", () => { it("should exclude humanInTheLoop for versions before 25.4.2", () => { const serializer = new AgentSpecSerializer(); @@ -166,6 +184,26 @@ describe("version-gated field serialization", () => { expect("requires_confirmation" in toolboxes[0]!).toBe(true); }); + it("should version-gate FlowNode propagatePendingInput before 26.2.0", () => { + const serializer = new AgentSpecSerializer(); + const node = createFlowNode({ + name: "flow-node", + subflow: makeFlow(), + propagatePendingInput: false, + }); + + const oldJson = serializer.toJson(node, { + agentspecVersion: AgentSpecVersion.V26_1_0, + }) as string; + expect("propagate_pending_input" in JSON.parse(oldJson)).toBe(false); + + const newJson = serializer.toJson(node, { + agentspecVersion: AgentSpecVersion.V26_2_0, + }) as string; + const newDict = JSON.parse(newJson); + expect(newDict["propagate_pending_input"]).toBe(false); + }); + it("should include everything for current version", () => { const serializer = new AgentSpecSerializer(); const tool = createServerTool({