import requests
import secrets
from datetime import datetime
API_KEY = "ak_..."
BASE_URL = "https://app.artanis.ai"
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
trace_id = f"trace_{secrets.token_hex(11)}"
# Helper function
def send_observation(obs_type, data, key=None):
payload = {
"trace_id": trace_id,
"type": obs_type,
"data": data,
"timestamp": datetime.utcnow().isoformat(timespec='milliseconds') + 'Z'
}
if key:
payload["key"] = key
requests.post(f"{BASE_URL}/api/v1/observations", headers=headers, json=payload)
# 1. Record input (user query)
send_observation("input", {
"question": "What is your refund policy?",
"model": "gpt-4"
})
# 2. Capture state (current document corpus)
send_observation("state", ["doc-123", "doc-456", "doc-789"], key="documents")
# 3. Your retrieval logic
retrieved_docs = [
{"id": "doc-123", "score": 0.95, "title": "Refund Policy"},
{"id": "doc-456", "score": 0.87, "title": "Returns FAQ"}
]
# 4. Capture retrieved chunks
send_observation("state", retrieved_docs, key="retrieved_chunks")
# 5. Capture generation config
send_observation("state", {
"model": "gpt-4",
"temperature": 0.7,
"prompt_template": "Answer based on: {context}"
}, key="config")
# 6. Your generation logic
response = "We offer a 30-day money-back guarantee on all purchases."
# 7. Record output
send_observation("output", response)
print(f"Recorded complete RAG pipeline for trace {trace_id}")