| | |
| | """ |
| | Simple usage example for the AI Backend Service |
| | Demonstrates how to interact with the OpenAI-compatible API |
| | """ |
| |
|
| | import requests |
| | import json |
| |
|
| | |
| | BASE_URL = "http://localhost:8000" |
| |
|
| | def test_simple_chat(): |
| | """Simple chat completion example""" |
| | print("π€ Simple Chat Example") |
| | print("-" * 30) |
| | |
| | response = requests.post(f"{BASE_URL}/v1/chat/completions", json={ |
| | "model": "microsoft/DialoGPT-medium", |
| | "messages": [ |
| | {"role": "system", "content": "You are a helpful assistant."}, |
| | {"role": "user", "content": "What is the capital of France?"} |
| | ], |
| | "max_tokens": 100, |
| | "temperature": 0.7 |
| | }) |
| | |
| | if response.status_code == 200: |
| | data = response.json() |
| | message = data["choices"][0]["message"]["content"] |
| | print(f"Assistant: {message}") |
| | else: |
| | print(f"Error: {response.status_code} - {response.text}") |
| | print() |
| |
|
| | def test_streaming_chat(): |
| | """Streaming chat completion example""" |
| | print("π Streaming Chat Example") |
| | print("-" * 30) |
| | |
| | response = requests.post(f"{BASE_URL}/v1/chat/completions", json={ |
| | "model": "microsoft/DialoGPT-medium", |
| | "messages": [ |
| | {"role": "user", "content": "Tell me a fun fact about space"} |
| | ], |
| | "max_tokens": 150, |
| | "temperature": 0.8, |
| | "stream": True |
| | }, stream=True) |
| | |
| | if response.status_code == 200: |
| | print("Assistant: ", end="", flush=True) |
| | for line in response.iter_lines(): |
| | if line: |
| | line_str = line.decode('utf-8') |
| | if line_str.startswith('data: '): |
| | data_part = line_str[6:] |
| | if data_part == '[DONE]': |
| | break |
| | try: |
| | chunk = json.loads(data_part) |
| | if 'choices' in chunk and chunk['choices']: |
| | delta = chunk['choices'][0].get('delta', {}) |
| | if 'content' in delta: |
| | print(delta['content'], end='', flush=True) |
| | except json.JSONDecodeError: |
| | pass |
| | print("\n") |
| | else: |
| | print(f"Error: {response.status_code} - {response.text}") |
| | print() |
| |
|
| | def test_text_completion(): |
| | """Text completion example""" |
| | print("π Text Completion Example") |
| | print("-" * 30) |
| | |
| | response = requests.post(f"{BASE_URL}/v1/completions", json={ |
| | "prompt": "The best programming language for beginners is", |
| | "max_tokens": 80, |
| | "temperature": 0.6 |
| | }) |
| | |
| | if response.status_code == 200: |
| | data = response.json() |
| | completion = data["choices"][0]["text"] |
| | print(f"Completion: {completion}") |
| | else: |
| | print(f"Error: {response.status_code} - {response.text}") |
| | print() |
| |
|
| | def test_service_info(): |
| | """Get service information""" |
| | print("βΉοΈ Service Information") |
| | print("-" * 30) |
| | |
| | |
| | health = requests.get(f"{BASE_URL}/health") |
| | if health.status_code == 200: |
| | print(f"Service Status: {health.json()['status']}") |
| | print(f"Model: {health.json()['model']}") |
| | |
| | |
| | models = requests.get(f"{BASE_URL}/v1/models") |
| | if models.status_code == 200: |
| | model_list = models.json()["data"] |
| | print(f"Available Models: {len(model_list)}") |
| | for model in model_list: |
| | print(f" - {model['id']}") |
| | print() |
| |
|
| | if __name__ == "__main__": |
| | print("π AI Backend Service - Usage Examples") |
| | print("=" * 50) |
| | |
| | try: |
| | test_service_info() |
| | test_simple_chat() |
| | test_text_completion() |
| | test_streaming_chat() |
| | |
| | print("β
All examples completed successfully!") |
| | |
| | except requests.exceptions.ConnectionError: |
| | print("β Could not connect to the service.") |
| | print("Make sure the backend service is running on http://localhost:8000") |
| | print("Start it with: python backend_service.py --port 8000") |
| | except Exception as e: |
| | print(f"β Error: {e}") |
| |
|