mingx-v2 0.1.2

Mingx tracing SDK - OpenTelemetry-based tracing compatible with any OTLP backend

Mingx Tracing SDK

Mingx 跟踪 SDK：基于 OpenTelemetry 的 Python 跟踪库，数据通过标准 OTLP 导出，可对接任意兼容 OpenTelemetry 的接收端（如 Jaeger、Grafana Tempo、Langfuse OTel 端点、自建 Collector 等）。

API 设计参考 Langfuse Python SDK，便于迁移与习惯统一。

要求

• Python 3.10+
• 推荐使用 uv 管理依赖与环境

安装

# 使用 uv（推荐）
uv add mingx

# 或从本地源码安装
cd mingx-v2
uv sync
uv pip install -e .

配置

通过环境变量配置：

• 启用/禁用：MINGX_TRACING_ENABLED（默认 true，设为 false 或 0 关闭跟踪）
• 导出端点：使用 OpenTelemetry 标准变量，SDK 不绑定任何具体后端：
  • OTEL_EXPORTER_OTLP_TRACES_ENDPOINT：轨迹导出地址，例如 http://localhost:4318/v1/traces
  • OTEL_EXPORTER_OTLP_TRACES_HEADERS：可选，请求头，如 Authorization=Bearer xxx
  • OTEL_SERVICE_NAME：可选，服务名，用于 Resource（默认 mingx）

也可在代码中指定接收端与认证（优先于环境变量）：

from mingx import get_client

# 首次调用时传入，用于创建默认 client
mingx = get_client(
    endpoint="https://ingest.example.com/v1/traces",
    token="your-api-token",   # 或 "username:password" 形式（含冒号则按用户名:密码做 Basic 认证）
)

或直接实例化 MingxClient(endpoint=..., token=...)。未设置 endpoint 时，会使用环境变量或需传入自定义 span_exporter。

使用

1. Context manager（推荐）

from mingx import get_client

mingx = get_client()

with mingx.start_as_current_span(name="process-request") as span:
    # 业务逻辑
    result = do_work()
    span.update(output=result)

# 嵌套：generation 用于 LLM 调用
with mingx.start_as_current_span(name="handle-query") as span:
    with mingx.start_as_current_observation(
        name="llm-call",
        as_type="generation",
        model="gpt-4",
        input={"prompt": "Hello"},
    ) as gen:
        response = call_llm()
        gen.update(
            output=response,
            usage_details={"prompt_tokens": 10, "completion_tokens": 20},
            cost_details={"total_cost": 0.001},
        )
    span.update(output=response)

# 进程退出时会自动 flush；若需在退出前确保发出可手动调用
mingx.flush()

2. 装饰器

from mingx import observe, get_client

@observe(name="my-operation", as_type="span")
def process(data: str) -> str:
    return data.upper()

@observe(as_type="generation", name="answer")
async def generate_answer(query: str) -> str:
    # 自动记录入参、返回值与异常
    return await llm.achat(query)

3. 更新当前 trace / span

mingx = get_client()
with mingx.start_as_current_span(name="request") as span:
    mingx.update_current_trace(user_id="user-1", session_id="sess-1")
    # ...
    mingx.update_current_span(output=result)

4. 集成：区分每次程序调用

每次请求/任务对应一条 trace。在根 span 上设置 user_id、session_id，并把业务侧的 request_id 放进 metadata，便于在 OTLP 后端按请求或用户筛选；根 span 的 span.trace_id 可写响应头或日志，用于在观测平台精确定位该次调用。

mingx = get_client()
with mingx.start_as_current_span(name="request", input={"query": q}) as root:
    mingx.update_current_trace(
        name="my-api",
        user_id=user_id,
        session_id=session_id,
        metadata={"request_id": request_id},
    )
    # ... 业务逻辑，所有子 span 都在同一条 trace 下
    root.update(output=result)
# 可选：把 root.trace_id 写入响应或日志，便于和观测平台关联

完整示例见 examples/integration_example.py。

5. 事件（瞬时点）

mingx.create_event(name="button-click", metadata={"id": "submit"})

6. LangChain 兼容（Callback）

安装可选依赖：pip install mingx[langchain] 或 uv sync --extra langchain。通过 CallbackHandler 将 LangChain 的 chain/LLM/tool/retriever 运行自动记为 Mingx 的 span/generation，并随 OTLP 上送。

from mingx.langchain import CallbackHandler
from langchain_core.runnables import RunnablePassthrough
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([("human", "{question}")])
chain = prompt | some_llm | parser

handler = CallbackHandler()
result = chain.invoke({"question": "..."}, config={"callbacks": [handler]})

# 进程退出时会自动 flush；需确保发出时可手动调用
get_client().flush()

示例见 examples/langchain_tracing.py、examples/langgraph_tracing.py（LangGraph StateGraph）、examples/langchain_agent_tracing.py（create_agent / ReAct Agent）。装饰器示例见 examples/decorator_example.py。

与 Langfuse 一致的数据转换：CallbackHandler 在记录 LLM/Chat 时会对输入输出与用量做与 Langfuse 相同的处理，便于后端解析或迁移：

• 消息：HumanMessage/AIMessage/SystemMessage/ToolMessage/FunctionMessage/ChatMessage 转为统一 {role, content, ...} 结构（含 tool_calls、tool_call_id 等）；Chat 的 output 在 on_llm_end 中同样按该格式记录。
• 用量：从 llm_output.token_usage、generation_info.usage_metadata、message.usage_metadata、response_metadata.usage 等多处收集，并统一为 prompt_tokens / completion_tokens / total_tokens（兼容 Anthropic、Bedrock、Vertex、IBM 等厂商字段名）。

数据出口

所有数据经标准 OpenTelemetry OTLP 发出，仅依赖上述环境变量或构造函数中传入的 span_exporter / tracer_provider。可对接：

• OpenTelemetry Collector
• Jaeger、Zipkin、Grafana Tempo
• Langfuse 的 OTel 接入
• 任何实现 OTLP 的后端

属性使用 mingx.* 命名空间（如 mingx.observation.type、mingx.trace.user_id），与 Langfuse 的 langfuse.* 区分，便于在同一 OTel 管线中共存或迁移。

Flush 与进程退出

OpenTelemetry 使用 BatchSpanProcessor 异步批量导出 span。若进程在批次发送前退出，未 flush 的 span 可能丢失，因此之前需要手动调用 get_client().flush()。
现在在首次调用 get_client() 时会注册 atexit，进程正常退出前会自动执行一次 flush()，因此短脚本（如示例）可以不写 flush()。在需要“确保在某一时刻前一定发出”（例如长驻进程里某段逻辑结束后）时再手动调用 flush() 即可。

上送属性与含义

通过 OTLP 上送的每条 span 会携带以下 属性名（均为字符串或由 SDK 序列化后的字符串）。接收端可根据这些 key 解析含义。

Observation 类型（枚举）

属性 mingx.observation.type 的取值由枚举 ObservationType 统一维护，与 Langfuse 观测类型 对齐，便于在同一 OTLP 管线或迁移时语义一致。可从 mingx 导入使用：

from mingx import ObservationType, get_client

# 使用枚举（推荐）
get_client().start_as_current_observation("llm-call", as_type=ObservationType.GENERATION, model="gpt-4", ...)
get_client().start_as_current_span("step", ...)  # 内部即 ObservationType.SPAN

枚举成员	取值（OTLP）	含义	Mingx 何时使用
ObservationType.EVENT	event	瞬时点事件	create_event()
ObservationType.SPAN	span	一段时间内的单元工作	默认；或未识别的 chain
ObservationType.GENERATION	generation	LLM 生成（prompt、token、成本）	LLM/Chat 回调；手动 as_type=ObservationType.GENERATION
ObservationType.AGENT	agent	Agent 决策/应用流（如 ReAct）	LangChain chain 且 serialized/name 含 agent
ObservationType.TOOL	tool	工具调用（如天气 API）	LangChain on_tool_start
ObservationType.CHAIN	chain	链式步骤（如 retriever→LLM）	LangChain chain 且非 agent
ObservationType.RETRIEVER	retriever	检索步骤（向量库/数据库）	LangChain on_retriever_start
ObservationType.EVALUATOR	evaluator	评估函数（相关性/正确性等）	手动或后续集成
ObservationType.EMBEDDING	embedding	Embedding 调用（含用量/成本）	手动或后续集成
ObservationType.GUARDRAIL	guardrail	护栏（防恶意/jailbreak）	手动或后续集成

LangChain 自动映射：使用 CallbackHandler 时，chain/tool/retriever/llm 会按上表自动设为对应类型；chain 若在 serialized 的 id 或 run 的 name 中含 "agent" 则记为 agent，否则记为 chain。与 Langfuse 的 _get_observation_type_from_serialized 规则一致。

观测类型转换（与 Langfuse 一致）：若某次 chain 在 on_chain_start 时被标为 chain（因 serialized/name 未含 "agent"），但后续 LangChain 触发了 on_agent_action 或 on_agent_finish，则将该 run 的 mingx.observation.type 强制改为 agent，并更新 input/output，保证实际执行的是 Agent 时类型正确。

API 中 as_type 参数均接受 ObservationType 或字符串，便于兼容旧代码。

Trace 级（根 span 或通过 update_current_trace 设置）

属性名	含义	典型取值/说明
mingx.trace.name	轨迹名称	字符串
mingx.trace.user_id	用户 ID	字符串
mingx.trace.session_id	会话 ID	字符串
mingx.trace.input	轨迹级输入	JSON 字符串
mingx.trace.output	轨迹级输出	JSON 字符串
mingx.trace.metadata	轨迹元数据（非对象时）	JSON 字符串
mingx.trace.metadata.<key>	轨迹元数据单字段	字符串/数字等
mingx.trace.tags	轨迹标签列表	JSON 数组字符串

Observation 级（每条 span 均可有）

属性名	含义	典型取值/说明
mingx.observation.type	观察类型	见上表，对应 ObservationType 各成员的取值（如 span / generation / agent / tool / chain / retriever 等）
mingx.observation.input	该观察的输入	JSON 字符串
mingx.observation.output	该观察的输出	JSON 字符串
mingx.observation.level	级别	INFO / WARNING / ERROR 等
mingx.observation.status_message	状态或错误信息	字符串
mingx.observation.metadata	观察元数据（非对象时）	JSON 字符串
mingx.observation.metadata.<key>	观察元数据单字段	字符串/数字等

Generation 专用（仅当 mingx.observation.type 为 generation 时）

属性名	含义	典型取值/说明
mingx.observation.model	模型名称	字符串
mingx.observation.model_parameters	模型参数	JSON 字符串（如 temperature、max_tokens）
mingx.observation.usage_details	用量统计	JSON 字符串（如 prompt_tokens、completion_tokens）
mingx.observation.cost_details	成本信息	JSON 字符串
mingx.observation.completion_start_time	补全开始时间	ISO8601 或纳秒时间戳字符串

通用

属性名	含义	典型取值/说明
mingx.environment	环境标识	字符串（如 default）
mingx.version	应用或 SDK 版本	字符串

说明：Trace 级属性通常出现在根 span 上，或在对当前 span 调用 update_current_trace 后写入当前 span；Observation 级与 Generation 专用属性写在对应 span 的 attributes 中，便于任意 OTLP 接收端解析与展示。观测与跟踪的 input/output（及 trace.tags）统一经 serialize_for_attribute() 序列化：

• message 型（含 content 且 type/role 为字符串，如 LangChain BaseMessage）：先转为标准结构 {role, content, tool_calls?, tool_call_id?, name?}，再输出 JSON，避免 repr 式字符串。
• 其它：递归将 dict/list 内 message 型同样标准化后，整体以标准 JSON 输出（ensure_ascii=False，简洁单行）。
  装饰器、LangChain Callback、update_current_span/update_current_trace、span.update() 等所有路径均使用该统一序列化。

发布到 PyPI

1. 构建包：uv build
2. 使用项目内的 .pypirc 上传（认证文件勿提交，已加入 .gitignore）：

# 需先安装 twine：uv add --dev twine  或  pip install twine
twine upload --config-file .pypirc -r pypi dist/*

其中 -r pypi 对应 .pypirc 里 [pypi] 段；若把 .pypirc 放在用户目录 ~/.pypirc，可省略 --config-file .pypirc，twine 会默认读取。

开发与测试

uv sync --all-extras   # 安装 dev 依赖
uv run pytest          # 运行测试

License

MIT