From 00d0055d8fda6f5ea74322220b31f7cb0ac45644 Mon Sep 17 00:00:00 2001 From: pliam <1033496103@qq.com> Date: Sat, 19 Apr 2025 11:42:01 +0800 Subject: [PATCH 1/2] core_streaming_response_fastapi --- .../core_streaming_response_fastapi/README.md | 104 +++++++++++++ .../core_streaming_response_fastapi/app.py | 143 ++++++++++++++++++ 2 files changed, 247 insertions(+) create mode 100644 python/samples/core_streaming_response_fastapi/README.md create mode 100644 python/samples/core_streaming_response_fastapi/app.py diff --git a/python/samples/core_streaming_response_fastapi/README.md b/python/samples/core_streaming_response_fastapi/README.md new file mode 100644 index 000000000000..48862a43b643 --- /dev/null +++ b/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://: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}") +``` + diff --git a/python/samples/core_streaming_response_fastapi/app.py b/python/samples/core_streaming_response_fastapi/app.py new file mode 100644 index 000000000000..b9cf12b0aea7 --- /dev/null +++ b/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) From 910d35e8051ec5cdfd69971ddcbbef58553b9b93 Mon Sep 17 00:00:00 2001 From: Eric Zhu Date: Mon, 21 Apr 2025 16:46:42 -0700 Subject: [PATCH 2/2] Add model_client.yaml --- .../.gitignore | 5 + .../core_streaming_response_fastapi/README.md | 11 +- .../core_streaming_response_fastapi/app.py | 172 +++++++++--------- .../model_config_template.yaml | 26 +++ 4 files changed, 118 insertions(+), 96 deletions(-) create mode 100644 python/samples/core_streaming_response_fastapi/.gitignore create mode 100644 python/samples/core_streaming_response_fastapi/model_config_template.yaml diff --git a/python/samples/core_streaming_response_fastapi/.gitignore b/python/samples/core_streaming_response_fastapi/.gitignore new file mode 100644 index 000000000000..77066c6b1263 --- /dev/null +++ b/python/samples/core_streaming_response_fastapi/.gitignore @@ -0,0 +1,5 @@ +model_config.yaml +agent_state.json +agent_history.json +team_state.json +team_history.json diff --git a/python/samples/core_streaming_response_fastapi/README.md b/python/samples/core_streaming_response_fastapi/README.md index 48862a43b643..0406b3e69b53 100644 --- a/python/samples/core_streaming_response_fastapi/README.md +++ b/python/samples/core_streaming_response_fastapi/README.md @@ -22,15 +22,8 @@ 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 -) -``` +Create a new file named `model_config.yaml` in the same directory as this README file to configure your model settings. +See `model_config_template.yaml` for an example. **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. diff --git a/python/samples/core_streaming_response_fastapi/app.py b/python/samples/core_streaming_response_fastapi/app.py index b9cf12b0aea7..9643b1c4a7c6 100644 --- a/python/samples/core_streaming_response_fastapi/app.py +++ b/python/samples/core_streaming_response_fastapi/app.py @@ -1,143 +1,141 @@ -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 asyncio import json import time +from contextlib import asynccontextmanager +from dataclasses import dataclass +from typing import AsyncGenerator, Dict, List -import asyncio - - +import aiofiles +import yaml 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() +from autogen_core.models import AssistantMessage, ChatCompletionClient, LLMMessage, SystemMessage, UserMessage +from fastapi import FastAPI, HTTPException, Request +from fastapi.responses import StreamingResponse @dataclass -class LlmResponse: +class AgentResponse: """ Represents the final accumulated response content from the LLM agent. - Note: The 'content' field currently holds the full string, not individual chunks. + Note: The 'content' field hold the final response content. """ - content: list + + content: str + @dataclass -class ChatHistory: +class UserRequest: """ 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]] + +# Runtime for the agent. +runtime = SingleThreadedAgentRuntime() + # Queue for streaming results from the agent back to the request handler -response_queue = asyncio.Queue() +response_queue: asyncio.Queue[str | object] = 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: + 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. + async def handle_user_message(self, message: UserRequest, ctx: MessageContext) -> AgentResponse: + 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 + user_messages: List[LLMMessage] = [] + for m in _message: + if m["source"] == "user": + user_messages.append(UserMessage(content=m["source"], source=m["source"])) + else: + user_messages.append(AssistantMessage(content=m["source"], source=m["source"])) + # Create a stream of messages to the model client. + async for i in self._model_client.create_stream(user_messages, 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) + return AgentResponse(content=accumulated_content) except Exception as e: - await self._response_queue.put('ERROR:'+e) - return LlmResponse(content=e) - -runtime = SingleThreadedAgentRuntime() + await self._response_queue.put("ERROR:" + str(e)) + return AgentResponse(content=str(e)) -@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.") - +@asynccontextmanager +async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: + # Get model client from config. + async with aiofiles.open("model_config.yaml", "r") as file: + model_config = yaml.safe_load(await file.read()) + model_client = ChatCompletionClient.load_component(model_config) -@app.on_event("startup") -async def startup_event(): + # Register the agent with the runtime. await MyAgent.register( runtime, "simple_agent", - lambda: MyAgent('myagent', - OpenAIChatCompletionClient( - model="gemini-2.0-flash", - api_key="YOUR_API_KEY_HERE", - ) + lambda: MyAgent( + "myagent", + model_client=model_client, ), ) + + # Start the agent runtime. runtime.start() + yield + await runtime.stop() + + +app = FastAPI(lifespan=lifespan) + + +@app.post("/chat/completions") +async def chat_completions_stream(request: Request): + json_data = await request.json() + messages = json_data.get("messages", "") + if not isinstance(messages, list): + raise HTTPException(status_code=400, detail="Invalid input: 'messages' must be a list.") + user_request = UserRequest(messages=messages) # type: ignore + + async def response_stream() -> AsyncGenerator[str, None]: + task1 = asyncio.create_task(runtime.send_message(user_request, 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" + + # Wait for the task to finish. + await task1 + + return StreamingResponse(response_stream(), media_type="text/plain") # type: ignore + if __name__ == "__main__": import uvicorn + uvicorn.run(app, host="0.0.0.0", port=8501) diff --git a/python/samples/core_streaming_response_fastapi/model_config_template.yaml b/python/samples/core_streaming_response_fastapi/model_config_template.yaml new file mode 100644 index 000000000000..9768f5df0fe1 --- /dev/null +++ b/python/samples/core_streaming_response_fastapi/model_config_template.yaml @@ -0,0 +1,26 @@ +# Use Open AI with key +provider: autogen_ext.models.openai.OpenAIChatCompletionClient +config: + model: gpt-4o + api_key: REPLACE_WITH_YOUR_API_KEY +# Use Azure Open AI with key +# provider: autogen_ext.models.openai.AzureOpenAIChatCompletionClient +# config: +# model: gpt-4o +# azure_endpoint: https://{your-custom-endpoint}.openai.azure.com/ +# azure_deployment: {your-azure-deployment} +# api_version: {your-api-version} +# api_key: REPLACE_WITH_YOUR_API_KEY +# Use Azure OpenAI with AD token provider. +# provider: autogen_ext.models.openai.AzureOpenAIChatCompletionClient +# config: +# model: gpt-4o +# azure_endpoint: https://{your-custom-endpoint}.openai.azure.com/ +# azure_deployment: {your-azure-deployment} +# api_version: {your-api-version} +# azure_ad_token_provider: +# provider: autogen_ext.auth.azure.AzureTokenProvider +# config: +# provider_kind: DefaultAzureCredential +# scopes: +# - https://cognitiveservices.azure.com/.default