使用Langfuse对LangGraph博客处理流程进行追踪：包含文本片段、评分和警告信息。

只有当失败的任务能够被重新构建时，各种防护机制（源代码检查、引用检查、重试次数限制、人工介入）才能发挥作用。而多智能体博客处理流程在此时会引发诸多疑问：研究来源是否不可靠？作者是否忽视了引用警告？审稿人拒绝该文章的理由是否正确？搜索操作是否超时？执行该任务需要消耗多少模型调用次数和令牌成本？

如果没有可观测性，系统就只是层层包裹的黑箱。多智能体追踪是这项工作中极为困难的部分。该方案将Langfuse集成到LangGraph博客处理流程中，每篇文章的处理过程都会生成一条追踪记录，其中包含任务交接信息、模型与工具的调用情况、耗时、成本、错误以及评估结果。

确定什么是追踪记录

在处理任何代码之前，先选择要检查的单元。在此流程中，一次追踪相当于生成一篇文章的一次尝试。它从主题被提交时开始，到流程完成、暂停等待人工处理或失败时结束。其中包含了针对每项已完成工作的观察记录：

blog-pipeline
├── research-agent
│   ├── search-web
│   └── research-model
├── writer-agent
│   └── writer-model
├── citation-check
│   └── citation-review-model
└── reviewer-agent
    └── reviewer-model

Langfuse 操作时涉及三个名词。trace指的是从开始到结束的完整运行过程。observation则是该过程中的单个步骤，可能是某段代码执行（应用逻辑）、一次模型调用（LLM 调用），或一个事件（微型标记）。session则将多个 trace 绑定在一起，比如针对同一博客主题的所有尝试。

这种层级结构反映了调试的流程：从整体运行开始，逐步深入到导致问题的具体代理、工具或模型调用。

设置 Langfuse

创建一个 Langfuse 项目，并通过环境变量提供其凭证。密钥绝不能被写入源代码文件中。

pip install -U langfuse

export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_BASE_URL="https://cloud.langfuse.com"
export LANGFUSE_TRACING_ENVIRONMENT="development"

langfuse 包是 Python SDK。这两个密钥用于验证应用与项目的关联，而 LANGFUSE_BASE_URL 用于指定部署地址。示例针对的是欧盟云区域，可根据需要更换为其他区域或自托管实例。

环境标签可将开发阶段的记录与生产环境的记录区分开。使用如 development、staging 和 production 这类简短且稳定的名称效果较好。如果没有这些标签，实验记录和实际文章运行记录会混在一起，导致调试变得十分混乱。

集成回调处理程序

Langfuse 提供了 LangChain 回调处理程序。LangGraph 基于 LangChain 的可执行机制和回调功能构建，因此同一个处理程序能够捕获图结构、链结构、工具及模型相关的事件：

from langfuse import get_client, propagate_attributes
from langfuse.langchain import CallbackHandler


langfuse = get_client()
langfuse_handler = CallbackHandler()

get_client()会根据环境变量构建客户端。该处理程序应在启动时创建一次，之后在每次运行中重复使用。它会监听LangChain事件并将它们作为观测结果传递出去，这样就无需手动追踪每一次模型调用。

接下来是propagate_attributes的作用。它会在单次运行中的所有观测结果上打上相同的标签，从而使搜索调用、内容生成以及审核生成能够作为同一文章处理流程的组成部分被筛选出来。

图结构调用被封装在根跨度中，处理程序则通过LangGraph配置传递：

def run_blog_pipeline(topic: str, blog_id: str, user_id: str):
    initial_state = {
        "topic": topic,
        "audience": "developers new to agent systems",
        "research_brief": "",
        "sources": [],
        "open_questions": [],
        "article_draft": "",
        "review_feedback": "",
        "citation_issues": [],
        "approved": False,
        "revision_count": 0,
        "status": "researching",
    }

    with langfuse.start_as_current_observation(
        as_type="span",
        name="blog-pipeline",
        input={"topic": topic, "audience": initial_state["audience"]},
    ) as pipeline_span:
        with propagate_attributes(
            trace_name="blog-pipeline",
            session_id=blog_id,
            user_id=user_id,
            tags=["blog-pipeline", "langgraph"],
            version="1.0.0",
            metadata={"workflow": "research-write-review"},
        ):
            trace_id = langfuse.get_current_trace_id()
            result = graph.invoke(
                initial_state,
                config={"callbacks": [langfuse_handler]},
            )

        pipeline_span.update(
            output={
                "status": result["status"],
                "approved": result["approved"],
                "revision_count": result["revision_count"],
            }
        )

    return result, trace_id

该封装结构的几个细节很重要。外层的start_as_current_observation()调用会创建根观测对象，其输入被刻意设计得极为简洁，仅包含主题和受众信息。这些内容能让人一目了然地看到跟踪列表中的内容，因此内部状态的信息无需显示在其中。

propagate_attributes()函数会为每个嵌套观测对象添加标签。blog_id用作会话标识，将针对同一篇文章的多次重试或重复操作归类到一起。version值则用于标识结果背后的工作流程及提示词版本，这样当行为发生变化时，就可以通过版本来筛选跟踪记录，而无需凭猜测。

处理程序会遍历LangGraph的常规config配置，进而自动收集框架层面的详细信息：包括每次节点调用、大语言模型生成过程、工具调用情况，以及输入内容、输出结果、延迟时间、token数量和成本等数据。

最终，根级跨度会更新为三个管道级结果：运行是否完成、草稿是否通过审核以及进行了多少次修改。返回的 trace_id 应存储在应用程序日志或用户界面中，以便直接从文章处理过程跳转到对应的 Langfuse 跟踪记录。

代理节点中的业务级跨度

自动跟踪功能只能证明模型已被调用，但无法显示各处理环节之间传递了什么数据。为此，可在代理节点内部添加一个轻量级跨度，并将 config 参数传递给嵌套的 LangChain 或 LangGraph 调用，从而使处理程序保持相同的跟踪上下文：

from langchain_core.runnables import RunnableConfig


def research_node(state: BlogState, config: RunnableConfig) -> dict:
    with langfuse.start_as_current_observation(
        as_type="span",
        name="research-agent",
        input={"topic": state["topic"], "audience": state["audience"]},
    ) as span:
        result = research_agent.invoke(
            {"topic": state["topic"], "audience": state["audience"]},
            config=config,
        )

        span.update(
            output={
                "source_count": len(result["sources"]),
                "open_question_count": len(result["open_questions"]),
                "brief": result["research_brief"],
            }
        )

    return {
        "research_brief": result["research_brief"],
        "sources": result["sources"],
        "open_questions": result["open_questions"],
        "status": "writing",
    }

该标签仅用于补充回调处理函数，而非替代它。它补充了框架无法推断的含义，即“研究成功”对于该处理流程而言具体指什么。此处会记录来源数量、未解决问题的数目以及交给写作者的简短说明。

将config: RunnableConfig传递给research_agent.invoke()时，会同时携带回调函数和追踪上下文。若忘记传递此参数，就会出现典型的可观测性故障：虽然顶层流程会被追踪，但嵌套调用却会显示为无关的追踪记录或完全消失。

请保持观测名称的一致性：research-agent、writer-agent、citation-check。主题名称、模型名称或用户ID不应被纳入这些名称中，因为这类数据属于输入内容、模型属性或元数据范畴。统一的名称能确保仪表板和筛选条件在实现方式变更后依然有效。

警告与错误

并非所有异常结果都意味着程序崩溃。可疑的数据来源、用尽的修订预算或不受支持的引用同样重要，也应以警告形式呈现。

def record_source_assessment(assessment: SourceAssessment) -> None:
    if assessment.suspicious_content:
        langfuse.update_current_span(
            level="WARNING",
            status_message="Untrusted source contained agent-directed instructions.",
        )


def record_pipeline_failure(error: Exception) -> None:
    langfuse.update_current_span(
        level="ERROR",
        status_message=f"Pipeline failed: {type(error).__name__}",
    )

Langfuse 支持为观测结果设置日志级别。WARNING 适用于已结束但需要进一步检查的运行，而 ERROR 则用于真正的故障情况。状态消息应描述运行中的问题，同时不得将敏感的源代码或机密信息包含在遥测数据中。

源码评估记录器应放在源码检查路径中，而故障记录器则应置于能够实际进行恢复或报告的错误处理模块中。仅仅为了记录而捕获所有异常并不是好做法，因为意外错误仍应被传递到错误监控系统并引发明显警报。

结果评分

追踪信息会说明发生了什么，而评分则用于判断结果是否合格。一旦运行完成，应根据管道状态立即生成一个确定的初始评分：

result, trace_id = run_blog_pipeline(
    topic="How AI agents use tools",
    blog_id="blog-ai-tools-001",
    user_id="philip",
)

if trace_id:
    langfuse.create_score(
        trace_id=trace_id,
        name="review_approved",
        value=1 if result["approved"] else 0,
        data_type="BOOLEAN",
        comment=result["status"],
    )

    langfuse.create_score(
        trace_id=trace_id,
        name="revision_count",
        value=float(result["revision_count"]),
        data_type="NUMERIC",
    )

第一个评分反映了审稿人明确的通过或拒绝意见，第二个评分则衡量草稿需要多少修改。随着时间积累，这些数据能揭示的远不止是一些零散案例。哪种模型、提示词版本、主题类别或资料获取方式能带来更少的修改和更多的通过率？

这两者都不是绝对标准，因为审稿人也可能出错。在阅读最终稿后给出的人工评分也可能出现得较晚。关键在于让质量可衡量，并在不同案例之间进行比较，而非依赖少数几个令人印象深刻的例子。

调试流程

当某篇文章质量不佳时，打开其相关记录并按顺序提出以下问题：

1. 流程是否按预期进行？ 检查研究、撰写、引用核查和审稿环节，以及任何修改循环。

研究交接的完整性如何？需查看资料数量、未解决的问题、引用情况以及任务说明本身。

缺陷出现在哪个环节？要将每次调用的提示、输出结果、延迟时间、Token使用量以及工具运行状态与之前的步骤进行对比。

是模型故障还是系统故障？超时问题和不充分的引用需要完全不同的解决方式。

防护机制是否有效？如果某项资料看起来可疑或某个论点缺乏依据，系统是否会将任务转至修订环节或人工审核而非直接批准？

正因如此，追踪路径的清晰度至关重要。一个命名合理的结构能将“博客内容有误”这样的模糊表述转化为精确的结论，例如“作者在第二次修订后添加了任务说明中未提及的论点。”

防止敏感数据泄露

追踪功能可以记录提示词、输出结果、工具运行结果以及元数据，这也意味着它可能会捕获本不应被存储的信息。API密钥、认证标头以及完整的敏感配置绝不能传入Langfuse。如果后续有用户数据进入处理流程，需先检查会被捕获的内容，并在导出追踪信息前设置掩码功能。Langfuse提供了掩码功能，可在敏感数据离开应用程序之前对其进行遮蔽。

一般原则是：记录的内容要足以解释某个决策，且不能超过安全保留的范围。

处理流程的益处

每次运行都会留下操作记录：

• 每一次文章生成尝试都对应一条追踪记录。
• 各种代理、模型和工具会以嵌套观察项的形式呈现。
• 固定的名称加上元数据、版本信息、会话状态及环境信息，便于进行筛选。

来源存疑的内容会触发警告，而真正的故障则会标记为错误。

通过率与修改次数会被记录为分数。

追踪编号可直接指向某篇文章对应的运行记录。

调试不再需要滚动终端日志并凭猜测进行，而是可以精确查看问题出现的具体环节。