File size: 3,566 Bytes
d058923 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | """
Concept E — many subjects → many REST endpoints
===============================================
A workflow isn't just a canvas — it's an API. Each *disconnected* pipeline
(weakly-connected group of nodes ending in a subject) becomes ONE REST endpoint,
named after its first subject. Free references become the endpoint's parameters.
This graph has two independent pipelines, so it exposes two endpoints:
/word_count (text → number)
/fahrenheit (number → number)
Once launched:
curl http://127.0.0.1:7860/gradio_api/call/word_count -s \
-H "Content-Type: application/json" -d '{"data": ["hello there friend"]}'
This file also prints its endpoint schema at startup via `describe_workflow_api`.
Run it: python concepts/e_multi_endpoint_api.py
"""
import json
import os
import gradio as gr
def word_count(text: str) -> int:
return len(text.split())
def to_fahrenheit(celsius: float) -> float:
return round(celsius * 9 / 5 + 32, 1)
def _ref(nid, label, ptype, default, y):
return {"id": nid, "role": "reference", "label": label, "asset_type": ptype,
"inputs": [{"id": "in", "label": label, "type": ptype}],
"outputs": [{"id": "out", "label": label, "type": ptype}],
"data": {"out": default}, "x": 60, "y": y, "width": 200, "height": 90}
def _op(nid, fn, in_type, out_type, y):
return {"id": nid, "role": "operator", "kind": "fn", "fn": fn, "label": fn,
"inputs": [{"id": "in_0", "label": "input", "type": in_type, "required": True}],
"outputs": [{"id": "out_0", "label": "output", "type": out_type}],
"data": {}, "x": 320, "y": y, "width": 200, "height": 90}
def _sub(nid, label, ptype, y):
return {"id": nid, "role": "subject", "label": label, "asset_type": ptype,
"inputs": [{"id": "in", "label": label, "type": ptype}],
"outputs": [{"id": "out", "label": label, "type": ptype}],
"data": {}, "x": 580, "y": y, "width": 200, "height": 120}
def _edge(eid, s, sp, t, tp, ty):
return {"id": eid, "from_node_id": s, "from_port_id": sp,
"to_node_id": t, "to_port_id": tp, "type": ty}
GRAPH = {
"schema_version": "2", "name": "Two Endpoints",
"references": [_ref("ref_text", "Text", "text", "hello there friend", 60),
_ref("ref_c", "Celsius", "number", 20, 260)],
"operators": [_op("op_wc", "word_count", "text", "number", 60),
_op("op_f", "to_fahrenheit", "number", "number", 260)],
"subjects": [_sub("sub_wc", "Word count", "number", 60),
_sub("sub_f", "Fahrenheit", "number", 260)],
"edges": [_edge("e1", "ref_text", "out", "op_wc", "in_0", "text"),
_edge("e2", "op_wc", "out_0", "sub_wc", "in", "number"),
_edge("e3", "ref_c", "out", "op_f", "in_0", "number"),
_edge("e4", "op_f", "out_0", "sub_f", "in", "number")],
}
GRAPH_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "e_multi_endpoint_api.json")
with open(GRAPH_PATH, "w", encoding="utf-8") as f:
json.dump(GRAPH, f, indent=2)
demo = gr.Workflow(GRAPH_PATH, bind={"word_count": word_count, "to_fahrenheit": to_fahrenheit})
if __name__ == "__main__":
from gradio.workflow_api import WorkflowGraph, describe_workflow_api
for ep in describe_workflow_api(WorkflowGraph.from_json(json.dumps(GRAPH))):
print(f" {ep['api_name']:14s} params={[p['type'] for p in ep['parameters']]} "
f"returns={[r['type'] for r in ep['returns']]}")
demo.launch()
|