Loading...
Loading...
Complete reference for the Galileo AI platform Python SDK for evaluating, observing, and protecting GenAI applications. Use when building Python applications that need LLM evaluation, production observability, tracing, or runtime guardrails with Galileo.
npx skill4agent add gyanesh-m/skills galileo-python-sdkgalileopip install galileopip install promptqualitypip install galileo-protectimport os
from galileo import galileo_context
from galileo.openai import openai
galileo_context.init(project="my-project", log_stream="my-log-stream")
client = openai.OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
response = client.chat.completions.create(
messages=[{"role": "user", "content": "Explain quantum computing in one sentence."}],
model="gpt-4o",
)
print(response.choices[0].message.content)
galileo_context.flush()# .env file or shell environment
GALILEO_API_KEY="your-api-key" # Required — from Galileo console
GALILEO_CONSOLE_URL="https://app.galileo.ai" # Console URL (or self-hosted URL)
GALILEO_PROJECT="my-project" # Optional — default project
GALILEO_LOG_STREAM="my-log-stream" # Optional — default log stream
GALILEO_LOGGING_DISABLED="false" # Optional — disable loggingpromptqualityimport promptquality as pq
pq.login("https://app.galileo.ai")from galileo import galileo_context
galileo_context.init(project="my-project", log_stream="my-log-stream")from galileo.openai import openai
client = openai.OpenAI()
response = client.chat.completions.create(
messages=[{"role": "user", "content": "Hello"}],
model="gpt-4o",
)@log@logworkflowllmretrievertoolfrom galileo import log
@log
def my_workflow():
result = call_openai()
return result
@log(span_type="retriever")
def retrieve_documents(query: str):
docs = vector_store.search(query)
return docs
@log(span_type="tool")
def search_web(query: str):
return web_api.search(query)from galileo import log
@log
def agent_pipeline(user_input: str):
context = retrieve_documents(user_input)
tool_result = search_web(user_input)
response = generate_response(user_input, context, tool_result)
return response
@log(span_type="retriever")
def retrieve_documents(query: str):
return ["doc1", "doc2"]
@log(span_type="tool")
def search_web(query: str):
return "search result"
@log
def generate_response(query: str, context: list, tool_result: str):
client = openai.OpenAI()
return client.chat.completions.create(
messages=[{"role": "user", "content": query}],
model="gpt-4o",
)from galileo import galileo_context
with galileo_context(project="my-project", log_stream="my-log-stream"):
result = my_workflow()
print(result)galileo_context.flush()promptqualityimport promptquality as pq
pq.login("https://app.galileo.ai")
template = "Explain {{topic}} to me like I'm a 5 year old"
data = {"topic": ["Quantum Physics", "Politics", "Large Language Models"]}
pq.run(
project_name="my-first-project",
template=template,
dataset=data,
settings=pq.Settings(
model_alias="ChatGPT (16K context)",
temperature=0.8,
max_tokens=400,
),
)from promptquality import EvaluateRun
import promptquality as pq
pq.login()
metrics = [pq.Scorers.context_adherence_plus, pq.Scorers.prompt_injection]
evaluate_run = EvaluateRun(
run_name="my_run",
project_name="my_project",
scorers=metrics,
)
eval_set = ["What are hallucinations?", "What are intrinsic hallucinations?"]
for input_text in eval_set:
output = llm.call(input_text)
evaluate_run.add_single_step_workflow(
input=input_text,
output=output,
model="gpt-4o",
)
evaluate_run.finish()from galileo import GalileoMetrics
from galileo.stages import create_protect_stage
from galileo_core.schemas.protect.rule import Rule, RuleOperator
from galileo_core.schemas.protect.ruleset import Ruleset
from galileo_core.schemas.protect.stage import StageType
rule = Rule(
metric=GalileoMetrics.input_toxicity,
operator=RuleOperator.gt,
target_value=0.1,
)
ruleset = Ruleset(rules=[rule])
stage = create_protect_stage(
name="toxicity-guard",
stage_type=StageType.central,
prioritized_rulesets=[ruleset],
description="Block toxic input.",
)from galileo.protect import invoke_protect, ainvoke_protect
from galileo_core.schemas.protect.payload import Payload
payload = Payload(input="User message to check.")
response = invoke_protect(payload=payload, stage_name="toxicity-guard")
# Async variant
response = await ainvoke_protect(payload=payload, stage_name="toxicity-guard")from galileo import log
from galileo.openai import openai
client = openai.OpenAI()
@log
def chat(messages: list):
response = client.chat.completions.create(
messages=messages,
model="gpt-4o",
)
return response.choices[0].message.content
messages = []
messages.append({"role": "user", "content": "What is RAG?"})
reply = chat(messages)
messages.append({"role": "assistant", "content": reply})
messages.append({"role": "user", "content": "How do I implement it?"})
reply = chat(messages)from galileo import log
from galileo.openai import openai
client = openai.OpenAI()
@log(span_type="retriever")
def retrieve(query: str):
results = vector_db.similarity_search(query, k=5)
return [doc.page_content for doc in results]
@log
def rag_pipeline(question: str):
context = retrieve(question)
prompt = f"Context: {context}\n\nQuestion: {question}"
response = client.chat.completions.create(
messages=[{"role": "user", "content": prompt}],
model="gpt-4o",
)
return response.choices[0].message.contentfrom galileo import log
@log(span_type="tool")
def calculator(a: float, b: float, op: str) -> str:
if op == "add":
return str(a + b)
elif op == "multiply":
return str(a * b)
raise ValueError(f"Unknown op: {op}")
@log(span_type="tool")
def web_search(query: str):
return search_api.query(query)
@log
def agent(user_input: str):
plan = plan_actions(user_input)
results = []
for action in plan:
if action.tool == "calculator":
results.append(calculator(action.input))
elif action.tool == "web_search":
results.append(web_search(action.input))
return synthesize(results)GALILEO_API_KEYGALILEO_CONSOLE_URLgalileo_context.flush()with galileo_context(...)retrievertoolllmworkflowflush()from galileo.openai import openai