Asynchronous Chat Streaming Guide

This guide demonstrates how to use the ~venice_ai.AsyncVeniceClient to stream chat completions asynchronously. Streaming allows you to receive parts of the response as they are generated, which can be useful for interactive applications.

Prerequisites

  • Ensure you have the venice-ai package installed.

  • Set your VENICE_API_KEY environment variable.

Core Concepts

  • AsyncVeniceClient: The primary client for asynchronous interactions.

  • client.chat.completions.create(): The method used to generate chat responses.

  • stream=True: Parameter to enable streaming.

  • AsyncIterator: The streaming response is an async iterator that yields ChatCompletionChunk objects.

Example: Streaming a Chat Response

The following example shows how to send a prompt and stream the response:

 1import os
 2import asyncio
 3from venice_ai import AsyncVeniceClient, APIError
 4
 5async def main():
 6    """Demonstrates asynchronous chat streaming with Venice AI."""
 7    try:
 8        # Ensure VENICE_API_KEY is set in your environment
 9        if not os.getenv("VENICE_API_KEY"):
10            print("Error: VENICE_API_KEY environment variable not set.")
11            return
12
13        async with AsyncVeniceClient() as client:
14            print("Streaming chat response from Venice AI...")
15            print("---")
16
17            prompt_messages = [
18                {"role": "user", "content": "Tell me a short story about a brave knight and a friendly dragon."}
19            ]
20
21            # Using a common model, replace if needed
22            model_id = "llama-3.2-3b"
23
24            stream = await client.chat.completions.create(
25                model=model_id,
26                messages=prompt_messages,
27                stream=True,
28                max_completion_tokens=150 # Optional: limit response length
29            )
30
31            full_response = []
32            async for chunk in stream:
33                # Ensure 'choices' and 'delta' are present and valid
34                if chunk.choices and len(chunk.choices) > 0:
35                    delta = chunk.choices[0].delta
36                    if delta and delta.content:
37                        content_delta = delta.content
38                        print(content_delta, end="", flush=True)
39                        full_response.append(content_delta)
40
41            print("\n---\nStream finished.")
42            # print(f"Full assembled response: {''.join(full_response)}")
43
44    except APIError as e:
45        print(f"\nAn API Error occurred: {e.status_code} - {e.message}")
46        if e.body:
47            print(f"Error details: {e.body}")
48    except Exception as e:
49        print(f"\nAn unexpected error occurred: {e}")
50
51if __name__ == "__main__":
52    asyncio.run(main())

Explanation

  1. Import necessary modules: os for environment variables, asyncio for running async code, and ~venice_ai.AsyncVeniceClient, APIError from venice_ai.

  2. Define an async main function: All asynchronous operations happen within this function.

  3. Initialize AsyncVeniceClient: The client is instantiated within an async with block to ensure proper resource management (automatic aclose() on exit).

  4. Define a prompt: A simple user message is created.

  5. Call `create` with `stream=True`: - model: Specify a valid chat model ID. - messages: Provide the list of messages. - stream=True: This is crucial for enabling streaming.

  6. Iterate over the stream: The async for chunk in stream: loop processes each ChatCompletionChunk as it arrives.

  7. Extract content: The content_delta is extracted from chunk[‘choices’][0][‘delta’].get(‘content’, ‘’).

  8. Print content: The partial content is printed immediately. flush=True ensures it’s displayed without buffering.

  9. Error Handling: Includes try…except blocks to catch potential APIError or other exceptions.

Running the Example

Save the code as a Python file (e.g., run_async_stream.py) and run it from your terminal:

python run_async_stream.py

You should see the story about the brave knight printed to your console, word by word, as it’s generated by the AI.

Key Takeaways

  • Asynchronous streaming is ideal for applications requiring real-time feedback.

  • Always use async with AsyncVeniceClient(…) for proper client lifecycle management.

  • Handle chunks carefully, as they represent partial updates to the overall message.

  • Implement robust error handling for API interactions.

Further Exploration

  • Explore other parameters of the create method to customize model behavior (e.g., temperature, max_completion_tokens).

  • When using models with larger context windows like “Venice Large” (128k tokens), you can adjust max_completion_tokens accordingly to leverage the full context for longer responses or more detailed analysis.

  • Integrate this streaming logic into a web application or a command-line interface for a more interactive experience.

  • Refer to the API Reference for detailed information on all available methods and parameters.

Synchronous Streaming

While this guide focuses on asynchronous streaming with AsyncVeniceClient, the library also supports synchronous streaming for chat completions using the VeniceClient.

You can find an example of synchronous streaming in the main docstring for the VeniceClient class within the API Reference.