Add an example using autogen-core and FastAPI to create streaming responses

python/samples/core_streaming_response_fastapi/README.md

@@ -0,0 +1,104 @@
+# AutoGen-Core Streaming Chat API with FastAPI
+
+This sample demonstrates how to build a streaming chat API with multi-turn conversation history using `autogen-core` and FastAPI.
+
+## Key Features
+
+1.  **Streaming Response**: Implements real-time streaming of LLM responses by utilizing FastAPI's `StreamingResponse`, `autogen-core`'s asynchronous features, and a global queue created with `asyncio.Queue()` to manage the data stream, thereby providing faster user-perceived response times.
+2.  **Multi-Turn Conversation**: The Agent (`MyAgent`) can receive and process chat history records (`ChatHistory`) containing multiple turns of interaction, enabling context-aware continuous conversations.
+
+## File Structure
+
+*   `app.py`: FastAPI application code, including API endpoints, Agent definitions, runtime settings, and streaming logic.
+*   `README.md`: (This document) Project introduction and usage instructions.
+
+## Installation
+
+First, make sure you have Python installed (recommended 3.8 or higher). Then, in your project directory, install the necessary libraries via pip:
+
+```bash
+pip install "fastapi" "uvicorn[standard]" "autogen-core" "autogen-ext[openai]"
+```
+
+## Configuration
+
+**API Key**: In the `app.py` file, find the instantiation of `OpenAIChatCompletionClient` and replace `"YOUR_API_KEY_HERE"` with your actual OpenAI API key.
+
+```python
+# In the startup_event function of app.py
+OpenAIChatCompletionClient(
+    model="gemini-2.0-flash",  # Or other OpenAI style models
+    api_key="YOUR_ACTUAL_API_KEY",  # Configure your API Key here
+)
+```
+
+**Note**: Hardcoding API keys directly in the code is only suitable for local testing. For production environments, it is strongly recommended to use environment variables or other secure methods to manage keys.
+
+## Running the Application
+
+In the directory containing `app.py`, run the following command to start the FastAPI application:
+
+```bash
+uvicorn app:app --host 0.0.0.0 --port 8501 --reload
+```
+
+After the service starts, the API endpoint will be available at `http://<your-server-ip>:8501/chat/completions`.
+
+## Using the API
+
+You can interact with the Agent by sending a POST request to the `/chat/completions` endpoint. The request body must be in JSON format and contain a `messages` field, the value of which is a list, where each element represents a turn of conversation.
+
+**Request Body Format**:
+
+```json
+{
+  "messages": [
+    {"source": "user", "content": "Hello!"},
+    {"source": "assistant", "content": "Hello! How can I help you?"},
+    {"source": "user", "content": "Introduce yourself."}
+  ]
+}
+```
+
+**Example (using curl)**:
+
+```bash
+curl -N -X POST http://localhost:8501/chat/completions \
+-H "Content-Type: application/json" \
+-d '{
+  "messages": [
+    {"source": "user", "content": "Hello, I'\''m Tory."},
+    {"source": "assistant", "content": "Hello Tory, nice to meet you!"},
+    {"source": "user", "content": "Say hello by my name and introduce yourself."}
+  ]
+}'
+```
+
+**Example (using Python requests)**:
+
+```python
+import requests
+import json
+url = "http://localhost:8501/chat/completions"
+data = {
+    'stream': True,
+    'messages': [
+            {'source': 'user', 'content': "Hello,I'm tory."},
+            {'source': 'assistant', 'content':"hello Tory, nice to meet you!"},
+            {'source': 'user', 'content': "Say hello by my name and introduce yourself."}
+        ]
+    }
+headers = {'Content-Type': 'application/json'}
+try:
+    response = requests.post(url, json=data, headers=headers, stream=True)
+    response.raise_for_status()
+    for chunk in response.iter_content(chunk_size=None):
+        if chunk:
+            print(json.loads(chunk)["content"], end='', flush=True)
+
+except requests.exceptions.RequestException as e:
+    print(f"Error: {e}")
+except json.JSONDecodeError as e:
+    print(f"JSON Decode Error: {e}")
+```
+

python/samples/core_streaming_response_fastapi/app.py

@@ -0,0 +1,143 @@
+from fastapi import FastAPI, HTTPException, Header, Depends,Request
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel,Field
+from dataclasses import dataclass, field
+from typing import List, Dict, Optional, Any
+
+import os
+import json
+import time
+
+import asyncio
+
+
+from autogen_core import (
+    AgentId,
+    ClosureAgent,
+    ClosureContext,
+    DefaultTopicId,
+    MessageContext,
+    RoutedAgent,
+    SingleThreadedAgentRuntime,
+    TopicId,
+    TypeSubscription,
+    default_subscription,
+    message_handler,
+    type_subscription,
+)
+
+from autogen_core.models import ChatCompletionClient, SystemMessage, UserMessage,AssistantMessage
+from autogen_ext.models.openai import OpenAIChatCompletionClient
+
+app = FastAPI()
+
+
+@dataclass
+class LlmResponse:
+    """
+    Represents the final accumulated response content from the LLM agent.
+    Note: The 'content' field currently holds the full string, not individual chunks.
+    """
+    content: list
+
+@dataclass
+class ChatHistory:
+    """
+    Represents the chat history, containing a list of messages.
+    Each message is expected to be a dictionary with 'source' and 'content' keys.
+    """
+    messages: List[Dict[str, str]]
+
+# Queue for streaming results from the agent back to the request handler
+response_queue = asyncio.Queue()
+# Sentinel object to signal the end of the stream
+STREAM_DONE = object()
+
+
+
+async def main(msg:list):
+    """
+    Sends messages to the agent and yields results received via the queue.
+
+    Args:
+        msg (list): A list of message dictionaries conforming to ChatHistory.
+
+    Yields:
+        str: JSON strings representing message chunks or completion signals.
+    """
+    message = ChatHistory(messages=msg)
+    task1 = asyncio.create_task(runtime.send_message(message, AgentId("simple_agent", "default")))
+
+    # Consume items from the response queue until the stream ends or an error occurs
+    while True:
+        item = await response_queue.get()
+        if item is STREAM_DONE:
+            print(f"{time.time():.2f} - MAIN: Received STREAM_DONE. Exiting loop.")
+            break
+        elif isinstance(item, str) and item.startswith("ERROR:"):
+            print(f"{time.time():.2f} - MAIN: Received error message from agent: {item}")
+            break
+        else:
+            yield json.dumps({'content': item}) + '\n'
+    await task1           
+    return
+
+
+@type_subscription(topic_type="simple_agent")
+class MyAgent(RoutedAgent):
+    def __init__(self,name:str, model_client: ChatCompletionClient) -> None:
+        super().__init__(name)
+        self._system_messages = [SystemMessage(content="You are a helpful assistant.")]
+        self._model_client = model_client
+        self._response_queue = response_queue
+
+    @message_handler
+    async def handle_user_message(self, message: ChatHistory, ctx: MessageContext) -> LlmResponse:
+        accumulated_content = '' # To store the full response.
+        try:
+            _message = message.messages
+            user_message = [UserMessage(**i) if i['source']=='user' else AssistantMessage(**i) for i in  _message ]
+            async for i in self._model_client.create_stream(
+                user_message, cancellation_token=ctx.cancellation_token
+            ):
+                if isinstance(i,str):
+                    accumulated_content+=i
+                    await self._response_queue.put(i)
+                else:
+                    break
+            await self._response_queue.put(STREAM_DONE)
+            return LlmResponse(content=accumulated_content)
+        except Exception as e:
+            await self._response_queue.put('ERROR:'+e)
+            return LlmResponse(content=e)
+
+runtime = SingleThreadedAgentRuntime()
+
+
+@app.post("/chat/completions")
+async def chat_completions_stream(request:Request):
+    json_data = await request.json()
+    messages = json_data.get('messages','')
+    if isinstance(messages,list):
+        return StreamingResponse(main(messages),media_type="text/plain")
+    else:
+        raise HTTPException(status_code=400, detail="Invalid input: 'messages' must be a list.")
+
+
+@app.on_event("startup")
+async def startup_event():
+    await MyAgent.register(
+        runtime,
+        "simple_agent",
+        lambda: MyAgent('myagent',
+            OpenAIChatCompletionClient(
+                model="gemini-2.0-flash",
+                api_key="YOUR_API_KEY_HERE",
+            )
+        ),
+    )
+    runtime.start()
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8501)