mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-07-24 12:41:02 +02:00
Flow configurable parameters implemented
This commit is contained in:
Cyber MacGeddon
parent
25e8629d59
commit
4552b1853d
6 changed files with 78 additions and 20 deletions
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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]:
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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",):
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue