0495dd4676
Implement core infrastructure: - BaseClientMixin with retry logic and validation - FastGPTClient base class with httpx - ChatClient with 11 chat operation methods - AppClient for analytics and logs - Custom exceptions (APIError, AuthenticationError, etc.) - Package configuration (pyproject.toml, setup.py) - Documentation (README.md, CLAUDE.md) - Basic usage examples 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
126 lines
3.8 KiB
Python
126 lines
3.8 KiB
Python
"""Basic usage example for FastGPT Python SDK."""
|
|
|
|
from fastgpt_client import ChatClient
|
|
from dotenv import load_dotenv
|
|
import os
|
|
|
|
load_dotenv()
|
|
|
|
# Configure your API key and base URL
|
|
API_KEY = os.getenv("API_KEY")
|
|
BASE_URL = os.getenv("BASE_URL")
|
|
|
|
|
|
def simple_chat():
|
|
"""Simple chat completion example."""
|
|
with ChatClient(api_key=API_KEY, base_url=BASE_URL) as client:
|
|
response = client.create_chat_completion(
|
|
messages=[{"role": "user", "content": "Hello! What's AI?"}],
|
|
stream=False
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
|
|
print("Response:", result['choices'][0]['message']['content'])
|
|
|
|
|
|
def streaming_chat():
|
|
"""Streaming chat completion example."""
|
|
import json
|
|
|
|
with ChatClient(api_key=API_KEY, base_url=BASE_URL) as client:
|
|
response = client.create_chat_completion(
|
|
messages=[{"role": "user", "content": "Tell me a short story"}],
|
|
stream=True
|
|
)
|
|
|
|
print("Streaming response: ", end="")
|
|
for line in response.iter_lines():
|
|
if line.startswith("data:"):
|
|
data = line[5:].strip()
|
|
if data and data != "[DONE]":
|
|
chunk = json.loads(data)
|
|
if "choices" in chunk and chunk["choices"]:
|
|
delta = chunk["choices"][0].get("delta", {})
|
|
content = delta.get("content", "")
|
|
if content:
|
|
print(content, end="", flush=True)
|
|
print()
|
|
|
|
|
|
def chat_with_context():
|
|
"""Chat with context using chatId example."""
|
|
with ChatClient(api_key=API_KEY, base_url=BASE_URL) as client:
|
|
chat_id = os.getenv("CHAT_ID")
|
|
|
|
# First message
|
|
print("User: What's AI?")
|
|
response = client.create_chat_completion(
|
|
messages=[{"role": "user", "content": "What's AI?"}],
|
|
chatId=chat_id,
|
|
stream=False
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
print(f"AI: {result['choices'][0]['message']['content']}\n")
|
|
|
|
# Second message (continues the conversation)
|
|
print("User: Tell me more about it")
|
|
response = client.create_chat_completion(
|
|
messages=[{"role": "user", "content": "Tell me more about it"}],
|
|
chatId=chat_id, # Same chatId maintains context
|
|
stream=False
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()
|
|
print(f"AI: {result['choices'][0]['message']['content']}")
|
|
|
|
|
|
def get_histories():
|
|
"""Get chat histories example."""
|
|
with ChatClient(api_key=API_KEY, base_url=BASE_URL) as client:
|
|
# You need to replace this with your actual app ID
|
|
app_id = os.getenv("APP_ID")
|
|
|
|
try:
|
|
histories = client.get_chat_histories(
|
|
appId=app_id,
|
|
offset=0,
|
|
pageSize=20,
|
|
source="api"
|
|
)
|
|
histories.raise_for_status()
|
|
data = histories.json()
|
|
|
|
print(f"Total chats: {data['data']['total']}")
|
|
for chat in data['data']['list']:
|
|
print(f" - {chat['title']}: {chat['chatId']}")
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("=== Simple Chat ===")
|
|
try:
|
|
simple_chat()
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
|
|
print("\n=== Streaming Chat ===")
|
|
try:
|
|
streaming_chat()
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
|
|
print("\n=== Chat with Context ===")
|
|
try:
|
|
chat_with_context()
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
|
|
print("\n=== Get Histories ===")
|
|
try:
|
|
get_histories()
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|