feat(streaming): initial implementation of streaming agentservice execution by douglas-reid · Pull Request #554 · steamship-core/python-client

douglas-reid · 2023-09-25T05:08:54Z

This PR establishes an initial implementation of streaming utilities for AgentService endpoints, based on FunctionsBasedAgent runs. This will allow clients to invoke an agent run through an async endpoint and stream, via SSE, block creation events related to the agent's execution. These events can include status messages (from Agents and Tools) as well as generated blocks (from LLMs and Tools).

With this code, new async endpoints can be exposed with code like the following:

    @post("async_prompt")
    def async_prompt(
        self, prompt: Optional[str] = None, context_id: Optional[str] = None, **kwargs
    ) -> StreamingResponse:
        ctx_id, history_file = self._streaming_context_id_and_file(context_id=context_id, **kwargs)
        task = self.invoke_later(
            "/prompt", arguments={"prompt": prompt, "context_id": ctx_id, **kwargs}
        )
        return StreamingResponse(task=task, file=history_file)

The PR ensures all Blocks created in the service of a run_agent call are tagged with the proper request-id, and that all calls to the LLM are streaming-compatible.

A test is provided that demonstrates an approach to consuming a stream of events, based on the sseclient-py library. A Generator is constructed that emits Blocks until a terminal block for a request is found in the stream.

eob · 2023-09-28T16:02:33Z

 from steamship.invocable import PackageService, post


+class StreamingResponse(BaseModel):


@douglas-reid if you extend BaseModel instead of CamelModel then the serialization is underscored rather than the camelcase that we use for engine comms.

It sneaks through in the case since it's internal to an object's response, but it breaks the parsing of the task in the Typescript client.

Would it be possible to add a test that, e.g., response.task.requestId is serialized as requestId instead of request_id just to make sure that the clients demarshalling it won't explode?

eob

A few non-blocking thoughts in there, but this looks fantastic. Excited for it to be merged!!

eob · 2023-09-29T18:40:42Z

+
+        assert num_blocks > 0, "Blocks should have been streamed during execution"
+        assert llm_prompt_event_count == 2, (
+            "At least 2 llm prompts should have happened (first for tool selection, "


These error messages are awesome -- thank you for adding them

eob · 2023-09-29T18:44:23Z

+
+        # if you can't find a consistent context_id, then something has gone wrong, preventing streaming
+        if not ctx_id:
+            # TODO(dougreid): this points to a slight flaw in the context_keys vs. context_id


Is this the situation where in practice we've just been using context_id whereas in theory we're hashing the whole context_keys object?

yeah. we have a mismatch in our exposed args in /prompt vs. our actual capability. not sure if that matter practically yet.

eob · 2023-09-29T18:45:43Z

+        self, prompt: Optional[str] = None, context_id: Optional[str] = None, **kwargs
+    ) -> StreamingResponse:
+        ctx_id, history_file = self._streaming_context_id_and_file(context_id=context_id, **kwargs)
+        task = self.invoke_later(


I think in a V2 we'll want to follow up and see if we can have the things that are later-invoked auto-block this task from completing (thinking about the comment string about the semantics of the Task below.. we could maybe try to make conformance to that automatic)

This PR presents a series of changes that should support a way to stream response information back to a client via an AgentService. In order to achieve the streaming result, a new method on the AgentService is exposed: `async_prompt`. This new method returns a new `StreamingResponse` that has two fields: `task` and `file`. These fields provide access to (a) the async task that will be streaming results and (b) a file (here the `ChatHistory` file) to which all status messages and assistant interactions will be saved. This PR relies on a full deployment of steamship-core/gpt4#10 to the target environment for testing / validation.

…agging

douglas-reid requested review from dkolas and eob September 25, 2023 05:08

douglas-reid force-pushed the doug/streaming-with-client-side-request-id-tagging branch from 12b0933 to 832eb1f Compare September 26, 2023 05:06

eob reviewed Sep 28, 2023

View reviewed changes

douglas-reid force-pushed the doug/streaming-with-client-side-request-id-tagging branch from 832eb1f to f9c4b17 Compare September 28, 2023 21:58

douglas-reid changed the title ~~feat(streaming): proof of concept for streaming agentservice execution (attempt 2)~~ feat(streaming): initial implementation of streaming agentservice execution Sep 28, 2023

douglas-reid marked this pull request as ready for review September 28, 2023 22:07

douglas-reid force-pushed the doug/streaming-with-client-side-request-id-tagging branch 3 times, most recently from 925d586 to 0b65f6c Compare September 28, 2023 22:21

eob previously approved these changes Sep 29, 2023

View reviewed changes

douglas-reid dismissed eob’s stale review via 0421c4c October 2, 2023 17:51

Douglas Reid added 11 commits October 3, 2023 14:30

modifications

f8c5dd1

clean up test

c84b8d7

full test with in-flight streaming

19998c7

add stop tag

d47d24a

add signal to context, refactor bits

3b61080

upgrade to python3.10

ce1c850

small test fixes

b3dca69

always use a different workspace

bef499d

better fixtures fix

100b3a6

moar test fixes

b64f2cb

douglas-reid force-pushed the doug/streaming-with-client-side-request-id-tagging branch from 0421c4c to a66d6c7 Compare October 3, 2023 21:50

test cleanup

ad653c6

douglas-reid force-pushed the doug/streaming-with-client-side-request-id-tagging branch from a66d6c7 to ad653c6 Compare October 3, 2023 21:54

Douglas Reid added 4 commits October 3, 2023 16:21

test cleanup

a986a4a

fix sorting issue

a08c82e

fix resource leak fix logic

675ca90

reorder system messages

a77b14d

Merge branch 'main' into doug/streaming-with-client-side-request-id-t…

2f4dc59

…agging

douglas-reid merged commit 687b25c into main Oct 5, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(streaming): initial implementation of streaming agentservice execution#554

feat(streaming): initial implementation of streaming agentservice execution#554
douglas-reid merged 17 commits into
mainfrom
doug/streaming-with-client-side-request-id-tagging

douglas-reid commented Sep 25, 2023 •

edited

Loading

Uh oh!

eob Sep 28, 2023

Uh oh!

douglas-reid Sep 28, 2023

Uh oh!

eob left a comment

Uh oh!

eob Sep 29, 2023

Uh oh!

eob Sep 29, 2023

Uh oh!

douglas-reid Oct 2, 2023

Uh oh!

eob Sep 29, 2023

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

		from steamship.invocable import PackageService, post


		class StreamingResponse(BaseModel):

Conversation

douglas-reid commented Sep 25, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

eob Sep 28, 2023

Choose a reason for hiding this comment

Uh oh!

douglas-reid Sep 28, 2023

Choose a reason for hiding this comment

Uh oh!

eob left a comment

Choose a reason for hiding this comment

Uh oh!

eob Sep 29, 2023

Choose a reason for hiding this comment

Uh oh!

eob Sep 29, 2023

Choose a reason for hiding this comment

Uh oh!

douglas-reid Oct 2, 2023

Choose a reason for hiding this comment

Uh oh!

eob Sep 29, 2023

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

douglas-reid commented Sep 25, 2023 •

edited

Loading