可观测性是调试工作流的关键。除了加 print() 语句，workflows 自带一套完善的 instrumentation 系统，会记录每个工作流步骤的输入与输出。

此外，你还能借助这套 instrumentation 系统，为工作流步骤之外的任意函数加上可观测性！更深入的示例见可观测性示例目录。

用内部事件观察步骤状态

除了你的步骤发出的事件，运行时还会发布描述自身执行的内部事件。它们默认会被 stream_events() 过滤掉。传入 expose_internal=True 就能看到：

from workflows.events import StepStateChanged, StepState

handler = workflow.run()
async for ev in handler.stream_events(expose_internal=True):
    if isinstance(ev, StepStateChanged):
        print(ev.name, ev.step_state, ev.worker_id)

StepStateChanged 在步骤执行状态发生变化时触发。它的 step_state 取 StepState.PREPARING（已入队，等待 worker 名额）、StepState.RUNNING（正在执行）或 StepState.NOT_RUNNING（已结束）之一。它还带有 name、worker_id、input_event_name 与 output_event_name。每次步骤执行触发一次，因此 @step(num_workers=N) 的步骤或扇出会按每个条目各发一个事件。

持久化工作流的检查点循环就用这个事件在步骤执行完成后做快照。

OpenTelemetry 集成

Workflows 可与 OpenTelemetry 集成以导出轨迹。安装 llama-index-observability-otel 包：

终端窗口

pip install llama-index-observability-otel

然后在你的应用里配置它：

from llama_index.observability.otel import LlamaIndexOpenTelemetry

# Initialize with your span exporter
instrumentor = LlamaIndexOpenTelemetry(
    span_exporter=your_span_exporter,
    service_name_or_resource="your_service_name",
)

# Start registering traces
instrumentor.start_registering()

所有工作流步骤、LLM 调用与自定义事件都会被自动捕获，并作为 OpenTelemetry span 导出，附带详细属性，包括：

• 每个工作流步骤的 span 名称
• 开始与结束时间
• 事件属性（输入数据、输出数据等）
• 展现执行流程的嵌套 span 关系

第三方可观测性工具

Workflows 与主流可观测性平台可以无缝集成：

Arize Phoenix

Arize Phoenix 为你的工作流提供实时追踪与可视化。

更多内容见示例 notebook。

Langfuse

Langfuse 直接与 workflows 自带的 instrumentation 系统集成。

更多内容见示例 notebook。

Opik

Opik 可以通过 workflows 使用的同一条 OpenTelemetry 管线接收工作流轨迹。

要配置导出到 Opik，使用上面的 OpenTelemetry 设置，并把 exporter 指向 Opik 的 OTLP 端点。参见 Opik OpenTelemetry Python SDK 指南。

自定义 span 与事件

你可以用 LlamaIndex 的 dispatcher 定义自定义 span 与事件，为自己的代码加上细粒度追踪：

from llama_index_instrumentation import get_dispatcher
from llama_index_instrumentation.base import BaseEvent

dispatcher = get_dispatcher()

# Define custom events
class ExampleEvent(BaseEvent):
    data: str

class AnotherExampleEvent(BaseEvent):
    print_statement: str

# Use the @dispatcher.span decorator
@dispatcher.span
def example_fn(data: str) -> None:
    dispatcher.event(ExampleEvent(data=data))
    s = "This are example string data: " + data
    dispatcher.event(AnotherExampleEvent(print_statement=s))
    print(s)

调用被 instrumentation 包裹的函数时，所有 span 与事件都会被已配置的追踪后端自动捕获。

完整示例见 examples/observability 目录。