diff --git a/trustgraph-base/trustgraph/api/flow.py b/trustgraph-base/trustgraph/api/flow.py index d1d5f95e..0214a4bd 100644 --- a/trustgraph-base/trustgraph/api/flow.py +++ b/trustgraph-base/trustgraph/api/flow.py @@ -87,7 +87,7 @@ class Flow: return json.loads(self.request(request = input)["flow"]) - def start(self, class_name, id, description): + def start(self, class_name, id, description, parameters=None): # The input consists of system and prompt strings input = { @@ -97,6 +97,9 @@ class Flow: "description": description, } + if parameters: + input["parameters"] = parameters + self.request(request = input) def stop(self, id): diff --git a/trustgraph-base/trustgraph/messaging/translators/flow.py b/trustgraph-base/trustgraph/messaging/translators/flow.py index f05767c8..8c1a019a 100644 --- a/trustgraph-base/trustgraph/messaging/translators/flow.py +++ b/trustgraph-base/trustgraph/messaging/translators/flow.py @@ -12,12 +12,13 @@ class FlowRequestTranslator(MessageTranslator): class_name=data.get("class-name"), class_definition=data.get("class-definition"), description=data.get("description"), - flow_id=data.get("flow-id") + flow_id=data.get("flow-id"), + parameters=data.get("parameters") ) def from_pulsar(self, obj: FlowRequest) -> Dict[str, Any]: result = {} - + if obj.operation is not None: result["operation"] = obj.operation if obj.class_name is not None: @@ -28,7 +29,9 @@ class FlowRequestTranslator(MessageTranslator): result["description"] = obj.description if obj.flow_id is not None: result["flow-id"] = obj.flow_id - + if obj.parameters is not None: + result["parameters"] = obj.parameters + return result @@ -40,7 +43,7 @@ class FlowResponseTranslator(MessageTranslator): def from_pulsar(self, obj: FlowResponse) -> Dict[str, Any]: result = {} - + if obj.class_names is not None: result["class-names"] = obj.class_names if obj.flow_ids is not None: @@ -51,7 +54,9 @@ class FlowResponseTranslator(MessageTranslator): result["flow"] = obj.flow if obj.description is not None: result["description"] = obj.description - + if obj.parameters is not None: + result["parameters"] = obj.parameters + return result def from_response_with_completion(self, obj: FlowResponse) -> Tuple[Dict[str, Any], bool]: diff --git a/trustgraph-base/trustgraph/schema/services/flow.py b/trustgraph-base/trustgraph/schema/services/flow.py index 0b5c1bfd..d03e559b 100644 --- a/trustgraph-base/trustgraph/schema/services/flow.py +++ b/trustgraph-base/trustgraph/schema/services/flow.py @@ -35,6 +35,9 @@ class FlowRequest(Record): # get_flow, start_flow, stop_flow flow_id = String() + # start_flow - optional parameters for flow customization + parameters = Map(String()) + class FlowResponse(Record): # list_classes @@ -52,6 +55,9 @@ class FlowResponse(Record): # get_flow description = String() + # get_flow - parameters used when flow was started + parameters = Map(String()) + # Everything error = Error() diff --git a/trustgraph-cli/trustgraph/cli/show_flows.py b/trustgraph-cli/trustgraph/cli/show_flows.py index a405d830..06fbfbe8 100644 --- a/trustgraph-cli/trustgraph/cli/show_flows.py +++ b/trustgraph-cli/trustgraph/cli/show_flows.py @@ -74,6 +74,13 @@ def show_flows(url): table.append(("id", id)) table.append(("class", flow.get("class-name", ""))) table.append(("desc", flow.get("description", ""))) + + # Display parameters if they exist + parameters = flow.get("parameters", {}) + if parameters: + param_str = json.dumps(parameters, indent=2) + table.append(("parameters", param_str)) + table.append(("queue", describe_interfaces(interface_defs, flow))) print(tabulate.tabulate( diff --git a/trustgraph-cli/trustgraph/cli/start_flow.py b/trustgraph-cli/trustgraph/cli/start_flow.py index 36048474..e8c1822c 100644 --- a/trustgraph-cli/trustgraph/cli/start_flow.py +++ b/trustgraph-cli/trustgraph/cli/start_flow.py @@ -10,7 +10,7 @@ import json default_url = os.getenv("TRUSTGRAPH_URL", 'http://localhost:8088/') -def start_flow(url, class_name, flow_id, description): +def start_flow(url, class_name, flow_id, description, parameters=None): api = Api(url).flow() @@ -18,6 +18,7 @@ def start_flow(url, class_name, flow_id, description): class_name = class_name, id = flow_id, description = description, + parameters = parameters, ) def main(): @@ -51,15 +52,34 @@ def main(): help=f'Flow description', ) + parser.add_argument( + '-p', '--parameters', + help=f'Flow parameters as JSON string (e.g., \'{"model": "gpt-4", "temp": 0.7}\')', + ) + + parser.add_argument( + '--parameters-file', + help=f'Path to JSON file containing flow parameters', + ) + args = parser.parse_args() try: + # Parse parameters from command line arguments + parameters = None + + if args.parameters_file: + with open(args.parameters_file, 'r') as f: + parameters = json.load(f) + elif args.parameters: + parameters = json.loads(args.parameters) start_flow( url = args.api_url, class_name = args.class_name, flow_id = args.flow_id, description = args.description, + parameters = parameters, ) except Exception as e: diff --git a/trustgraph-flow/trustgraph/config/service/flow.py b/trustgraph-flow/trustgraph/config/service/flow.py index 3e83f8fa..5789ef0d 100644 --- a/trustgraph-flow/trustgraph/config/service/flow.py +++ b/trustgraph-flow/trustgraph/config/service/flow.py @@ -68,11 +68,14 @@ class FlowConfig: async def handle_get_flow(self, msg): - flow = await self.config.get("flows").get(msg.flow_id) + flow_data = await self.config.get("flows").get(msg.flow_id) + flow = json.loads(flow_data) return FlowResponse( error = None, - flow = flow, + flow = flow_data, + description = flow.get("description", ""), + parameters = flow.get("parameters", {}), ) async def handle_start_flow(self, msg): @@ -92,16 +95,24 @@ class FlowConfig: if msg.class_name not in await self.config.get("flow-classes").values(): raise RuntimeError("Class does not exist") - def repl_template(tmp): - return tmp.replace( + cls = json.loads( + await self.config.get("flow-classes").get(msg.class_name) + ) + + # Get parameters from message (default to empty dict if not provided) + parameters = msg.parameters if msg.parameters else {} + + # Apply parameter substitution to template replacement function + def repl_template_with_params(tmp): + result = tmp.replace( "{class}", msg.class_name ).replace( "{id}", msg.flow_id ) - - cls = json.loads( - await self.config.get("flow-classes").get(msg.class_name) - ) + # Apply parameter substitutions + for param_name, param_value in parameters.items(): + result = result.replace(f"{{{param_name}}}", str(param_value)) + return result for kind in ("class", "flow"): @@ -109,10 +120,10 @@ class FlowConfig: processor, variant = k.split(":", 1) - variant = repl_template(variant) + variant = repl_template_with_params(variant) v = { - repl_template(k2): repl_template(v2) + repl_template_with_params(k2): repl_template_with_params(v2) for k2, v2 in v.items() } @@ -131,10 +142,10 @@ class FlowConfig: def repl_interface(i): if isinstance(i, str): - return repl_template(i) + return repl_template_with_params(i) else: return { - k: repl_template(v) + k: repl_template_with_params(v) for k, v in i.items() } @@ -152,6 +163,7 @@ class FlowConfig: "description": msg.description, "class-name": msg.class_name, "interfaces": interfaces, + "parameters": parameters, }) ) @@ -177,15 +189,20 @@ class FlowConfig: raise RuntimeError("Internal error: flow has no flow class") class_name = flow["class-name"] + parameters = flow.get("parameters", {}) cls = json.loads(await self.config.get("flow-classes").get(class_name)) def repl_template(tmp): - return tmp.replace( + result = tmp.replace( "{class}", class_name ).replace( "{id}", msg.flow_id ) + # Apply parameter substitutions + for param_name, param_value in parameters.items(): + result = result.replace(f"{{{param_name}}}", str(param_value)) + return result for kind in ("flow",):