AgentInsight Python SDK provides a Python client for the AgentInsight platform, supporting LLM application observability, tracing, evaluation, and prompt management. The SDK is built entirely on the OpenTelemetry standard and offers three integration methods: out-of-the-box OpenAI / LangChain auto-instrumentation (recommended), the @observe decorator, and direct SDK calls.
- 🤖 LLM Auto-Instrumentation (Recommended) — Out-of-the-box OpenAI and LangChain integrations; change a single import line to automatically trace every LLM call, including prompts/completions, token usage, and costs — no business-logic changes required
- 🔍 Automatic Tracing — Use the
@observedecorator to automatically trace function calls, capturing inputs/outputs, latency, and errors - 📊 Scoring & Evaluation — Built-in evaluation framework and batch evaluation system, supporting NUMERIC / BOOLEAN / CATEGORICAL scores
- 🔄 Context Propagation — Cross-service context propagation based on OpenTelemetry Baggage
- 📝 Prompt Management — Version-controlled prompt management and template compilation
- 📁 Datasets & Experiments — Dataset management and A/B experiment framework
- 🛡️ Multi-Project Isolation — Client isolation across projects via
ContextVar, preventing trace data leakage between projects - ⚡ High Performance — Batch span export, background threads for media upload and score ingestion
pip install agentinsightFor OpenAI integration, install the OpenAI package additionally:
pip install agentinsight openaiFor LangChain integration, install the LangChain packages additionally:
pip install agentinsight 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()The fastest way to gain full LLM observability. With a single import change, AgentInsight automatically instruments every API call — capturing prompts/completions, token usage, cost, latency, and errors — without touching your business logic.
OpenAI — just swap the import:
- import openai
+ from agentinsight.openai import openaifrom 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)LangChain — register the callback handler:
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)See the OpenAI Integration and LangChain Integration sections for full details.
The @observe decorator is the simplest way to add tracing to your own functions, automatically capturing 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")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?")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_generation(
name="generate-response",
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={"input": 10, "output": 50},
cost_details={"input": 0.001, "output": 0.0023},
)
client.flush()
⚠️ Important:usage_detailsandcost_detailsKey NamingAgentInsight server follows OpenTelemetry GenAI semantic conventions. Always use
"input"/"output"/"total"as the dictionary keys for bothusage_detailsandcost_details. Other key names (such asprompt_tokens,completion_tokens,total_cost,input_cost,output_cost) will be stored but will NOT be recognized by the server's cost calculation and analytics features.
usage_details:{"input": <int>, "output": <int>, "total": <int>}(token counts)cost_details:{"input": <float>, "output": <float>, "total": <float>}(monetary cost)- The server automatically computes
total = input + outputwhentotalis omitted.- The SDK's OpenAI/LangChain auto-instrumentation already follows this convention; manual
update()calls must follow it too.
Add scores to any span, supporting NUMERIC, BOOLEAN, and CATEGORICAL types:
from agentinsight import observe
@observe()
def my_function(query: str) -> str:
return f"Processed: {query}"
result = my_function("Hello")
from agentinsight import get_client
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="NUMERIC")
span.score(name="is_valid", value=True, data_type="BOOLEAN")
span.score(name="sentiment", value="positive", data_type="CATEGORICAL")
client.flush()Automatically trace all OpenAI API calls by changing a single import line:
- import openai
+ from agentinsight.openai import openaiFull 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 automatically traces:
- All prompts and completions (supports streaming, async, and function calling)
- Request latency
- API errors
- Token usage and costs
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)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()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 provides a built-in experiment and evaluation framework:
from agentinsight import AgentInsight, Evaluation
client = AgentInsight()
def my_task(*, input, **kwargs):
return f"Processed: {input}"
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}")The SDK supports 9 observation types, each corresponding to different span semantics:
| Type | Class | Usage |
|---|---|---|
span |
AgentInsightSpan |
General workflow step |
generation |
AgentInsightGeneration |
LLM call |
agent |
AgentInsightAgent |
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 server URL | https://agent.goldebridge.com |
AGENTINSIGHT_TRACING_ENABLED |
Enable tracing | True |
AGENTINSIGHT_TRACING_ENVIRONMENT |
Environment identifier | default |
AGENTINSIGHT_RELEASE |
Release version identifier | — |
AGENTINSIGHT_FLUSH_AT |
Batch span export threshold | 512 |
AGENTINSIGHT_FLUSH_INTERVAL |
Batch export interval (seconds) | 5 |
AGENTINSIGHT_SAMPLE_RATE |
Sample rate (0.0 - 1.0) | 1.0 |
AGENTINSIGHT_TIMEOUT |
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 |
Decorator I/O capture switch | True |
AGENTINSIGHT_PROMPT_CACHE_DEFAULT_TTL_SECONDS |
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
For full documentation, please refer to the AgentInsight Official Documentation.
Please refer to CONTRIBUTING.md for contribution guidelines.
Please refer to SECURITY.md for security policy and vulnerability reporting.
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.
AgentInsight Python SDK 为 AgentInsight 平台提供 Python 客户端,支持 LLM 应用的可观测性、追踪、评估和 Prompt 管理。SDK 完全构建在 OpenTelemetry 标准之上,提供三种集成方式:OpenAI / LangChain 开箱即用的自动埋点(推荐)、@observe 装饰器、以及直接 SDK 调用。
- 🤖 LLM 自动埋点(推荐) — 开箱即用的 OpenAI 和 LangChain 集成,仅需修改一行 import 即可自动追踪所有 LLM 调用,包含 prompts/completions、token 用量和成本,无需改动业务代码
- 🔍 自动追踪 — 使用
@observe装饰器自动追踪函数调用,捕获输入/输出、耗时和错误 - 📊 评分与评估 — 内置评估框架和批量评估系统,支持 NUMERIC / BOOLEAN / CATEGORICAL 评分
- 🔄 上下文传播 — 基于 OpenTelemetry Baggage 的跨服务上下文传播
- 📝 Prompt 管理 — 版本控制的 Prompt 管理和模板编译
- 📁 数据集与实验 — 数据集管理和 A/B 实验框架
- 🛡️ 多项目隔离 — 通过
ContextVar实现多项目场景下的客户端隔离,防止 trace 数据跨项目泄漏 - ⚡ 高性能 — 批量发送 span、后台线程处理媒体上传和评分摄入
pip install agentinsightOpenAI 集成需要额外安装 OpenAI 包:
pip install agentinsight openaiLangChain 集成需要额外安装 LangChain 包:
pip install agentinsight langchain langchain-openaiimport agentinsight
agentinsight.init(
public_key="pk-...",
secret_key="sk-...",
base_url="https://agent.goldebridge.com",
)或者通过环境变量配置:
export AGENTINSIGHT_PUBLIC_KEY="pk-..."
export AGENTINSIGHT_SECRET_KEY="sk-..."
export AGENTINSIGHT_BASE_URL="https://agent.goldebridge.com"from agentinsight import AgentInsight
client = AgentInsight()获得完整 LLM 可观测性的最快方式。只需修改一行 import,AgentInsight 即可自动埋点每一次 API 调用 —— 捕获 prompts/completions、token 用量、成本、延迟和错误 —— 完全无需改动业务代码。
OpenAI —— 仅需替换 import:
- import openai
+ from agentinsight.openai import openaifrom 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)LangChain —— 注册回调处理器:
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)完整细节请参阅 OpenAI 集成 和 LangChain 集成 小节。
@observe 装饰器是为自有函数添加追踪的最简方式,自动捕获函数的输入、输出和耗时:
from agentinsight import observe
@observe(name="my-function")
def my_function(query: str) -> str:
return f"Processed: {query}"
result = my_function("Hello, AgentInsight!")装饰器支持嵌套调用,自动建立父子关系:
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 用量:
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:
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_generation(
name="generate-response",
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={"input": 10, "output": 50},
cost_details={"input": 0.001, "output": 0.0023},
)
client.flush()
⚠️ 重要提示:usage_details和cost_details键名约定AgentInsight 服务端遵循 OpenTelemetry GenAI 语义约定。务必使用
"input"/"output"/"total"作为字典键名,同时适用于usage_details和cost_details。其他键名(如prompt_tokens、completion_tokens、total_cost、input_cost、output_cost)虽会被存储,但不会被服务端的成本计算和分析功能识别。
usage_details:{"input": <int>, "output": <int>, "total": <int>}(token 数量)cost_details:{"input": <float>, "output": <float>, "total": <float>}(货币成本)- 当省略
total时,服务端会自动计算total = input + output。- SDK 的 OpenAI/LangChain 自动埋点已遵循此约定;手动
update()调用也必须遵循。
为任何 span 添加评分,支持 NUMERIC、BOOLEAN 和 CATEGORICAL 类型:
from agentinsight import observe
@observe()
def my_function(query: str) -> str:
return f"Processed: {query}"
result = my_function("Hello")
from agentinsight import get_client
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="NUMERIC")
span.score(name="is_valid", value=True, data_type="BOOLEAN")
span.score(name="sentiment", value="positive", data_type="CATEGORICAL")
client.flush()只需修改一行 import,即可自动追踪所有 OpenAI API 调用:
- import openai
+ from agentinsight.openai import openai完整示例:
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 自动追踪:
- 所有 prompts 和 completions(支持 streaming、async 和 function calling)
- 请求延迟
- API 错误
- Token 用量和成本
使用 CallbackHandler 追踪 LangChain 链的执行:
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:
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 头部):
from agentinsight import propagate_attributes
with propagate_attributes(
user_id="user_123",
session_id="session_abc",
as_baggage=True,
):
passAgentInsight 提供内置的实验和评估框架:
from agentinsight import AgentInsight, Evaluation
client = AgentInsight()
def my_task(*, input, **kwargs):
return f"Processed: {input}"
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 语义:
| 类型 | 类名 | 用途 |
|---|---|---|
span |
AgentInsightSpan |
通用工作流步骤 |
generation |
AgentInsightGeneration |
LLM 调用 |
agent |
AgentInsightAgent |
Agent 执行 |
tool |
AgentInsightTool |
工具调用 |
chain |
AgentInsightChain |
链式调用 |
embedding |
AgentInsightEmbedding |
向量嵌入 |
evaluator |
AgentInsightEvaluator |
评估器 |
retriever |
AgentInsightRetriever |
检索器 |
guardrail |
AgentInsightGuardrail |
安全护栏 |
| 变量 | 说明 | 默认值 |
|---|---|---|
AGENTINSIGHT_PUBLIC_KEY |
项目公钥(必填) | — |
AGENTINSIGHT_SECRET_KEY |
项目私钥(必填) | — |
AGENTINSIGHT_BASE_URL |
AgentInsight 服务地址 | https://agent.goldebridge.com |
AGENTINSIGHT_TRACING_ENABLED |
是否启用 tracing | True |
AGENTINSIGHT_TRACING_ENVIRONMENT |
环境标识 | default |
AGENTINSIGHT_RELEASE |
发布版本标识 | — |
AGENTINSIGHT_FLUSH_AT |
批量发送 span 的阈值 | 512 |
AGENTINSIGHT_FLUSH_INTERVAL |
批量发送间隔(秒) | 5 |
AGENTINSIGHT_SAMPLE_RATE |
采样率(0.0 - 1.0) | 1.0 |
AGENTINSIGHT_TIMEOUT |
HTTP 请求超时(秒) | 5 |
AGENTINSIGHT_DEBUG |
调试模式 | False |
AGENTINSIGHT_MEDIA_UPLOAD_ENABLED |
是否启用媒体上传 | True |
AGENTINSIGHT_MEDIA_UPLOAD_THREAD_COUNT |
媒体上传线程数 | 1 |
AGENTINSIGHT_OBSERVE_DECORATOR_IO_CAPTURE_ENABLED |
装饰器 I/O 捕获开关 | True |
AGENTINSIGHT_PROMPT_CACHE_DEFAULT_TTL_SECONDS |
Prompt 缓存 TTL(秒) | 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 官方文档。
请参阅 CONTRIBUTING.md 了解贡献指南。
请参阅 SECURITY.md 了解安全策略和漏洞报告方式。
本项目基于 Langfuse Python SDK 构建并演化而来,感谢 Langfuse 团队的出色工作。Langfuse Python SDK 采用 MIT 许可证发布。