[Feature][api][python] Introduce Python API for Workflow, Event and Action #32
+493
−1
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
wenjin272
changed the title
xintongsong
force-pushed
the
init-workflow-api
branch
from
cdaa475 to
af9056f
Compare
xintongsong
reviewed
Jun 22, 2025
| @action(InputEvent) | ||
| @staticmethod | ||
| def first_action(event: Event, ctx: WorkflowRunnerContext): #noqa D102 | ||
| event.input += " first_action" |
There was a problem hiding this comment.
This is not a good example. Events received by an action should be immutable, and we should not encourage users to modify them.
Comment on lines
+27
to
+31
| if __name__ == "__main__": | ||
| workflow = MyWorkflow() | ||
| session_id = workflow.run(input="input", runner='LocalRunner') | ||
| for output in workflow.get_outputs(session_id): | ||
| print(output) |
There was a problem hiding this comment.
- We should not have
runandget_outputson a workflow, because they only make sense in local runner. It would be problematic if user calls them when executing in Flink runner. - We probably should replace the concept of
session_idwithkey. - If
workflow.runis only for local runner, we should not require user to specify therunner='LocalRunner'
There was a problem hiding this comment.
I'd suggest something like:
in_q = Queue()
out_q = (FlinkAgentsExecutionEnvironment
.from_queue(in_q)
.apply(workflow)
.to_queue()
)
in_q.put(("intput1", "key"))
in_q.put("input2") # automatically generate new unique key
while not out_q.empty():
(key, output) = out_q.get()
print(key, output)
| Internal workflow runner instance used to execute the workflow. | ||
| """ | ||
|
|
||
| __runner: WorkflowRunner = None |
There was a problem hiding this comment.
The workflow should not be aware of the runner.
| from uuid import UUID | ||
|
|
||
|
|
||
| class WorkflowRunner(ABC): |
There was a problem hiding this comment.
The workflow runner should not be a public API. We don't want users to call it or implement it.
| from flink_agents.api.event import Event | ||
|
|
||
|
|
||
| class WorkflowRunnerContext(ABC): |
There was a problem hiding this comment.
Let's just call this RunnerContext.
Comment on lines
+25
to
+47
| @abstractmethod | ||
| def add_output(self, output: Any) -> None: | ||
| """Add an output generate by workflow execution to the context. | ||
|
|
||
| Parameters | ||
| ---------- | ||
| output : Any | ||
| The output to be added to the queue. | ||
| """ | ||
|
|
||
| @abstractmethod | ||
| def get_outputs(self) -> deque[Any]: | ||
| """Get outputs stored in this context. | ||
|
|
||
| Returns: | ||
| ------- | ||
| deque[Any] | ||
| The outputs generated by workflow execution on this context. | ||
| """ | ||
|
|
||
| @abstractmethod | ||
| def clear_output(self) -> None: | ||
| """Clear outputs stored in this context.""" |
There was a problem hiding this comment.
Why do we need these in the user-facing API?
| from flink_agents.api.event import Event | ||
|
|
||
|
|
||
| #TODO: implement Closure to support access self in action function, like llama-index |
| from flink_agents.plan.function import PythonFunction | ||
|
|
||
|
|
||
| def get_actions(workflow: Workflow) -> List[Action]: |
There was a problem hiding this comment.
Why introducing this util file?
|
|
||
| class MyWorkflow(Workflow): #noqa D101 | ||
| @action(InputEvent) | ||
| @staticmethod |
There was a problem hiding this comment.
Is it possible to get rid of this @staticmethod?
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Linked issue: #15
This PR is based on the previous PR #6 , and the commit from [[Feature][api][python]] is the code of this PR.
This PR import Workflow which users can implement it to design their own workflow logic, WorkflowRunner & WorkflowRunnerContext for workflow execution, and other utils for converting user defined workflow to workflow plan.
And I implement LocalRunner and LocalRunnerContext for giving an example for WorkflowRunner and WorkflowRunnerContext, together with an example of how to run user defined workflow.