File size: 4,925 Bytes
62a1756 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 | """import requests
import json
# Base URL
BASE_URL = "http://localhost:8000"
# 1. Register a new user
register_data = {
"username": "testuser",
"email": "test@example.com",
"company": "Test Co",
"password": "securepassword123"
}
response = requests.post(f"{BASE_URL}/auth/register", json=register_data)
print("Register:", response.json())
# 2. Login (get access/refresh tokens)
login_data = {
"username": "testuser",
"password": "securepassword123"
}
response = requests.post(f"{BASE_URL}/auth/login", data=login_data)
tokens = response.json()
access_token = tokens["access_token"]
refresh_token = tokens["refresh_token"]
print("Login:", tokens)
# Headers for authenticated requests
headers = {"Authorization": f"Bearer {access_token}"}
# 3. Create a conversation
response = requests.post(f"{BASE_URL}/rag/conversations", headers=headers)
conv_id = response.json()["conversation_id"]
print("Conversation ID:", conv_id)
# 4. Send a message (with optional files, web search)
# Example: Text-only message
files = [] # Or: [('files', open('doc.pdf', 'rb'))] for uploads
data = {
"model": "llama-3.1-8b-instant",
"enable_web_search": True,
"message": "What is the capital of France?"
}
response = requests.post(
f"{BASE_URL}/rag/conversations/{conv_id}/messages",
headers=headers,
data=data,
files=files if files else None,
stream=True
)
for chunk in response.iter_content(chunk_size=1024):
if chunk:
print(chunk.decode(), end='', flush=True) # Streaming output
# 5. Get conversation history
response = requests.get(f"{BASE_URL}/rag/conversations/{conv_id}", headers=headers)
print("History:", response.json())
# 6. Refresh token
refresh_data = {"refresh_token": refresh_token}
response = requests.post(f"{BASE_URL}/auth/refresh", json=refresh_data)
new_tokens = response.json()
print("New Tokens:", new_tokens)
# 7. Logout
logout_data = {"refresh_token": refresh_token}
response = requests.post(f"{BASE_URL}/auth/logout", json=logout_data)
print("Logout:", response.json())"""
import requests
import json
# Base URL
BASE_URL = "http://localhost:8000"
# 1. Login (get access/refresh tokens) - Change credentials if needed
login_data = {
"username": "testuser", # Update if your username is different
"password": "securepassword123" # Update with your actual password
}
response = requests.post(f"{BASE_URL}/auth/login", data=login_data)
if response.status_code != 200:
print("Login Failed:", response.status_code, response.text)
else:
tokens = response.json()
access_token = tokens["access_token"]
refresh_token = tokens["refresh_token"]
print("Login Success:", tokens)
# Headers for authenticated requests
headers = {"Authorization": f"Bearer {access_token}"}
# 2. Create a conversation
response = requests.post(f"{BASE_URL}/rag/conversations", headers=headers)
if response.status_code == 201:
conv_id = response.json()["conversation_id"]
print("Conversation Created - ID:", conv_id)
else:
print("Failed to create conversation:", response.status_code, response.text)
conv_id = None
if conv_id:
# 3. Send a message (text-only example)
data = {
"model": "llama-3.1-8b-instant", # Change model if desired (from ALLOWED_MODELS)
"enable_web_search": "true", # "true" or "false" as string for form data
"message": "What is the capital of France?"
}
# Optional: Add files for document RAG
# files = [('files', open('your_document.pdf', 'rb'))]
response = requests.post(
f"{BASE_URL}/rag/conversations/{conv_id}/messages",
headers=headers,
data=data,
# files=files if 'files' in locals() else None,
stream=True
)
print("\n--- Assistant Response ---")
if response.status_code == 200:
for chunk in response.iter_content(chunk_size=1024, decode_unicode=True):
if chunk:
print(chunk, end='', flush=True)
print("\n--- End of Response ---")
else:
print("Message Send Failed:", response.status_code)
print("Response:", response.text)
# 4. Get conversation history
response = requests.get(f"{BASE_URL}/rag/conversations/{conv_id}", headers=headers)
print("\nConversation History:", json.dumps(response.json(), indent=2))
# 5. Refresh token (optional)
refresh_data = {"refresh_token": refresh_token}
response = requests.post(f"{BASE_URL}/auth/refresh", json=refresh_data)
print("Token Refresh:", response.json() if response.status_code == 200 else response.text)
# 6. Logout (optional)
logout_data = {"refresh_token": refresh_token}
response = requests.post(f"{BASE_URL}/auth/logout", json=logout_data)
print("Logout:", response.json()) |