Use SDK

.github/workflows/autoblocks-replays.yml

@@ -34,6 +34,10 @@ jobs:
 
       - name: Install dependencies
         run: poetry install
+        env:
+          # Temporary while the SDK is not published
+          POETRY_HTTP_BASIC_GIT_AUTOBLOCKSAI_PYTHON_SDK_USERNAME: nicolewhite
+          POETRY_HTTP_BASIC_GIT_AUTOBLOCKSAI_PYTHON_SDK_PASSWORD: ${{ secrets.NICOLES_GITHUB_TOKEN_DO_NOT_USE }}
 
       - name: Start the app
         run: poetry run start &

demo_replays/app.py

@@ -1,11 +1,12 @@
 import uuid
 
+from autoblocks.tracer import AutoblocksTracer
 from flask import Flask
 from flask import request
 
 from demo_replays import bot
-from demo_replays.autoblocks_sdk import AutoblocksLogger
-from demo_replays.settings import settings
+from demo_replays.settings import AUTOBLOCKS_REPLAYS_TRACE_ID_PARAM_NAME
+from demo_replays.settings import env
 
 app = Flask(__name__)
 
@@ -25,19 +26,17 @@ def main():
 
     # In production we generate a new trace id for each request,
     # but in a replay scenario we use the trace id passed in from the replay
-    trace_id = payload.get(settings.AUTOBLOCKS_REPLAYS_TRACE_ID_PARAM_NAME) or str(uuid.uuid4())
+    trace_id = payload.get(AUTOBLOCKS_REPLAYS_TRACE_ID_PARAM_NAME) or str(uuid.uuid4())
 
-    autoblocks = AutoblocksLogger(
-        ingestion_key=settings.AUTOBLOCKS_INGESTION_KEY,
-        trace_id=trace_id,
-        source="DEMO_REPLAYS",
+    autoblocks = AutoblocksTracer(
+        env.AUTOBLOCKS_INGESTION_KEY, trace_id=trace_id, properties=dict(source="DEMO_REPLAYS")
     )
-    autoblocks.send_event("request.payload", {"payload": payload})
+    autoblocks.send_event("request.payload", properties=dict(payload=payload))
 
     output = bot.get_response(autoblocks, query)
 
     response = {"output": output}
-    autoblocks.send_event("request.response", {"response": response})
+    autoblocks.send_event("request.response", properties=dict(response=response))
 
     return response
 

demo_replays/bot.py

@@ -1,7 +1,7 @@
 import requests
+from autoblocks.tracer import AutoblocksTracer
 
-from demo_replays.autoblocks_sdk import AutoblocksLogger
-from demo_replays.settings import settings
+from demo_replays.settings import env
 
 PROVIDER = "openai"
 MODEL = "gpt-3.5-turbo"
@@ -12,7 +12,7 @@
 )
 
 
-def get_response(autoblocks: AutoblocksLogger, query: str) -> str:
+def get_response(autoblocks: AutoblocksTracer, query: str) -> str:
     payload = {
         "model": MODEL,
         "messages": [
@@ -31,19 +31,19 @@ def get_response(autoblocks: AutoblocksLogger, query: str) -> str:
     req = requests.post(
         "https://api.openai.com/v1/chat/completions",
         json=payload,
-        headers={"Authorization": f"Bearer {settings.OPENAI_API_KEY}"},
+        headers={"Authorization": f"Bearer {env.OPENAI_API_KEY}"},
         timeout=10,
     )
     req.raise_for_status()
     response = req.json()
 
     autoblocks.send_event(
         "chat.completion",
-        {
-            "provider": PROVIDER,
-            "payload": payload,
-            "response": response,
-        },
+        properties=dict(
+            provider=PROVIDER,
+            payload=payload,
+            response=response,
+        ),
     )
 
     return response["choices"][0]["message"]["content"]

demo_replays/replay.py

@@ -1,10 +1,10 @@
 import argparse
-from datetime import datetime
 
 import requests
+from autoblocks.replays import replay_events_from_view
 
-from demo_replays.autoblocks_sdk import replay_events_from_view
-from demo_replays.settings import settings
+from demo_replays.settings import AUTOBLOCKS_REPLAYS_TRACE_ID_PARAM_NAME
+from demo_replays.settings import env
 
 
 def main():
@@ -21,27 +21,19 @@ def main():
     )
     args = parser.parse_args()
 
-    replay_id = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
-
-    print("#" * 80)
-    print(f"Your replay id is {replay_id}")
-    print("#" * 80)
-    print()
-
     for event in replay_events_from_view(
-        api_key=settings.AUTOBLOCKS_API_KEY,
-        replay_id=replay_id,
+        api_key=env.AUTOBLOCKS_API_KEY,
         view_id=args.view_id,
         num_traces=args.num_traces,
     ):
-        if event["message"] == "request.payload":
+        if event.message == "request.payload":
             print(f"Replaying event {event}")
 
             # The original payload
-            payload = event["properties"]["payload"]
+            payload = event.properties["payload"]
 
             # Modify the payload to pass in the replay trace id
-            payload[settings.AUTOBLOCKS_REPLAYS_TRACE_ID_PARAM_NAME] = event["traceId"]
+            payload[AUTOBLOCKS_REPLAYS_TRACE_ID_PARAM_NAME] = event.trace_id
 
             # Replay the request
             requests.post("http://localhost:5000", json=payload)

demo_replays/settings.py

@@ -1,18 +1,19 @@
 from pydantic_settings import BaseSettings
 
+# A hidden param that is used to override the trace id that would usually
+# be randomly generated for each request with the trace id of the event
+# that is being replayed
+AUTOBLOCKS_REPLAYS_TRACE_ID_PARAM_NAME = "__autoblocks_replay_trace_id"
 
-class Settings(BaseSettings):
+
+# Environment variables
+class Env(BaseSettings):
     # Autoblocks secrets
     AUTOBLOCKS_API_KEY: str = ""
     AUTOBLOCKS_INGESTION_KEY: str = ""
 
-    # A hidden param that is used to override the trace id that would usually
-    # be randomly generated for each request with the trace id of the event
-    # that is being replayed
-    AUTOBLOCKS_REPLAYS_TRACE_ID_PARAM_NAME: str = "__autoblocks_replay_trace_id"
-
     # OpenAI secrets
     OPENAI_API_KEY: str = ""
 
 
-settings = Settings()
+env = Env()