Galileo AI 平台 Python SDK 完整参考,用于评估、监控和保护 GenAI 应用,适用于构建 Python 应用。
Galileo Python SDK.是一项面向实际任务的技能,主要用于Galileo Python SDK( galileo) 为GenAI 应用软件提供统一接口—— 使 GenAI AI 平台能够进行评价,可观察性,以及运行时的守护护栏。它将相关步骤、工具调用和结果整理方式集中到统一流程中,帮助使用者更快完成目标并减少重复操作。
实际使用前应先确认任务范围、数据来源、运行环境、必要权限和关键参数,再依据技能说明逐步执行;若输入条件不完整,应先补齐信息或采用保守配置,避免因错误假设导致结果偏离需求。
执行过程中需要关注工具调用是否成功、接口或依赖是否可用、输出格式是否符合预期,并对异常提示、缺失字段和边界情况进行处理;涉及批量任务时,还应保存进度,避免中断后重复操作。
Galileo Python SDK(galileo)为 Galileo AI 平台提供统一接口,支持大语言模型应用的评估(evaluation)、可观测性(observability)及运行时防护机制(runtime guardrails)。它支持 LLM 调用的自动追踪、自定义 Span 日志记录、评估实验(evaluation experiments),以及生产级防护能力。
在编写任何代码前,请先检查已安装版本,以选择正确的文档参考:
import importlib.metadata, importlib.util
galileo_ver = importlib.metadata.version("galileo") # 例如 "2.1.1"
pq_installed = importlib.util.find_spec("promptquality") is not None
pq_ver = importlib.metadata.version("promptquality") if pq_installed else None
print(f"galileo={galileo_ver}, promptquality={pq_ver}")
| 已安装的技术栈 | 应参考的文档 |
|---|---|
galileo >= 2.0(无论是否同时安装 promptquality 0.x) |
本文档 — GalileoLogger、@log、galileo_context |
galileo < 2.0 + promptquality >= 1.0 |
Promptquality 1.x Reference |
注意:
promptquality >= 1.0与galileo >= 2.0互不兼容 —— 它们依赖不同主版本的galileo-core。同时安装将导致依赖冲突。
其他参考文档:
EvaluateRun、Scorers、ScorersConfigurationpip install galileo
如需使用传统提示工程接口的评估功能:
pip install promptquality
如需启用运行时防护能力:
pip install galileo-protect
import 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 文件或 Shell 环境 GALILEO_API_KEY="your-api-key" # 必填 — 来自 Galileo 控制台 GALILEO_CONSOLE_URL="https://app.galileo.ai" # 控制台 URL(或私有部署地址) GALILEO_PROJECT="my-project" # 可选 — 默认项目名 GALILEO_LOG_STREAM="my-log-stream" # 可选 — 默认日志流名 GALILEO_LOGGING_DISABLED="false" # 可选 — 禁用日志记录
对于传统的 promptquality 包,请通过编程方式认证:
import 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")
导入 Galileo 封装的 OpenAI 客户端,即可自动追踪所有调用:
from galileo.openai import openai
client = openai.OpenAI()
response = client.chat.completions.create(
messages=[{"role": "user", "content": "Hello"}],
model="gpt-4o",
)
@log 装饰器使用 @log 为函数创建 Span。支持的 Span 类型包括:workflow、llm、retriever、tool。
from 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",
)
将日志记录作用域限定在特定代码块内,并在退出时自动 flush:
from galileo import galileo_context
with galileo_context(project="my-project", log_stream="my-log-stream"):
result = my_workflow()
print(result)
将捕获的追踪数据上传至 Galileo:
galileo_context.flush()
promptquality 运行实验import 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,
),
)
使用 GalileoLogger 记录用于评估的追踪数据:
from galileo import GalileoLogger
logger = GalileoLogger(project="my_project", log_stream="my_run")
eval_set = ["What are hallucinations?", "What are intrinsic hallucinations?"]
for input_text in eval_set:
output = llm.call(input_text)
logger.add_single_llm_span_trace(
input=input_text,
output=output,
model="gpt-4o",
)
logger.flush()
若使用
galileo < 2.0+promptquality >= 1.0技术栈,请使用EvaluateRun—— 参见 Promptquality 1.x Reference。
更多内容请参阅“高级评估模式”。
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") # 异步调用方式 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}nnQuestion: {question}"
response = client.chat.completions.create(
messages=[{"role": "user", "content": prompt}],
model="gpt-4o",
)
return response.choices[0].message.content
from galileo import log
@log(span_type="tool")
def math_operation(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 == "math_operation":
results.append(math_operation(action.input))
elif action.tool == "web_search":
results.append(web_search(action.input))
return synthesize(results)
GALILEO_API_KEY 和 GALILEO_CONSOLE_URL,切勿硬编码凭据。galileo_context.flush(),确保追踪数据及时上传;在 Web 服务中,应在每个请求处理器末尾执行 flush。with galileo_context(...))实现作用域受限的日志记录,并在退出时自动 flush。retriever、tool、llm、workflow),以充分利用 Galileo 的追踪可视化能力。flush() 调用包裹在 try/except 中,避免日志失败导致应用崩溃。from galileo.openai import openai),实现零配置的 OpenAI 调用自动追踪。