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
|
import json
import logging
import os
from typing import Optional
import requests
from agbenchmark.__main__ import BENCHMARK_START_TIME
from agbenchmark.agent_interface import HELICONE_GRAPHQL_LOGS
logger = logging.getLogger(__name__)
def get_data_from_helicone(challenge: str) -> Optional[float]:
# Define the endpoint of your GraphQL server
url = "https://www.helicone.ai/api/graphql"
# Set the headers, usually you'd need to set the content type and possibly an authorization token
headers = {"authorization": f"Bearer {os.environ.get('HELICONE_API_KEY')}"}
# Define the query, variables, and operation name
query = """
query ExampleQuery($properties: [PropertyFilter!]){
aggregatedHeliconeRequest(properties: $properties) {
costUSD
}
}
"""
variables = {
"properties": [
{
"value": {"equals": os.environ.get("AGENT_NAME")},
"name": "agent",
},
{
"value": {"equals": BENCHMARK_START_TIME},
"name": "benchmark_start_time",
},
{"value": {"equals": challenge}, "name": "challenge"},
]
}
if HELICONE_GRAPHQL_LOGS:
logger.debug(f"Executing Helicone query:\n{query.strip()}")
logger.debug(f"Query variables:\n{json.dumps(variables, indent=4)}")
operation_name = "ExampleQuery"
data = {}
response = None
try:
response = requests.post(
url,
headers=headers,
json={
"query": query,
"variables": variables,
"operationName": operation_name,
},
)
data = response.json()
except requests.HTTPError as http_err:
logger.error(f"Helicone returned an HTTP error: {http_err}")
return None
except json.JSONDecodeError:
raw_response = response.text # type: ignore
logger.error(
f"Helicone returned an invalid JSON response: '''{raw_response}'''"
)
return None
except Exception as err:
logger.error(f"Error while trying to get data from Helicone: {err}")
return None
if data is None or data.get("data") is None:
logger.error("Invalid response received from Helicone: no data")
logger.error(f"Offending response: {response}")
return None
return (
data.get("data", {}).get("aggregatedHeliconeRequest", {}).get("costUSD", None)
)
|