Documentation Index
Fetch the complete documentation index at: https://docs.helicone.ai/llms.txt
Use this file to discover all available pages before exploring further.
This integration method is maintained but no longer actively developed. For the best experience and latest features, use our new AI Gateway with unified API access to 100+ models. Proxy Integration
Python SDK
Create an account + Generate an API Key
Log into helicone or create an account. Once you have an account, you can generate an API key. Set API keys as environment variables
export HELICONE_API_KEY=<your API key>
export PROJECT_ID=<your Google Cloud project ID>
export LOCATION=<your location>
Install required packages
Ensure you have the necessary packages installed in your Python environment:pip install google-cloud-aiplatform
Import libraries
from vertexai.generative_models import GenerativeModel
import vertexai
import os
Initialize Vertex AI with Helicone
HELICONE_API_KEY = os.environ.get("HELICONE_API_KEY")
PROJECT_ID = os.environ.get("PROJECT_ID")
LOCATION = os.environ.get("LOCATION")
vertexai.init(
project=PROJECT_ID,
location=LOCATION,
api_endpoint="gateway.helicone.ai",
api_transport="rest", # Must be 'rest' or else it will not work
request_metadata=[
('helicone-target-url', f'https://{LOCATION}-aiplatform.googleapis.com'),
('helicone-auth', f'Bearer {HELICONE_API_KEY}')
]
)
Initialize the model and generate content
model = GenerativeModel("gemini-1.5-flash-001")
response = model.generate_content("Tell me a fun fact about space.")
print(response.text)
Manual Logging with Log Builder
For more control over logging, especially with asynchronous and streaming responses, you can use the Helicone Manual Logger:
Install additional required packages
pip install python-dotenv helicone
Create a log builder example
This example demonstrates using the Helicone log builder with both async and streaming responses:import os
import asyncio
from dotenv import load_dotenv
import vertexai
from vertexai.generative_models import GenerativeModel
from helicone_helpers import HeliconeManualLogger
import warnings
import sys
# Load environment variables
load_dotenv()
HELICONE_API_KEY = os.getenv("HELICONE_API_KEY")
PROJECT_ID = os.getenv("PROJECT_ID")
LOCATION = os.getenv("LOCATION")
helicone_logger = HeliconeManualLogger(api_key=HELICONE_API_KEY)
async def generate_content_async(model, prompt="Tell me a joke"):
generate_kwargs = {
"contents": [prompt],
}
log_builder = helicone_logger.new_builder(
request=generate_kwargs
)
log_builder.add_model("gemini-1.5-pro")
try:
response = await model.generate_content_async(**generate_kwargs)
log_builder.add_response(response.to_dict())
print(f"Response: {response.text}")
return response.text
except Exception as e:
print(f"Error: {e}")
return str(e)
finally:
await log_builder.send_log()
async def generate_content_streaming(model: GenerativeModel, prompt="Tell me a joke"):
generate_kwargs = {
"contents": [prompt],
"stream": True,
}
log_builder = helicone_logger.new_builder(
request=generate_kwargs
)
log_builder.add_model("gemini-1.5-pro")
try:
responses = await model.generate_content_async(**generate_kwargs)
full_response = []
print("Streaming response:")
async for response in responses:
log_builder.add_chunk(response.to_dict())
chunk = response.text
print(chunk, end="", flush=True)
full_response.append(chunk)
print("\n")
return "".join(full_response)
except Exception as e:
print(f"Error in streaming: {e}")
return str(e)
finally:
await log_builder.send_log()
async def main():
print("Vertex AI Gemini with Helicone Example")
vertexai.init(
project=PROJECT_ID,
location=LOCATION,
)
model = GenerativeModel(
"gemini-1.5-pro",
generation_config={"response_mime_type": "application/json"}
)
print("\n=== Regular Async Response ===")
await generate_content_async(model, "Tell me a joke about programming")
print("\n=== Streaming Response ===")
await generate_content_streaming(model, "Tell me a short 2 sentence story about a programmer")
if __name__ == "__main__":
if not sys.warnoptions:
warnings.filterwarnings("ignore", category=ResourceWarning)
asyncio.run(main())
sys.exit(0)
Create the HeliconeManualLogger helper
Create a file named helicone_helpers.py with the following content:import json
import os
import aiohttp
import requests
class HeliconeManualLogger:
def __init__(self, api_key=None):
self.api_key = api_key or os.environ.get("HELICONE_API_KEY")
if not self.api_key:
raise ValueError("Helicone API key is required")
self.base_url = "https://api.helicone.ai/v1/log"
def new_builder(self, request):
return LogBuilder(self.api_key, request)
class LogBuilder:
def __init__(self, api_key, request):
self.api_key = api_key
self.request = request
self.model = None
self.response = None
self.chunks = []
def add_model(self, model):
self.model = model
return self
def add_response(self, response):
self.response = response
return self
def add_chunk(self, chunk):
self.chunks.append(chunk)
return self
async def send_log(self):
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}"
}
payload = {
"provider": "vertex",
"model": self.model,
"request": self.request
}
if self.response:
payload["response"] = self.response
if self.chunks:
payload["response_chunks"] = self.chunks
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.helicone.ai/v1/log",
headers=headers,
json=payload
) as response:
if response.status != 200:
print(f"Failed to log to Helicone: {response.status}")
text = await response.text()
print(f"Response: {text}")
return response.status == 200
- Precise control over what gets logged
- Support for both async and streaming responses
- Ability to add custom properties and metadata
- Manual control over when logs are sent
This is particularly useful for complex applications where you need more control over the logging process.
