实用笔记：从零构建多智能体系统——第6部分

以下笔记为“从零构建多智能体系统——第6部分：可观测性与调试”提供了实用的学习路径。重点在于契约、校验机制以及可直接插入的代码占位符，而非动机性阐述。 在完成概览阶段时，首先明确契约内容：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 优先选择小型、可测试的单元，而非庞大的脚本。当某个步骤出错时，错误应指向单一责任模块，而非复杂的流程链。

追踪内容应当代表什么？

“应当追踪什么”这一阶段若被视为可测量的界面，则效果最佳。在扩大范围之前，先记录一份典型的成功案例、一个失败案例以及回滚说明。 将这一阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许默默地仅完成部分工作。 保持图结构的状态简洁且具有类型定义。嵌套的数据块会掩盖是哪个节点修改了哪个字段，还会在中断后导致无法继续处理。

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

将 Langfuse 连接到 LangGraph

将 Langfuse 与 LangGraph 相连的阶段若视为可度量的界面，效果会最佳。在扩大范围之前，先记录一份理想的转录结果、一个失败案例以及回滚说明。在功能结果旁同时记录处理时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。需保持图结构简洁且类型明确，嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，且在中断后还会导致流程无法继续。

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"

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


langfuse = get_client()
langfuse_handler = CallbackHandler()

追踪一次完整的文章处理流程

将 Trace one complete article 阶段视为可测量的对象来处理时，其效果最佳。在扩大范围之前，需记录一份完美的日志、一个故障案例以及回滚说明。 应将配置与应用程序代码分开。环境文件、密钥存储和功能标志应集中存放于一处，以便操作人员无需查看整个结构即可进行审计。 要保持图结构的扁平化与类型化。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会导致在出现中断后无法继续处理。 将 Trace one complete article 阶段视为可测量的对象来处理时，其效果最佳。在扩大范围之前，需记录一份完美的日志、一个故障案例以及回滚说明。 相比庞大的脚本，更应优先使用小型且可测试的单元。当某个步骤失败时，故障应指向单一的责任主体，而非复杂的处理流程。

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

添加用于说明交接过程的观察记录

在修改代码之前，对于“添加用于解释当前阶段的观测值”这一操作，需先明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 应将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功判定标准，杜绝默许的半完成状态。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务上的完整性。

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",
    }

标记需要关注的信号

在修改代码之前，需为该阶段标记信号、明确输入参数、确定步骤负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。对于会耗费资金或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。

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__}",
    )

将审核人员的决策转化为评分

要将 Turn Reviewer 的决策转化为具体阶段，需在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审计。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。 要将 Turn Reviewer 的决策转化为具体阶段，需在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 相比庞大的脚本，应优先使用小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向单一责任主体，而非多个部分共同导致的问题。

带角度的管道。

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",
    )

用五个问题调试失败的运行

在处理“调试失败运行”阶段时，首先写下相关契约：所需的输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功检测标准，并杜绝无声的局部完成。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。

可观测性也有隐私边界

在处理可观测性相关工作时，也有一个步骤：首先明确契约内容——所需的输入参数、成功信号，以及部分失败时会发生什么。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，就能避免在从演示环境过渡到共享环境时出现意外费用。 在耗时的操作之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次对同一次大型语言模型调用收费。

我们的成果

在“我们构建了什么”阶段工作时，首先需明确相关规范：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持透明可追溯。 应将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中于一处，以便操作人员无需查看整个系统结构即可进行审计。 在耗时较长的步骤之后设置检查点。当操作人员重新执行后续节点时，恢复流程不应再次调用相同的大型语言模型接口。 在“我们构建了什么”阶段工作时，首先需明确相关规范：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持透明可追溯。 相较于庞大的脚本，更应优先使用小型且易于测试的单元。当某个步骤失败时，故障原因应能明确指向某个特定功能模块，而非整个复杂的流程。

操作检查清单

在操作检查清单阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新执行该步骤，而无需猜测隐藏状态。

需同时记录正常流程和故障恢复流程。重试机制、人工审核环节以及错误处理都是产品不可或缺的部分，而非后续需要补充的内容。

对于涉及资金支出或修改生产数据的操作，必须经过人工审批。仅靠编译时的配置并不能保证业务的完整性。

编写简短的操作手册：说明如何轮换密钥、如何清空队列以及如何回滚上一次的数据导入操作。

优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出现故障时，故障原因应能明确指向某个具体的责任环节，而非整个复杂的流程链。

对于那些会花费资金或更改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。

在推广该技术栈之前，应先冻结版本，为关键流程记录完整的操作日志，并明确回滚步骤。共享环境需要设置速率限制、租户验证机制，以及明确的密钥轮换负责人。与其追求花哨的一次性演示，不如注重扎实的可靠性。

关于 cf19385cb4a9 的批注：请将提供方密钥移出代码仓库，为每个会话设置令牌使用上限，并将操作日志存储在评估用配置文件旁边，以便后续模型更换时仍能保持数据可比性。