AgentInsight Python SDK — 用于将 AI 应用可观测性数据发送到 AgentInsight 平台的官方 Python 客户端库。
The official Python client library for sending AI application observability data to the AgentInsight platform.
基于 OpenTelemetry 协议，支持自动追踪 AI 模型调用、评分评估、Prompt 管理等功能。提供三种集成方式：直接 SDK 调用、@observe 装饰器、以及 OpenAI / LangChain 开箱即用的集成。
Built on the OpenTelemetry protocol, supporting automatic AI model call tracing, scoring & evaluation, prompt management, and more. Offers three integration methods: direct SDK calls, the @observe decorator, and out-of-the-box OpenAI / LangChain integrations.
- 🔍 自动追踪 / Automatic Tracing — 使用
@observe装饰器自动追踪函数调用，捕获输入/输出、耗时和错误 / Use the@observedecorator to automatically trace function calls, capturing inputs/outputs, latency, and errors - 🤖 LLM 集成 / LLM Integrations — 开箱即用的 OpenAI 和 LangChain 集成，仅需修改一行 import / Out-of-the-box OpenAI and LangChain integrations, requiring only a single import change
- 📊 评分与评估 / Scoring & Evaluation — 内置评估框架和批量评估系统，支持 NUMERIC / BOOLEAN / CATEGORICAL 评分 / Built-in evaluation framework and batch evaluation system, supporting NUMERIC / BOOLEAN / CATEGORICAL scores
- 🔄 上下文传播 / Context Propagation — 基于 OpenTelemetry Baggage 的跨服务上下文传播 / Cross-service context propagation based on OpenTelemetry Baggage
- 📝 Prompt 管理 / Prompt Management — 版本控制的 Prompt 管理和模板编译 / Version-controlled prompt management and template compilation
- 📁 数据集与实验 / Datasets & Experiments — 数据集管理和 A/B 实验框架 / Dataset management and A/B experiment framework
- 🛡️ 多项目隔离 / Multi-Project Isolation — 通过
ContextVar实现多项目场景下的客户端隔离，防止 trace 数据跨项目泄漏 / Client isolation across projects viaContextVar, preventing trace data leakage between projects - ⚡ 高性能 / High Performance — 批量发送 span、后台线程处理媒体上传和评分摄入 / Batch span export, background threads for media upload and score ingestion
pip install agentinsight-sdkOpenAI 集成需要额外安装 OpenAI 包 / For OpenAI integration, install the OpenAI package additionally:
pip install agentinsight-sdk openaiLangChain 集成需要额外安装 LangChain 包 / For LangChain integration, install the LangChain packages additionally:
pip install agentinsight-sdk langchain langchain-openaiimport agentinsight
agentinsight.init(
public_key="pk-...",
secret_key="sk-...",
base_url="https://agent.goldebridge.com",
)或者通过环境变量配置 / Or configure via environment variables:
export AGENTINSIGHT_PUBLIC_KEY="pk-..."
export AGENTINSIGHT_SECRET_KEY="sk-..."
export AGENTINSIGHT_BASE_URL="https://agent.goldebridge.com"from agentinsight import AgentInsight
client = AgentInsight()@observe 装饰器是最简单的追踪方式，自动捕获函数的输入、输出和耗时 / The @observe decorator is the simplest way to add tracing, automatically capturing function inputs, outputs, and latency:
from agentinsight import observe
@observe(name="my-function")
def my_function(query: str) -> str:
return f"Processed: {query}"
result = my_function("Hello, AgentInsight!")装饰器支持嵌套调用，自动建立父子关系 / The decorator supports nested calls and automatically establishes parent-child relationships:
from agentinsight import observe
@observe(as_type="agent")
def run_agent(query: str) -> str:
plan = plan_task(query)
result = execute_task(plan)
return result
@observe(as_type="chain")
def plan_task(query: str) -> str:
return f"Plan for: {query}"
@observe(as_type="tool")
def execute_task(plan: str) -> str:
return f"Executed: {plan}"
run_agent("Build a web app")使用 as_type="generation" 标记 LLM 调用，记录模型参数和 token 用量 / Use as_type="generation" to mark LLM calls, recording model parameters and token usage:
from agentinsight import observe
@observe(as_type="generation")
def call_llm(prompt: str) -> str:
# Your LLM call here
return "LLM response"
result = call_llm("What is AI?")也可以使用低级 API 手动管理 span / You can also use the low-level API to manually manage spans:
from agentinsight import AgentInsight
client = AgentInsight()
with client.start_as_current_observation(
name="process-query",
as_type="span",
) as span:
with span.start_as_current_observation(
name="generate-response",
as_type="generation",
model="gpt-4",
input={"query": "Tell me about AI"},
model_parameters={"temperature": 0.7, "max_tokens": 500},
) as generation:
response = "AI is a field of computer science..."
generation.update(
output=response,
usage_details={"prompt_tokens": 10, "completion_tokens": 50},
cost_details={"total_cost": 0.0023},
)
client.flush()为任何 span 添加评分，支持 NUMERIC、BOOLEAN 和 CATEGORICAL 类型 / Add scores to any span, supporting NUMERIC, BOOLEAN, and CATEGORICAL types:
from agentinsight import observe, get_client
from agentinsight.api.commons.types.score_data_type import ScoreDataType
@observe()
def my_function(query: str) -> str:
return f"Processed: {query}"
result = my_function("Hello")
client = get_client()
with client.start_as_current_observation(name="scored-task", as_type="span") as span:
span.score(name="relevance", value=0.95, data_type=ScoreDataType.NUMERIC)
span.score(name="is_valid", value=1.0, data_type=ScoreDataType.BOOLEAN)
span.score(name="sentiment", value="positive", data_type=ScoreDataType.CATEGORICAL)
client.flush()只需修改一行 import，即可自动追踪所有 OpenAI API 调用 / Automatically trace all OpenAI API calls by changing a single import line:
- import openai
+ from agentinsight.openai import openai完整示例 / Full example:
from agentinsight.openai import openai
client = openai.OpenAI()
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "What is AI?"},
],
)
print(response.choices[0].message.content)AgentInsight 自动追踪 / AgentInsight automatically traces:
- 所有 prompts 和 completions（支持 streaming、async 和 function calling）/ All prompts and completions (supports streaming, async, and function calling)
- 请求延迟 / Request latency
- API 错误 / API errors
- Token 用量和成本 / Token usage and costs
使用 CallbackHandler 追踪 LangChain 链的执行 / Use CallbackHandler to trace LangChain chain execution:
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from agentinsight.langchain import CallbackHandler
handler = CallbackHandler()
llm = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant."),
("human", "{input}"),
])
chain = prompt | llm
result = chain.invoke(
{"input": "What is AI?"},
config={"callbacks": [handler]},
)
print(result.content)使用 propagate_attributes 在 trace 内设置用户级、会话级属性，自动传播到所有子 span / Use propagate_attributes to set user-level and session-level attributes within a trace, automatically propagating them to all child spans:
from agentinsight import AgentInsight, propagate_attributes
client = AgentInsight()
with client.start_as_current_observation(name="user-workflow", as_type="span") as span:
with propagate_attributes(
user_id="user_123",
session_id="session_abc",
metadata={"environment": "production", "variant": "a"},
tags=["production", "v2"],
):
with client.start_as_current_observation(name="llm-call", as_type="generation") as gen:
pass
client.flush()跨服务传播（通过 HTTP 头部）/ Cross-service propagation (via HTTP headers):
from agentinsight import propagate_attributes
with propagate_attributes(
user_id="user_123",
session_id="session_abc",
as_baggage=True,
):
passAgentInsight 提供内置的实验和评估框架 / AgentInsight provides a built-in experiment and evaluation framework:
from agentinsight import AgentInsight, Evaluation
client = AgentInsight()
def my_task(*, item, **kwargs):
input_data = item["input"] if isinstance(item, dict) else item.input
return f"Processed: {input_data}"
def accuracy_evaluator(*, input, output, expected_output=None, **kwargs):
if not expected_output:
return Evaluation(name="accuracy", value=0, comment="No expected output")
is_correct = output.strip().lower() == expected_output.strip().lower()
return Evaluation(
name="accuracy",
value=1.0 if is_correct else 0.0,
comment="Correct" if is_correct else "Incorrect",
)
result = client.run_experiment(
name="my-experiment",
data=[
{"input": "What is 2+2?", "expected_output": "4"},
{"input": "What is 3+3?", "expected_output": "6"},
],
task=my_task,
evaluators=[accuracy_evaluator],
)
for item_result in result.item_results:
print(f"Input: {item_result.item}")
print(f"Output: {item_result.output}")
for evaluation in item_result.evaluations:
print(f" {evaluation.name}: {evaluation.value}")SDK 支持 9 种观察类型，对应不同的 span 语义 / The SDK supports 9 observation types, each corresponding to different span semantics:
| 类型 / Type | 类名 / Class | 用途 / Usage |
|---|---|---|
span |
AgentInsightSpan |
通用工作流步骤 / General workflow step |
generation |
AgentInsightGeneration |
LLM 调用 / LLM call |
agent |
AgentInsightAgent |
Agent 执行 / Agent execution |
tool |
AgentInsightTool |
工具调用 / Tool call |
chain |
AgentInsightChain |
链式调用 / Chain call |
embedding |
AgentInsightEmbedding |
向量嵌入 / Vector embedding |
evaluator |
AgentInsightEvaluator |
评估器 / Evaluator |
retriever |
AgentInsightRetriever |
检索器 / Retriever |
guardrail |
AgentInsightGuardrail |
安全护栏 / Safety guardrail |
| 变量 / Variable | 说明 / Description | 默认值 / Default |
|---|---|---|
AGENTINSIGHT_PUBLIC_KEY |
项目公钥（必填）/ Project public key (required) | — |
AGENTINSIGHT_SECRET_KEY |
项目私钥（必填）/ Project secret key (required) | — |
AGENTINSIGHT_BASE_URL |
AgentInsight 服务地址 / AgentInsight server URL | https://agent.goldebridge.com |
AGENTINSIGHT_TRACING_ENABLED |
是否启用 tracing / Enable tracing | True |
AGENTINSIGHT_TRACING_ENVIRONMENT |
环境标识 / Environment identifier | default |
AGENTINSIGHT_RELEASE |
发布版本标识 / Release version identifier | — |
AGENTINSIGHT_FLUSH_AT |
批量发送 span 的阈值 / Batch span export threshold | 512 |
AGENTINSIGHT_FLUSH_INTERVAL |
批量发送间隔（秒）/ Batch export interval (seconds) | 5 |
AGENTINSIGHT_SAMPLE_RATE |
采样率（0.0 - 1.0）/ Sample rate (0.0 - 1.0) | 1.0 |
AGENTINSIGHT_TIMEOUT |
HTTP 请求超时（秒）/ HTTP request timeout (seconds) | 5 |
AGENTINSIGHT_DEBUG |
调试模式 / Debug mode | False |
AGENTINSIGHT_MEDIA_UPLOAD_ENABLED |
是否启用媒体上传 / Enable media upload | True |
AGENTINSIGHT_MEDIA_UPLOAD_THREAD_COUNT |
媒体上传线程数 / Media upload thread count | 1 |
AGENTINSIGHT_OBSERVE_DECORATOR_IO_CAPTURE_ENABLED |
装饰器 I/O 捕获开关 / Decorator I/O capture switch | True |
AGENTINSIGHT_PROMPT_CACHE_DEFAULT_TTL_SECONDS |
Prompt 缓存 TTL（秒）/ Prompt cache TTL (seconds) | 60 |
from agentinsight import AgentInsight
client = AgentInsight(
public_key="pk-...",
secret_key="sk-...",
base_url="https://agent.goldebridge.com",
timeout=5,
debug=False,
tracing_enabled=True,
flush_at=512,
flush_interval=5.0,
environment="production",
release="1.0.0",
sample_rate=1.0,
media_upload_thread_count=1,
)- Python >= 3.10, < 4.0
完整文档请参阅 AgentInsight 官方文档。
For full documentation, please refer to the AgentInsight Official Documentation.
- 快速开始指南 / Quick Start Guide
- OpenAI 集成 / OpenAI Integration
- LangChain 集成 / LangChain Integration
- 评估系统 / Evaluation System
- Prompt 管理 / Prompt Management
请参阅 CONTRIBUTING.md 了解贡献指南。
Please refer to CONTRIBUTING.md for contribution guidelines.
请参阅 SECURITY.md 了解安全策略和漏洞报告方式。
Please refer to SECURITY.md for security policy and vulnerability reporting.
本项目基于 Langfuse Python SDK 构建并演化而来，感谢 Langfuse 团队的出色工作。Langfuse Python SDK 采用 MIT 许可证发布。
This project is built upon and evolved from the Langfuse Python SDK. We thank the Langfuse team for their excellent work. The Langfuse Python SDK is released under the MIT License.