利用LangGraph与LangSmith评估结果构建智能体改进循环。

将智能体投入实际使用所需精力最少。而之后如何让它变得更好，才是区分业余开发作品与真正可靠系统的关键。许多团队发布第一个版本后，就要花数月时间逐一处理各种投诉。每个补丁都是针对某个具体问题的解决方案，每项修复都仅限于局部范围，导致智能体整体从未得到提升，只会不断积累临时解决方案。

常见的原因在于将智能体当作普通软件来对待。在常规程序中，代码会明确说明系统的功能，因此出现漏洞时只需逐行阅读源代码就能找到问题所在。但对于智能体而言，这种假设并不成立。代码仅描述系统被允许执行的功能，而实际运行记录则记载了它真实执行的内容。正是这一差异构成了所有严谨评估流程的基础，包括Anthropic在其工程文档中以及LangChain在LangSmith文档中描述的那些评估方法。

以下是 LangGraph 智能体完整的改进循环流程。它从一个简单的 ReAct 智能体开始，将其运行轨迹记录在 LangSmith 中，加入三层评分机制，识别反复出现的故障模式，将这些故障转化为永久性的离线测试集，并在任何修复方案发布前进行回归评估。每个阶段都配有代码、说明以及预期输出。

该循环的运作方式如下：

1. 智能体处理真实用户的查询，生成涵盖所有大语言模型调用、工具调用及中间输出的轨迹记录。
2. 基于代码的检查会从格式和工具使用正确性方面对每条轨迹进行评分，大语言模型评估器则会从有用性和相关性角度打分，最后由人工审查那些较为隐蔽的故障。
3. 评分较低的轨迹会被归类为诸如使用了错误工具、推理偏差或未能理解用户意图等故障模式。

每一次重复出现的故障都会变成一个具有已知预期行为的离线测试用例。

在新的智能体版本发布之前，会先让它在这些不断扩充的测试用例集中运行。

只有那些能提升基准水平的改动才会被应用到实际环境中，而每一个已修复的故障都会永久保留在测试用例集中。

这就是智能体为何能随着时间不断改进，而不仅仅是堆积补丁的原因。整个过程分为六个阶段：

• 第一阶段：基础构建——配备LangSmith追踪功能的LangGraph ReAct智能体。
• 第二阶段：轨迹收集——将实际运行记录整理为结构化数据。
• 第三阶段：评分评估——通过代码检查、大语言模型评判以及人工审核来完成。
• 第四阶段：模式识别——将评分后的轨迹转化为不同的故障类别。
• 第五阶段：离线测试套件——将各类故障转化为不会引发回归问题的数据集。

第6阶段：闭环处理——在发货前进行评估并提升基准水平。

第1阶段：基础智能体

第一阶段会构建一个小型客户支持智能体，用于回答与账户相关的问题。该智能体拥有两个工具：一个用于查询账户详情，另一个用于检查近期订单；它使用LangGraph预构建的ReAct智能体，这是该平台中工具调用智能体的标准架构。同时启用了LangSmith追踪功能，因此每次运行时，大语言模型的调用以及工具的调用都会被自动记录下来。

实现这一功能的要素有三个：

1. LangGraph StateGraph是底层架构，负责在决定执行操作的节点与实际运行工具的节点之间进行数据传递。create_react_agent辅助函数用于封装这一架构。

当环境变量 LANGSMITH_TRACING=true 被设置时，LangSmith 追踪功能就会启用。此后，每次调用 LangChain 或 LangGraph 都会生成追踪信息，且无需修改代码。

工具是由 LangChain 中的 @tool 装饰的 Python 函数，这些函数会为每个工具提供名称和描述，以便智能体选择合适的工具。

环境配置

首先需要安装相关库：

# Install the libraries we need
pip install langgraph langchain langchain-openai langsmith

将安装四个包。langgraph提供图计算运行时以及预构建的ReAct智能体。langchain提供@tool装饰器及消息类型。langchain-openai将OpenAI聊天模型封装在LangChain的接口中。langsmith则是用于连接LangSmith后端以进行追踪和评估的客户端库。

接下来是环境变量，它们无需修改代码即可开启追踪功能：

export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY=your_langsmith_key
export LANGSMITH_PROJECT=agent-improvement-loop
export OPENAI_API_KEY=your_openai_key

LANGSMITH_TRACING=true 是对应的开关。LANGSMITH_PROJECT 用于指定用于归类追踪记录的项目，这样之后就可以对这些记录进行过滤。另外两个变量则是 LangSmith 和模型提供方的认证信息。

设置好这四个变量后，流程中的任何 LangChain 或 LangGraph 调用都会将追踪记录发送到 LangSmith。由于追踪功能在后台运行，因此无需进行任何封装或装饰处理。

构建智能体

首先是需要导入的模块：

# Step 1: Import the pieces we need
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

这些导入模块包含了三个组件。ChatOpenAI为LangChain封装了OpenAI的聊天完成API。tool装饰器可将普通的Python函数转换为智能体能够调用的功能。create_react_agent则是LangGraph提供的现成辅助函数，用于构建ReAct风格的图结构以调用各类工具。

接着是两个工具：

# Step 2: Define two tools the agent can call
@tool
def get_account_info(account_id: str) -> str:
    """Look up basic information for an account by account_id."""
    fake_accounts = {
        "A100": "Account A100: plan=pro, status=active, email=ada@example.com",
        "A200": "Account A200: plan=free, status=suspended, email=turing@example.com",
    }
    return fake_accounts.get(account_id, f"No account found with id {account_id}")

@tool
def get_recent_orders(account_id: str) -> str:
    """Return the three most recent orders for a given account_id."""
    fake_orders = {
        "A100": "Orders for A100: #9001 shipped, #9002 processing, #9003 returned",
        "A200": "Orders for A200: no orders in the last 90 days",
    }
    return fake_orders.get(account_id, f"No orders for {account_id}")

每个工具都会接收一个account_id并返回一个字符串。每个函数的文档字符串非常重要，因为大语言模型在决定调用哪个工具时会读取这些文本。装饰器会提取函数名、参数及文档字符串，从而构建模型所能识别的结构。在真实系统中，这些函数会查询数据库或API，但本演示为了保持独立性和可重复性，直接返回硬编码的字符串。

最后，代理就被组装完成了：

# Step 3: Wire the tools into a ReAct agent
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [get_account_info, get_recent_orders]
agent = create_react_agent(model, tools)

为确保结果可复现，聊天模型是在温度值为零的情况下创建的。这些工具会被列入列表，并与模型一起传递给create_react_agent函数，该函数会返回一个已编译好的LangGraph StateGraph，其中包含推理步骤、工具执行步骤以及连接这两者的条件链接。

测试智能体

可以通过一个简单的查询来对其进行测试：

# Step 4: Invoke the agent with a simple query
result = agent.invoke({
    "messages": [{"role": "user", "content": "What plan is account A100 on?"}]
})

for message in result["messages"]:
    print(message.type, ":", message.content)

该调用会传递一个包含 messages 键的字典。这是因为 create_react_agent 依赖于预先构建的 MessagesState 模式，其状态仅为一组聊天消息列表，代理在思考及调用工具时会对该列表进行扩展。之后的循环会打印每种消息类型及其内容，从而让整个执行过程一目了然。

预期输出：

human : What plan is account A100 on?
ai :
tool : Account A100: plan=pro, status=active, email=ada@example.com
ai : Account A100 is on the pro plan.

智能体接收到了用户消息，随后发出一条AI消息以触发工具调用（由于该消息仅包含工具调用指令而无实际文本，因此内容为空）。工具返回了账户字符串，智能体最终给出回复。现在打开LangSmith，可在agent-improvement-loop项目中看到一条新的追踪记录，每一步都被完整保存。

之前的代码并未对这类追踪信息进行捕获。一旦在环境中设置LANGSMITH_TRACING=true，所有的调用都会将输入、输出及中间步骤发送至LangSmith。正是这种全局可观测性为后续循环流程提供了依据。

第一阶段会让在LangGraph下运行的工具调用型智能体在每次执行时都生成追踪记录。不过，仅存在于控制面板中的追踪记录尚不能视为改进信号，还需将其提取出来并加以利用，而这正是下一阶段要完成的任务。

第二阶段：收集轨迹数据

此阶段负责从 LangSmith 中提取轨迹数据并导入分析代码中。轨迹数据指的是一次完整的智能体运行记录，包含输入内容、输出结果、工具调用信息以及延迟时间。LangSmith 的 Python 客户端会列出项目中的所有运行记录，根据时间范围和状态进行筛选，然后将这些记录转换为评分阶段可使用的对象。

以下术语有助于理解代码：

1. “运行记录”指的是一段被记录下来的工作内容，可以是一次完整的智能体调用，也可以是其内部的一次大语言模型调用。多次智能体调用会形成一棵运行记录树。
2. “根级运行记录”是指某次智能体调用的最高层级记录，当你希望每个用户查询只显示一条记录时，显示的就是这类记录。
3. “项目”指的是运行记录所属的命名空间，由 LANGSMITH_PROJECT 参数设定。通过项目进行筛选可以将生产环境中的运行记录与本地开发环境中的记录区分开来。

获取最近的运行记录

首先创建客户端：

# Step 1: Create a LangSmith client
from langsmith import Client
from datetime import datetime, timedelta

client = Client()

Client是从langsmith包中导入的，创建时无需传入任何参数。它会自动从环境中读取LANGSMITH_API_KEY，因此无需手动传递任何凭证。

接着会列出最近的根目录操作记录：

# Step 2: List recent root runs from our project
recent_runs = list(client.list_runs(
    project_name="agent-improvement-loop",
    is_root=True,
    start_time=datetime.utcnow() - timedelta(days=1),
))

print(f"Found {len(recent_runs)} root runs in the last 24 hours")

client.list_runs 方法会使用三个过滤器进行调用。project_name 用于将查询范围限定在特定项目中，is_root=True 只保留最顶层的代理调用，而不会包含所有嵌套的 LLM 和工具调用，start_time 则将结果限制在过去的 24 小时内。该迭代器是惰性的，因此需要通过 list() 方法将其转换为实际的列表。

预期输出：

Found 42 root runs in the last 24 hours

仅保留重要的字段

原始的LangSmith运行结果包含大量字段，但评分仅需五项信息：输入内容、输出结果、延迟时间、任何错误以及显示使用了哪些工具的子运行记录。

# Step 3: Build a compact trace record from each run
def compact_trace(run):
    return {
        "run_id": str(run.id),
        "inputs": run.inputs,
        "outputs": run.outputs,
        "error": run.error,
        "latency_ms": (run.end_time - run.start_time).total_seconds() * 1000
            if run.end_time else None,
        "tool_calls": [
            child.name for child in client.list_runs(
                trace_id=run.trace_id, run_type="tool"
            )
        ],
    }

traces = [compact_trace(run) for run in recent_runs]
print(f"Collected {len(traces)} compact traces")
print("Example tool calls:", traces[0]["tool_calls"])

该助手接收原始运行结果并将其简化为一个小字典。输入和输出（run.inputs、run.outputs）分别包含用户查询和最终回复，而run.error在成功时为None，失败时则为字符串。延迟是通过用end_time减去start_time得到的。工具调用需要使用run_type="tool"以及相同的trace_id再次进行LangSmith查询，这样只能获取该运行树中的工具执行记录。

预期输出：

Collected 42 compact traces
Example tool calls: ['get_account_info']

其目的并非从 LangSmith 中复制追踪数据后弃之不用。简化版便于评分，而完整的追踪数据则保留在 LangSmith 中以便深入调试。在控制面板中保留详细视图，同时为分析提供简化版本，这是将可观测性与评估功能分开的合理方式。

检查收集器

# Step 4: Inspect the first trace end to end
import json
print(json.dumps(traces[0], indent=2, default=str))

第一条轨迹会通过json.dumps以缩进格式的JSON形式输出。default=str参数会让json将那些它自身无法序列化的内容转换为字符串，这包括了LangSmith返回的datetime和UUID类型。

预期输出：

{
  "run_id": "b1d2e3f4-...",
  "inputs": {"messages": [{"role": "user", "content": "What plan is account A100 on?"}]},
  "outputs": {"messages": [{"role": "ai", "content": "Account A100 is on the pro plan."}]},
  "error": null,
  "latency_ms": 1423.51,
  "tool_calls": ["get_account_info"]
}

现在的每次生产运行都会生成结构清晰、条理分明的记录，后续的评分阶段会对这些记录进行进一步丰富。收集追踪数据成本很低，关键在于如何让它们发挥实际作用。许多团队停留在控制台层面，不明白为何可观测性始终无法转化为改进。没有评分的追踪数据只是噪音，而没有评分的单条记录也不过是一行日志而已。

第三阶段：对追踪数据评分

三层评分机制共同作用，从而生成一条经过丰富处理的追踪数据。Anthropic 关于评估方法的工程文章也描述了同样的架构：基于代码的评分工具，具备快速、客观且可重复的特点；基于模型的评分工具，则具有灵活性和可扩展性；而人工评分则是黄金标准，但成本较高。这里会构建每种类型的评分工具各一个。

1. 基于代码的检查是快速的确定性函数，用于验证结构上的事实，比如智能体是否调用了正确的工具，或者输出是否符合预期格式。这类检查无需额外成本即可发现明显的错误。
2. 大语言模型评估器是另一个模型，用于评定实用性、相关性及语气等主观质量。由于它会对每个处理记录进行评估，因此具备扩展性，但在获得人们信任之前必须先通过人类判断进行校准。
3. 人工审核会让领域专家查看自动化系统遗漏或评分不一致的处理记录。这些记录就是另外两层评估体系所依据的基准真相。

第一层：工具正确性

# Step 1: Define a rule that says which tool SHOULD be called for a given query
import re

def expected_tool_for_query(query: str) -> str:
    q = query.lower()
    if re.search(r"\b(order|shipment|shipped|return)\b", q):
        return "get_recent_orders"
    if re.search(r"\b(plan|account|status|email)\b", q):
        return "get_account_info"
    return "any"

def score_tool_correctness(trace):
    query = trace["inputs"]["messages"][0]["content"]
    expected = expected_tool_for_query(query)
    actual = trace["tool_calls"][0] if trace["tool_calls"] else None
    if expected == "any":
        return 1.0
    return 1.0 if actual == expected else 0.0

定义了两个函数。expected_tool_for_query会使用正则表达式分析查询文本，以此推测应该由哪个工具来处理该查询。这是一种简单的启发式方法，适合识别明显的路由错误，但不适用于实际生产环境。score_tool_correctness则从追踪信息中获取用户查询，计算出预期的工具，再与智能体实际调用的第一个工具进行比较。若不匹配则得分为0.0，匹配则为1.0。

在正式版本中，会用更强大的分类器或基于意图的规则引擎取代正则表达式。核心原则依然不变：在采用成本较高的基于大语言模型的评估方法之前，先进行确定性检查。

第二层：用于判断有用性的大语言模型评估器

# Step 2: Score qualitative properties with an LLM judge
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

judge_model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

judge_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are an expert reviewer of AI agent responses. "
               "Rate the response on a scale of 1 to 5 for helpfulness. "
               "Respond with only a single integer, nothing else."),
    ("user", "User question: {question}\n\nAgent response: {response}\n\nScore:"),
])

def score_helpfulness(trace):
    question = trace["inputs"]["messages"][0]["content"]
    response = trace["outputs"]["messages"][-1]["content"]
    chain = judge_prompt | judge_model
    result = chain.invoke({"question": question, "response": response})
    try:
        return int(result.content.strip()) / 5.0
    except ValueError:
        return 0.0

通过将提示模板输入到判断模型中，即可构建一个简单的 LangChain 链。系统消息要求判断模型只能返回 1 到 5 之间的一个整数，因此结果可以确定性地解析。在 score_helpfulness 中，会从追踪信息中提取用户问题及最终智能体响应，随后运行该链，并将整数转换为 0 到 1 之间的浮点数，以便与工具评分顺利结合。gpt-4o-mini 被用作判断模型，因为它成本低且速度快。

大型语言模型作为判断工具具有随机性，因此对同一追踪信息进行两次处理可能会得到略有不同的评分。Anthropic 的建议是在依赖此类判断模型之前，先使用人工标注的数据集对其进行校准，而第三层正是用于构建这样的数据集。

第三层：以人工审核作为标注队列

LangSmith拥有一个原生标注队列，用户可在控制面板中对跟踪数据进行标记。为保证示例的完整性，代码中也采用了相同的结构进行建模。

# Step 3: Model a human review signal
def score_human(trace, human_labels: dict) -> float | None:
    run_id = trace["run_id"]
    return human_labels.get(run_id)

human_labels = {
    "b1d2e3f4-...": 1.0,
    "c2e3f4g5-...": 0.0,
}

该函数会在由人工提供的标签构成的字典中查找对应的运行ID。如果存在该标签，则返回对应值；否则函数会返回None，表示该运行尚未被审核。在实际应用中，这个字典会通过LangSmith的注释队列API来访问。重要的是这些评分的表现方式：由于只有部分运行会获得评分，因此评分较为稀少；当出现不同意见时，这些评分具有优先性，可以覆盖其他评分；同时它们还用于调整LLM评估器的校准参数。

整合这三层结构

# Step 4: Run all three scorers and attach scores to each trace
def enrich(trace):
    trace["scores"] = {
        "tool_correctness": score_tool_correctness(trace),
        "helpfulness": score_helpfulness(trace),
        "human": score_human(trace, human_labels),
    }
    return trace

enriched = [enrich(t) for t in traces]
print(json.dumps(enriched[0]["scores"], indent=2))

enrich函数会将三个分数都添加到轨迹中并返回该轨迹。它会被应用于收集过程中获取的每个轨迹，从而生成一个包含多个增强后轨迹的列表，这些轨迹均包含结构化的scores字典。tool_correctness由基于规则的正则表达式生成，helpfulness由大语言模型评估器生成，而human则为浮点数或None。

预期输出：

{
  "tool_correctness": 1.0,
  "helpfulness": 0.8,
  "human": null
}

这三个评分故意存在差异。工具正确性评估认为调用了正确的工具，而大型语言模型评估则认为该答案应得5分中的4分。稍后参与审核的人类评审员可能会认为工具使用正确，但表述方式不符合品牌风格。每一层都能发现其他层遗漏的问题，因此通过叠加这些评估可以在不同成本下获得更全面的分析结果与真实依据。

现在每条记录都包含三个评分：基于规则的评分、大型语言模型评估的评分，以及在有相关数据时的人类评审评分。缺乏结构化的评分几乎和完全没有评分一样没有用处，因此接下来要分析的是各种模式。

第4阶段：查找故障模式

一旦追踪数据带有评分，同样的故障往往会反复出现。智能体会在处理特定表述时选择错误的工具，其在多步骤查询中的推理也会出现偏差。虽然其答案在技术上正确，但却无法满足用户的需求。每种模式都需要不同的解决方法，可能是重新编写工具描述、更新提示词，或是重构工作流程。识别出这种模式才是难点，而之后的修复通常较为简单。

LangChain 提供了专为大规模处理此类问题而设计的 Insights Agent。它能接收追踪数据，按故障类型对它们进行分类，并帮助团队确定哪些问题需要优先处理。下面代码实现了同一概念的简化版本。

筛选故障

# Step 1: Keep only traces where at least one score is low
def is_low_scoring(trace):
    scores = trace["scores"]
    if scores["human"] is not None and scores["human"] < 0.5:
        return True
    if scores["tool_correctness"] < 0.5:
        return True
    if scores["helpfulness"] < 0.6:
        return True
    return False

failures = [t for t in enriched if is_low_scoring(t)]
print(f"{len(failures)} of {len(enriched)} traces flagged as low scoring")

is_low_scoring会进行三项检查。如果人工评分低于0.5，则将该轨迹标记为失败。否则由自动评分决定：工具正确性低于0.5表示智能体选择了错误的路径，而有用性低于0.6则表示大语言模型评估者不认可该答案。这些阈值可以调整，目前为合理的初始默认值。该过滤器会对上一阶段处理后的完整轨迹进行筛选，从而生成失败集合。

预期输出：

9 of 42 traces flagged as low scoring

将故障分类

这里使用一种简单的启发式方法为故障添加标签。在生产环境中，可以利用聚类技术处理这些数据，或者使用 LangChain 的 Insights Agent，它能够在 LangSmith 的追踪数据基础上完成此项工作。

# Step 2: Tag each failure with a category based on its scores
def categorize(trace):
    scores = trace["scores"]
    if scores["tool_correctness"] < 0.5:
        return "wrong_tool"
    if scores["helpfulness"] < 0.6 and scores["tool_correctness"] >= 0.5:
        return "unhelpful_answer"
    if scores["human"] is not None and scores["human"] < 0.5:
        return "human_flagged"
    return "other"

from collections import Counter
categories = Counter(categorize(t) for t in failures)
print(categories.most_common())

categorize函数会按优先级顺序进行检查。使用错误的工具属于最严重的结构问题，因此会被首先标记。即使使用了正确的工具但给出的答案毫无用处，也说明提示词或响应生成存在问题。那些被人工标记为错误但自动化系统却未识别的情况，则会被归入另一类。Python的标准Counter用于统计各类数量，由于这是首次使用该函数，因此直接在代码中导入。

预期输出：

[('wrong_tool', 5), ('unhelpful_answer', 3), ('human_flagged', 1)]

该类别能提示对应的解决方案。连续出现五次使用错误工具的故障，往往与工具描述或路径提示有关。即便使用了正确的工具，仍得到三个无用答案，则问题出在生成响应的提示语上。只有人工才能发现的故障意味着自动评估系统遗漏了某些内容，因此可能需要新的评估人员。

分析单一模式

# Step 3: Print the inputs and outputs for every wrong_tool failure
wrong_tool = [t for t in failures if categorize(t) == "wrong_tool"]
for trace in wrong_tool[:3]:
    print("QUERY:", trace["inputs"]["messages"][0]["content"])
    print("TOOLS CALLED:", trace["tool_calls"])
    print("RESPONSE:", trace["outputs"]["messages"][-1]["content"])
    print("---")

故障范围已缩小到 wrong_tool 类型的情况，循环会打印前三个案例的用户查询、实际调用的工具以及最终响应。此时，人工或更智能的大型语言模型可以分析这些故障并形成假设，例如认为所有包含“subscription”字样的查询都会触发 get_recent_orders 而非 get_account_info。该假设便成为修复的目标。

预期输出：

QUERY: Is my subscription active?
TOOLS CALLED: ['get_recent_orders']
RESPONSE: I could not find any recent orders to confirm your subscription status.
---
QUERY: Tell me about my subscription to your product
TOOLS CALLED: ['get_recent_orders']
RESPONSE: There are no recent orders for this account.
---

屏幕上的三个示例让这一模式一目了然。“订阅”在语义上与“账户状态”的关联度高于“订单”，但客服人员却因关键词的表层相似性而做出了错误处理。可行的解决办法是在get_account_info的文档字符串中添加关于订阅的描述，或通过简短的系统提示来阐明二者之间的区别。

该阶段以包含出现频率、分类及示例的故障模式作为结尾。如果下周有其他工程师的修改又破坏了今天已解决的故障模式，那么这样的修复就几乎毫无意义，因此必须通过测试来确保修复的稳定性。

第5阶段：离线测试套件

在这里，故障会转化为静态测试用例。每一个真实的生产环境故障都会被添加到 LangSmith 数据集中，作为包含输入内容（用户查询）、预期行为（正确的工具或可接受的回复）以及可选元数据（如故障类别）的示例。一旦存储，这些数据就永远不会被移除，因此固定的故障无法再次悄然出现。

在 LangSmith 中进行离线评估包含三个部分：

1. 数据集是由多个示例组成的命名集合，每个示例都包含输入内容，以及可选的预期输出。
2. 目标函数是指正在被测试的代码，对于智能体而言通常是其 .invoke 方法。
3. 评估函数则是用于根据每个示例对目标函数的输出进行评分的函数，其结构与评分阶段的评分器类似。

创建数据集

# Step 1: Create a LangSmith dataset for our agent
dataset_name = "agent-production-failures"

dataset = client.create_dataset(
    dataset_name=dataset_name,
    description="Production traces where the agent failed. Each example captures the user query and the expected tool call.",
)
print(f"Created dataset: {dataset.id}")

在收集阶段，同一个client会使用名称和描述调用create_dataset方法，进而返回一个数据集对象，其.id可在后续被引用。实际部署时会通过“若不存在则创建”的检查来确保这一点，但本演示假设是首次运行。

预期输出：

Created dataset: 3f2a1b0c-...

将失败转化为案例

# Step 2: Convert each failure trace into a dataset example
examples = []
for trace in failures:
    query = trace["inputs"]["messages"][0]["content"]
    expected_tool = expected_tool_for_query(query)
    examples.append({
        "inputs": {"question": query},
        "outputs": {"expected_tool": expected_tool},
        "metadata": {"category": categorize(trace), "source_run_id": trace["run_id"]},
    })

client.create_examples(dataset_id=dataset.id, examples=examples)
print(f"Added {len(examples)} examples to {dataset_name}")

该代码会遍历每一个失败轨迹，并构建一个包含三个字段的示例字典。inputs存储用户提出的问题，outputs存储需要测试的行为，即本应调用的工具，而metadata则存储类别和原始运行编号，以便追踪每个示例的来源。通过一次批量调用的client.create_examples即可上传所有内容。

预期输出：

Added 9 examples to agent-production-failures

在实际部署中，通常由审核人员决定预期输出，因为正则表达式启发式方法仅是一个起点。由专人逐一确认每个示例能提供更可靠的真实值，而这正是LangSmith标注队列的作用。

回归评估器

# Step 3: Write an evaluator that checks the agent's tool call against the expected tool
def tool_match_evaluator(inputs: dict, outputs: dict, reference_outputs: dict):
    actual_messages = outputs.get("messages", [])
    tool_called = None
    for m in actual_messages:
        if getattr(m, "tool_calls", None):
            tool_called = m.tool_calls[0]["name"]
            break
    expected = reference_outputs["expected_tool"]
    score = 1.0 if (expected == "any" or tool_called == expected) else 0.0
    return {"key": "tool_match", "score": score}

tool_match_evaluator遵循LangSmith要求的接口规范：它接收示例的inputs、智能体的实际outputs以及示例的reference_outputs。该函数会遍历智能体的输出消息，找到第一个包含工具调用的消息并提取出工具名称，随后将其与预期的工具名称进行比对，最终返回一个包含key和score键值的字典。key值即为在LangSmith中显示的指标名称。其结构遵循LangSmith官方的评估快速入门指南。

目标函数

# Step 4: Wrap our agent in a function that takes a LangSmith example and returns its output
def run_agent(inputs: dict) -> dict:
    result = agent.invoke({
        "messages": [{"role": "user", "content": inputs["question"]}]
    })
    return result

run_agent 被刻意设计得极为简洁。根据数据集示例的inputs，它会调用智能体并返回其返回的原始字典，这正好符合 LangSmith 的evaluate 函数的预期。被调用的agent是在基础阶段创建的。

刻意保持此封装层简洁是有原因的：任何放入其中的代码都会在每次评估时执行，从而增加成本；而直接传递数据则意味着评估器测试的是智能体本身而非封装层。

如此一来，数据集便存在了，评估器也已连接，每次运行失败都会作为永久性的测试用例被记录下来。最终阶段会在任何更改发布之前运行该评估器。

第6阶段：闭环

这就是回归检测机制。智能体会收到一些变更，比如为解决之前发现的“订阅”路由错误而更新的工具描述。在正式发布之前，所有历史故障数据都会通过新版本进行测试。LangSmith会将这两次测试都记录为实验，并逐个展示新版本在哪些方面有所改进，又在哪些方面出现退步。只有当新基线表现优于旧基线且之前已修复的问题没有再次出现退化时，才会进行正式发布。

运行基线评估

# Step 1: Run the evaluator across the dataset with the current agent
from langsmith.evaluation import evaluate

experiment_results = evaluate(
    run_agent,
    data=dataset_name,
    evaluators=[tool_match_evaluator],
    experiment_prefix="agent-v1-baseline",
    max_concurrency=4,
)

evaluate函数从langsmith.evaluation中导入。它被赋予run_agent目标、数据集名称（LangSmith会将其转换为上一阶段的数据集）、评估器列表、用于标记该次运行的实验前缀，以及值为4的并发度，以便并行执行评估。该函数会返回一个实验结果对象，指向LangSmith中的对应运行记录。

预期输出：

View the evaluation results for experiment: 'agent-v1-baseline-...' at:
https://smith.langchain.com/o/.../datasets/.../compare?selectedSessions=...

9/9 runs | avg tool_match: 0.33

在九个失败的测试案例中，当前智能体仍会将其中大部分错误地路由，这就是基准水平。

推送修复方案并再次测试

模式识别阶段建议的修复方法是使用订阅相关表述来重写get_account_info的文档字符串。

# Step 2: Update the tool docstring to teach the agent about subscriptions
@tool
def get_account_info_v2(account_id: str) -> str:
    """Look up basic information for an account, including plan, status,
    subscription tier, and contact email. Use this tool for any question
    about the account itself, including subscription status."""
    fake_accounts = {
        "A100": "Account A100: plan=pro, status=active, email=ada@example.com",
        "A200": "Account A200: plan=free, status=suspended, email=turing@example.com",
    }
    return fake_accounts.get(account_id, f"No account found with id {account_id}")

tools_v2 = [get_account_info_v2, get_recent_orders]
agent_v2 = create_react_agent(model, tools_v2)

def run_agent_v2(inputs: dict) -> dict:
    result = agent_v2.invoke({
        "messages": [{"role": "user", "content": inputs["question"]}]
    })
    return result

账户信息工具经过重新设计，增加了更详细的文档字符串，明确说明了订阅状态。由于大语言模型选择工具时会读取文档字符串，因此扩展该字符串可直接解决路由失败的问题。代理程序使用更新后的工具重新构建，并被封装在新的run_agent_v2函数中。其他部分保持不变。

# Step 3: Run the evaluator on the v2 agent
experiment_results_v2 = evaluate(
    run_agent_v2,
    data=dataset_name,
    evaluators=[tool_match_evaluator],
    experiment_prefix="agent-v2-docstring-fix",
    max_concurrency=4,
)

使用相同的评估工具在相同的数据集上再次运行相同的评估，仅目标内容会发生变化。新的experiment_prefix用于在LangSmith界面中区分不同的运行次数，以便与基准结果进行对比。LangSmith会让每个示例通过v2智能体处理，存储完整的执行轨迹，并用评估工具对每个输出结果进行评分。

预期输出：

View the evaluation results for experiment: 'agent-v2-docstring-fix-...' at:
https://smith.langchain.com/o/.../datasets/.../compare?selectedSessions=...

9/9 runs | avg tool_match: 0.89

得分从0.33上升到了0.89。九次失败中有八次已得到解决，第九次则是一个极其模糊的查询，或许应该放入人工标注队列中处理。

Before (v1):       After (v2):
wrong_tool: 5      wrong_tool: 1
unhelpful: 3       unhelpful: 0
human: 1           human: 0
avg score: 0.33    avg score: 0.89

有几项因素使得这项比较具有可信度：

1. 在修改任何代码之前，基准评估已经得出了具体的分数。
2. 修复方式仅是一次精简的修改，即添加一条文档字符串。
3. 重新评估时使用了一组精心挑选的实际故障案例来衡量该修改的准确效果。
4. 测试数据集保持不变，因此比较是在完全相同的条件下进行的。
5. 每条原始故障记录都被固定为测试用例，这样后续的任何修改都不会悄无声息地破坏它们。

这正是循环复利效应的体现。每次修复都会为数据集添加新的案例，而后续版本则需要与所有历史故障案例进行对比。随着数据集的不断完善，发布速度会越来越快，而不会变慢。

现在整个循环可以从头到尾完整运行。系统会收集各种轨迹数据并为其添加评分，潜在模式会显现出来，故障会被转化为离线测试，评估会在产品发布之前进行，只有当基准值上升时才会推进变更。每个循环都在更高的起点上开始。

一次完整的每日运行

在全新的生产数据日，单个驱动程序即可完成整个循环的一次运行。

# Full loop driver
def run_improvement_loop():
    # 1. Pull traces
    runs = list(client.list_runs(
        project_name="agent-improvement-loop",
        is_root=True,
        start_time=datetime.utcnow() - timedelta(days=1),
    ))
    traces = [compact_trace(r) for r in runs]

    # 2. Enrich with scores
    enriched = [enrich(t) for t in traces]

    # 3. Find failures and categorize them
    failures = [t for t in enriched if is_low_scoring(t)]
    categories = Counter(categorize(t) for t in failures)

    # 4. Add failures to the dataset
    new_examples = [{
        "inputs": {"question": t["inputs"]["messages"][0]["content"]},
        "outputs": {"expected_tool": expected_tool_for_query(
            t["inputs"]["messages"][0]["content"])},
        "metadata": {"category": categorize(t), "source_run_id": t["run_id"]},
    } for t in failures]

    if new_examples:
        client.create_examples(dataset_id=dataset.id, examples=new_examples)

    # 5. Re-evaluate the current agent against the full dataset
    result = evaluate(
        run_agent_v2,
        data=dataset_name,
        evaluators=[tool_match_evaluator],
        experiment_prefix="daily-regression",
    )

    return {
        "traces_collected": len(traces),
        "failures_found": len(failures),
        "by_category": dict(categories),
        "examples_added": len(new_examples),
    }

print(run_improvement_loop())

该函数将六个阶段整合在一起。它获取昨天的追踪数据，对其进行评分，对故障进行分类，将新的故障作为数据集示例添加进去，并进行回归评估。它会返回一个小型控制面板字典，可用于每日记录或发送到 Slack。

预期输出：

{
  "traces_collected": 186,
  "failures_found": 12,
  "by_category": {"wrong_tool": 4, "unhelpful_answer": 6, "human_flagged": 2},
  "examples_added": 12,
  "regression_score": 0.87
}

输出结果显示：当天共产生了186条生成记录，代理代码未发生任何变化，其中12条的评分至少为较低水平。目前无用答案的数量超过了错误工具调用的数量，这表明路由优化已见成效，接下来的改进应集中在响应生成方面。回归评分仍为0.87，说明本周的所有改动均未引发之前的故障。又有12个新示例被加入永久数据集，用以防止重复出现。

接下来值得进行的五项升级

这些改进均对应LangChain、LangGraph或LangSmith中的某些功能，无需重新构建现有架构即可启用。

1. 用嵌入模型替代正则表达式分类器。OpenAIEmbeddings能够为每个查询生成嵌入向量，从而使故障轨迹根据语义相似度而非关键词规则进行聚类，从而发现诸如语气不匹配等更细微的问题。
2. 用OpenEvals库替换内置的LLM评估器。LangSmith的快速入门指南使用了带有预置提示语如CORRECTNESS_PROMPT的openevals，其效果优于手动设计的评估器，并且还提供了无需参考标准的版本用于在线监控。
3. 添加多轮评估功能。LangSmith的多轮评估会针对整个对话过程而非单条回复来运行LLM评估器，从而发现基于聊天的智能体在单轮评估中可能遗漏的推理偏差。

引入 LangChain 的 Insights Agent 实现大规模的模式识别。正则表达式分类器虽能处理数十次故障，但面对数千次故障时仍显力不从心，此时 Insights Agent 会自动将追踪数据集中到相关项目组中，识别故障模式，并使用相同的 数据集结构进行分析。

通过分阶段部署的 LangGraph 方式逐步推广该功能。先将 v2 版本放在设置为 LANGSMITH_PROJECT=agent-canary 的测试项目中运行一天，再将其实际表现与对照项目进行比较，只有当其性能明显更优时才会替换生产环境中的版本。

如此循环往复，每次都会以更高的基准水平为起点，从而获得更优质的追踪数据、更精准的评估结果以及更出色的智能体。