Hi 👋 I'm trying to use ControlFlow with a Prefect server. I have managed to submit flows to the prefect server where they execute. I've found some rough edges though and I wanted to check that I am doing things correctly. I am trying to create a chatbot interface somewhat similar to ChatGPT. It would use specialized tools that are triggered by the conversation. You would be able to save and resume conversations in future. ControlFlow seems really good for this. The LLM usage, especially in interactive mode, is very slick and the prefect workflow system gives me confidence that more complex workflows will be possible. I am trying to work through a very simple example human-in-the-loop flow, using an LLM and a self hosted Prefect server. To submit the flow I create a RunnerDeployment with

<http://flow.to|flow.to>_deployment(name, workpool)

. This leads to the first problem: I have to manually change the entrypoint to replace the relative filename of the python file containing the flow with an absolute path, as the worker appears to execute the flow in a temporary directory. The code is:

runner_deployment: RunnerDeployment = await flow.to_deployment(
        name=f"{flow.name}-deployment",
        work_pool_name="workpool",
    )
    runner_deployment.entrypoint = resolve_entrypoint(runner_deployment.entrypoint)
...
def resolve_entrypoint(entrypoint: str) -> str:
    # PROBLEM:
    # The entrypoint of the flow is relative to the project root, and the local
    # process worker cannot find it
    # SOLUTION:
    # Update the entrypoint to have an absolute path
    relative_file, name = entrypoint.split(":")
    absolute_file = Path(relative_file).resolve()
    return f"{absolute_file}:{name}"

How can I get the code to resolve correctly for a

process

type worker? Is this some sort of local storage thing? I would like to be able to iterate quickly so packaging and publishing docker images is undesirable. Then I

apply

the runner_deployment to get the deployment_id, and I can read the deployment details using the prefect client. I can then run this deployment with

prefect.deployments.run_deployment(name=deployment_id, timeout=0, parameters=...)

. This then returns a FlowRun object which has an

id

which I can use with the

/api/flow_runs/{id}

api endpoint to read details about the flow, or the

/api/flow_runs/{id}/resume

endpoint to resume it. I would like to know how I can pass messages from the flow to the initiating process which may be on another machine etc etc. I can see in the prefect server that the flow run has associated artifacts but I cannot see a way to find the artifacts associated with a flow. I am also unsure if the artifacts are ordered, I would need to be able to see the messages that have been produced since some prior point. Further to this the return value of the ControlFlow flow does not appear to be an artifact - in the example code this is usually the result of the computation, so how can I access it? My aim here is to be able to replace the cli_input method of getting human feedback with a separate tool which suspends the current flow and provides a message to prompt the user with (potentially a rich message with images etc). Then when the user has responded the flow can be resumed with their input. Is this an appropriate use of suspend / resume functionality? Do I have to handle the inter process communication in some separate way from the prefect server? The final part would be the whole suspend resume stuff. I would like to use this to provide the list of previous conversations. Is this a suitable use of it? Doing suspend and resume involves rerunning every stage of the flow until the end, and caching is used to both ensure that the results are the same as the prior run and skip the various costs involved in rerunning everything. One thing I like about the ControlFlow LLM invocations is the interactive ones can ask further questions to the user to clarify the answer, and it is these kinds of situations where the suspend / resume would be helpful. Is there a way to partition the flows as a conversation could potentially be infinitely long? Are there constraints around the function caching that I should be aware of? Is there a way to retrieve the state of a flow including the cached task calls so that I can store and restore the cache if needed? Thanks again for such a great tool

hi @Matthew - tldr I think you don't need to leverage so many of the implementation details here's how you can "deploy controlflow"

oh is this how you do the openai key? i provided it to the worker when i started it:

"env": {"OPENAI_API_KEY": os.getenv("OPENAI_API_KEY")},

@Matthew did you figure out how to pass messages without the console?

i'm gonna be real, suspend/resume using prefect is not the way. iirc prefect api has a resume which can accept arguments to pas, inspecting the console or reviewing the swagger docs will help you there. the problem is that it "resumes" the flow by running all the steps again so you are dependent on caching everything all the time. in contrast marvin-3 has threads which have serializable messages. with a bit of hackery you can suspend resume the threads by persisting the thread, reloading it, and then recreating the orchestrator that was running it