评分标准

RubricMiddleware 需要 deepagents>=0.6.5。它处于 测试版；API 未来可能会变更。

某些智能体任务对于“完成”有明确的定义，但工作模型本身在首次尝试时通常无法可靠地达成：一首符合正确音节模式的俳句、一次所有测试都通过的重构，或一份涵盖所有必填部分的报告。RubricMiddleware 让您可以将完成的样子声明为评分标准，并让智能体 自我评估和迭代，直到满足评分标准，或达到配置的最大迭代次数上限。 LLM-as-a-judge 是一种模式，其中一个语言模型根据定义的标准评估另一个模型的输出。在 LangSmith 评估中，LLM-as-a-judge 评估器在批量离线评估中对应用输出进行评分。RubricMiddleware 在运行时应用相同的模式：在深度智能体生成输出后，一个专门的评分器模型会根据您的评分标准审查对话记录，并驱动修改，直到每个标准都通过（或达到配置的迭代上限）。当深度智能体完成推理后，LLM-as-a-judge 评分器子代理会审查输出并返回裁决。如果返回 needs_revision，则针对每个标准的反馈会被注入回对话中，智能体将再次运行。循环在 satisfied、max_iterations_reached、failed 或 grader_error 时终止。

配置中间件

在调用 create_deep_agent 时，将 RubricMiddleware 添加到 middleware 列表中：

from deepagents import RubricMiddleware, create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    middleware=[
        RubricMiddleware(
            model="anthropic:claude-haiku-4-5",
            max_iterations=3,
        ),
    ],
    checkpointer=InMemorySaver(),
)

from deepagents import RubricMiddleware, create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_deep_agent(
    model="openai:gpt-5.5",
    middleware=[
        RubricMiddleware(
            model="anthropic:claude-haiku-4-5",
            max_iterations=3,
        ),
    ],
    checkpointer=InMemorySaver(),
)

from deepagents import RubricMiddleware, create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    middleware=[
        RubricMiddleware(
            model="anthropic:claude-haiku-4-5",
            max_iterations=3,
        ),
    ],
    checkpointer=InMemorySaver(),
)

from deepagents import RubricMiddleware, create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_deep_agent(
    model="openrouter:z-ai/glm-5.2",
    middleware=[
        RubricMiddleware(
            model="anthropic:claude-haiku-4-5",
            max_iterations=3,
        ),
    ],
    checkpointer=InMemorySaver(),
)

from deepagents import RubricMiddleware, create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_deep_agent(
    model="fireworks:accounts/fireworks/models/glm-5p2",
    middleware=[
        RubricMiddleware(
            model="anthropic:claude-haiku-4-5",
            max_iterations=3,
        ),
    ],
    checkpointer=InMemorySaver(),
)

from deepagents import RubricMiddleware, create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_deep_agent(
    model="baseten:zai-org/GLM-5.2",
    middleware=[
        RubricMiddleware(
            model="anthropic:claude-haiku-4-5",
            max_iterations=3,
        ),
    ],
    checkpointer=InMemorySaver(),
)

from deepagents import RubricMiddleware, create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_deep_agent(
    model="ollama:north-mini-code-1.0",
    middleware=[
        RubricMiddleware(
            model="anthropic:claude-haiku-4-5",
            max_iterations=3,
        ),
    ],
    checkpointer=InMemorySaver(),
)

参数	必填	默认值	描述
`model`	是	`None`	LLM-as-a-judge 评分器子代理使用的聊天模型。接受 `"provider:model-id"` 字符串或 `BaseChatModel` 实例。通常选择比深度智能体的工作模型更小或更便宜的模型。
`system_prompt`	否	内置评分器提示	自定义评分指令。如果未提供，则回退到默认系统提示，该提示会教导评分器裁决格式及其可使用的工具。
`tools`	否	`None`	评分器在生成裁决前可用于收集证据（运行测试、统计令牌数、读取文件）的工具。如果没有提供，评分器将仅根据对话记录进行推理。
`max_iterations`	否	`3`	每次评分标准尝试的最大评分器迭代次数；必须是正整数。达到上限但未获得 `satisfied` 裁决时，智能体将以状态 `max_iterations_reached` 终止。
`on_evaluation`	否	`None`	可选的回调函数，在每次评分迭代后（无论是使用 `invoke()`、`stream()` 还是 `stream_events()`）调用，并接收 `RubricEvaluation`。用于日志记录、自定义指标、评估数据集或 UI 更新。

在调用时传递评分标准

在调用状态中传递一个 rubric 字符串以启动自我评估循环。使用 invoke() 进行单次阻塞调用，或使用 stream_events(..., version="v3") 配合 CustomTransformer 在 stream.custom 上实时接收评分事件：

invoke()
stream_events()

from langchain.messages import HumanMessage

config = {"configurable": {"thread_id": "my-rubric-thread"}}
result = agent.invoke(
    {
        "messages": [HumanMessage("Write a haiku about spring.")],
        "rubric": (
            "- The poem has three lines\n"
            "- Lines follow a 5-7-5 syllable pattern\n"
            "- The theme is spring"
        ),
    },
    config=config,
)

from langchain.messages import HumanMessage
from langgraph.stream import CustomTransformer

config = {"configurable": {"thread_id": "my-rubric-thread"}}
stream = agent.stream_events(
    {
        "messages": [HumanMessage("Write a haiku about spring.")],
        "rubric": (
            "- The poem has three lines\n"
            "- Lines follow a 5-7-5 syllable pattern\n"
            "- The theme is spring"
        ),
    },
    config=config,
    version="v3",
    transformers=[CustomTransformer],
)

for event in stream.custom:
    event_type = event.get("type")
    if event_type == "rubric_evaluation_start":
        print(
            f"Grading iteration {event['iteration']} "
            f"(run {event['grading_run_id']})"
        )
    elif event_type == "rubric_evaluation_end":
        print(f"Verdict: {event['result']} — {event.get('explanation', '')}")

评分标准评分会在 stream.custom 上触发以下自定义事件：

事件	触发时机	负载字段
`rubric_evaluation_start`	评分器运行前。	`type`: 事件名称 `grading_run_id`: 在一次评分标准尝试的所有事件中共享 `iteration`: 当前评分运行的从零开始的索引
`rubric_evaluation_end`	评分器返回后或评分器发生异常后。	`type`: 事件名称 `grading_run_id`: 在一次评分标准尝试的所有事件中共享 `iteration`: 当前评分器遍历的从零开始的索引 `result`: 本次遍历的最终裁决 `explanation`: 来自评分器的摘要 `criteria`: 针对每个标准的裁决

评分标准裁决

当深度智能体完成推理并产生输出后，LLM-as-a-judge 评分器子代理会根据评分标准审查输出，并产生以下裁决之一：

状态	含义	是否循环回？
`satisfied`	评分标准中的每个标准都通过。	否
`needs_revision`	至少一个标准未通过；评分器的反馈会被注入，智能体将再次运行。	是
`max_iterations_reached`	评分器仍希望修改，但已达到 `max_iterations`。	否
`failed`	评分器判定评分标准格式错误或无法根据对话记录进行评估。	否
`grader_error`	LLM-as-a-judge 评分器子代理本身抛出异常（提供者超时、缺少凭据、结构化响应格式错误等）。	否

观察迭代进度

on_evaluation 是一个回调函数，在每次评分迭代后（无论是调用 invoke() 还是 stream_events()）触发，并带有评分器的裁决。如果您没有从 stream.custom（使用 CustomTransformer）读取评分标准事件，或没有使用 LangSmith 追踪运行，那么这是检查评分过程中发生了什么的主要方式。

from deepagents import RubricMiddleware, create_deep_agent
from deepagents.middleware.rubric import RubricEvaluation
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver


def log_evaluation(ev: RubricEvaluation) -> None:
    print(f"迭代 {ev['iteration']}: {ev['result']} — {ev['explanation']}")


agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    middleware=[
        RubricMiddleware(
            model="anthropic:claude-haiku-4-5",
            on_evaluation=log_evaluation,
        ),
    ],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "rubric-eval-session"}}
agent.invoke(
    {
        "messages": [HumanMessage("用一句话总结光合作用。")],
        "rubric": (
            "- 答案必须是一句话\n"
            "- 答案中需提及光和叶绿素"
        ),
    },
    config=config,
)

from deepagents import RubricMiddleware, create_deep_agent
from deepagents.middleware.rubric import RubricEvaluation
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver


def log_evaluation(ev: RubricEvaluation) -> None:
    print(f"迭代 {ev['iteration']}: {ev['result']} — {ev['explanation']}")


agent = create_deep_agent(
    model="openai:gpt-5.5",
    middleware=[
        RubricMiddleware(
            model="anthropic:claude-haiku-4-5",
            on_evaluation=log_evaluation,
        ),
    ],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "rubric-eval-session"}}
agent.invoke(
    {
        "messages": [HumanMessage("用一句话总结光合作用。")],
        "rubric": (
            "- 答案必须是一句话\n"
            "- 答案中需提及光和叶绿素"
        ),
    },
    config=config,
)

from deepagents import RubricMiddleware, create_deep_agent
from deepagents.middleware.rubric import RubricEvaluation
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver


def log_evaluation(ev: RubricEvaluation) -> None:
    print(f"迭代 {ev['iteration']}: {ev['result']} — {ev['explanation']}")


agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    middleware=[
        RubricMiddleware(
            model="anthropic:claude-haiku-4-5",
            on_evaluation=log_evaluation,
        ),
    ],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "rubric-eval-session"}}
agent.invoke(
    {
        "messages": [HumanMessage("用一句话总结光合作用。")],
        "rubric": (
            "- 答案必须是一句话\n"
            "- 答案中需提及光和叶绿素"
        ),
    },
    config=config,
)

from deepagents import RubricMiddleware, create_deep_agent
from deepagents.middleware.rubric import RubricEvaluation
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver


def log_evaluation(ev: RubricEvaluation) -> None:
    print(f"迭代 {ev['iteration']}: {ev['result']} — {ev['explanation']}")


agent = create_deep_agent(
    model="openrouter:z-ai/glm-5.2",
    middleware=[
        RubricMiddleware(
            model="anthropic:claude-haiku-4-5",
            on_evaluation=log_evaluation,
        ),
    ],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "rubric-eval-session"}}
agent.invoke(
    {
        "messages": [HumanMessage("用一句话总结光合作用。")],
        "rubric": (
            "- 答案必须是一句话\n"
            "- 答案中需提及光和叶绿素"
        ),
    },
    config=config,
)

from deepagents import RubricMiddleware, create_deep_agent
from deepagents.middleware.rubric import RubricEvaluation
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver


def log_evaluation(ev: RubricEvaluation) -> None:
    print(f"迭代 {ev['iteration']}: {ev['result']} — {ev['explanation']}")


agent = create_deep_agent(
    model="fireworks:accounts/fireworks/models/glm-5p2",
    middleware=[
        RubricMiddleware(
            model="anthropic:claude-haiku-4-5",
            on_evaluation=log_evaluation,
        ),
    ],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "rubric-eval-session"}}
agent.invoke(
    {
        "messages": [HumanMessage("用一句话总结光合作用。")],
        "rubric": (
            "- 答案必须是一句话\n"
            "- 答案中需提及光和叶绿素"
        ),
    },
    config=config,
)

from deepagents import RubricMiddleware, create_deep_agent
from deepagents.middleware.rubric import RubricEvaluation
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver


def log_evaluation(ev: RubricEvaluation) -> None:
    print(f"迭代 {ev['iteration']}: {ev['result']} — {ev['explanation']}")


agent = create_deep_agent(
    model="baseten:zai-org/GLM-5.2",
    middleware=[
        RubricMiddleware(
            model="anthropic:claude-haiku-4-5",
            on_evaluation=log_evaluation,
        ),
    ],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "rubric-eval-session"}}
agent.invoke(
    {
        "messages": [HumanMessage("用一句话总结光合作用。")],
        "rubric": (
            "- 答案必须是一句话\n"
            "- 答案中需提及光和叶绿素"
        ),
    },
    config=config,
)

from deepagents import RubricMiddleware, create_deep_agent
from deepagents.middleware.rubric import RubricEvaluation
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver


def log_evaluation(ev: RubricEvaluation) -> None:
    print(f"迭代 {ev['iteration']}: {ev['result']} — {ev['explanation']}")


agent = create_deep_agent(
    model="ollama:north-mini-code-1.0",
    middleware=[
        RubricMiddleware(
            model="anthropic:claude-haiku-4-5",
            on_evaluation=log_evaluation,
        ),
    ],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "rubric-eval-session"}}
agent.invoke(
    {
        "messages": [HumanMessage("用一句话总结光合作用。")],
        "rubric": (
            "- 答案必须是一句话\n"
            "- 答案中需提及光和叶绿素"
        ),
    },
    config=config,
)

中间件在每次评分器遍历后使用 RubricEvaluation 字典调用您的函数。RubricEvaluation 字典包含：

字段	类型	描述
`grading_run_id`	`str`	在一次评分标准尝试中，所有评估共享的标识符。当调用者提供不同的 `rubric`，或在最终裁决后再次调用相同的 `rubric` 时，会开始新的运行。
`iteration`	`int`	在该运行中，当前评分器遍历的从零开始的索引。
`result`	`str`	本次遍历的评分器裁决：`satisfied`、`needs_revision`、`failed` 或 `grader_error`。
`explanation`	`str`	来自评分器的自由格式摘要。在基础设施故障时，这包括异常类型和消息。
`criteria`	`list`	针对每个标准的裁决。每个条目是 `{name, passed: true}` 或 `{name, passed: false, gap}`，其中 `gap` 是针对未通过标准的可操作反馈。

评分器遍历事件

事件	描述
评分成功	每次遍历触发一次，包括中间的 `needs_revision` 裁决和最终的 `satisfied` 或 `failed` 裁决。当评分器返回 `needs_revision` 但已达到 `max_iterations` 时，回调仍会收到 `result: "needs_revision"`（评分器的裁决）。该运行的最终状态是 `max_iterations_reached`，存储在私有状态 `_rubric_status` 上，而不是评估记录上。在 `invoke` 完成后检查 `_rubric_status`，或结合 `_rubric_iterations` 读取 `_rubric_evaluations` 中的最后一项，以判断是否达到上限。
评分器异常	触发时带有 `result: "grader_error"`、根据异常派生的 `explanation` 以及空的 `criteria` 列表。
您的回调中的错误	异常会被记录并抑制。评分循环将继续。不要使用 `on_evaluation` 来强制控制流（例如，抛出异常以停止智能体）。

跨调用持久化评分标准

单次 agent.invoke() 或 agent.stream_events() 调用会运行评分标准循环直至完成，并以最终裁决结束：satisfied、failed 或 max_iterations_reached。要将评分标准传递给后续调用，请附加一个检查点器，并在调用时传递相同的 thread_id。在这些情况下，相同的 rubric 会在未来的 invoke() 或 stream_events() 调用中保持，直到您传递一个新的。中断（KeyboardInterrupt、asyncio.CancelledError）会从 agent.invoke() 和 agent.stream_events() 中未捕获地传播出来。在已设置检查点的线程上，使用相同评分标准的下一次调用将恢复进行中的评分运行。

示例：生成经过审查的 Python 代码

以下示例构建了一个编写 find_duplicates 函数的深度智能体。它定义 RubricMiddleware 一次，将其附加到智能体，然后在调用时传递一个 rubric 字符串。本示例没有让评分器抽象地推理正确性，而是给了它一个 run_test_suite 工具来直接验证行为。评分器在生成裁决前会调用此工具以获取额外信息，当没有提供工具时，则回退到从对话记录推理。

定义 RubricMiddleware

此中间件在基础智能体之上添加了一个 LLM-as-a-judge 评分器循环。配置评分器模型、可选的自定义提示、用于证据收集的工具以及最大迭代次数上限。

from deepagents import RubricMiddleware
from langchain.tools import tool


@tool
def run_test_suite(code: str) -> dict:
    """运行针对Python源代码的find_duplicates测试套件。"""
    namespace: dict = {"__builtins__": __builtins__}
    try:
        exec(code, namespace)
    except Exception as exc:
        return {"ok": False, "failures": [f"执行代码失败: {exc}"]}

    find_duplicates = namespace.get("find_duplicates")
    if find_duplicates is None:
        return {"ok": False, "failures": ["函数 find_duplicates 未定义"]}

    tests = [
        ("test_basic", [1, 2, 2, 3, 1], [2, 1]),
        ("test_empty", [], []),
        ("test_no_duplicates", [1, 2, 3], []),
        ("test_unhashable", [[1], [1], 2], [[1]]),
    ]
    failures: list[str] = []
    for name, args, expected in tests:
        try:
            actual = find_duplicates(args)
            if actual != expected:
                failures.append(f"{name}: 期望 {expected}, 得到 {actual}")
        except Exception as exc:
            failures.append(f"{name}: {exc}")

    return {"ok": not failures, "failures": failures}


rubric_middleware = RubricMiddleware(
    model="google_genai:gemini-3.6-flash",
    system_prompt="您是一位代码评审员，正在根据评分标准对生成的代码进行评分。",
    tools=[run_test_suite],
    max_iterations=5,
)

from deepagents import RubricMiddleware
from langchain.tools import tool


@tool
def run_test_suite(code: str) -> dict:
    """运行针对Python源代码的find_duplicates测试套件。"""
    namespace: dict = {"__builtins__": __builtins__}
    try:
        exec(code, namespace)
    except Exception as exc:
        return {"ok": False, "failures": [f"执行代码失败: {exc}"]}

    find_duplicates = namespace.get("find_duplicates")
    if find_duplicates is None:
        return {"ok": False, "failures": ["函数 find_duplicates 未定义"]}

    tests = [
        ("test_basic", [1, 2, 2, 3, 1], [2, 1]),
        ("test_empty", [], []),
        ("test_no_duplicates", [1, 2, 3], []),
        ("test_unhashable", [[1], [1], 2], [[1]]),
    ]
    failures: list[str] = []
    for name, args, expected in tests:
        try:
            actual = find_duplicates(args)
            if actual != expected:
                failures.append(f"{name}: 期望 {expected}, 得到 {actual}")
        except Exception as exc:
            failures.append(f"{name}: {exc}")

    return {"ok": not failures, "failures": failures}


rubric_middleware = RubricMiddleware(
    model="openai:gpt-5.5",
    system_prompt="您是一位代码评审员，正在根据评分标准对生成的代码进行评分。",
    tools=[run_test_suite],
    max_iterations=5,
)

from deepagents import RubricMiddleware
from langchain.tools import tool


@tool
def run_test_suite(code: str) -> dict:
    """运行针对Python源代码的find_duplicates测试套件。"""
    namespace: dict = {"__builtins__": __builtins__}
    try:
        exec(code, namespace)
    except Exception as exc:
        return {"ok": False, "failures": [f"执行代码失败: {exc}"]}

    find_duplicates = namespace.get("find_duplicates")
    if find_duplicates is None:
        return {"ok": False, "failures": ["函数 find_duplicates 未定义"]}

    tests = [
        ("test_basic", [1, 2, 2, 3, 1], [2, 1]),
        ("test_empty", [], []),
        ("test_no_duplicates", [1, 2, 3], []),
        ("test_unhashable", [[1], [1], 2], [[1]]),
    ]
    failures: list[str] = []
    for name, args, expected in tests:
        try:
            actual = find_duplicates(args)
            if actual != expected:
                failures.append(f"{name}: 期望 {expected}, 得到 {actual}")
        except Exception as exc:
            failures.append(f"{name}: {exc}")

    return {"ok": not failures, "failures": failures}


rubric_middleware = RubricMiddleware(
    model="anthropic:claude-sonnet-4-6",
    system_prompt="您是一位代码评审员，正在根据评分标准对生成的代码进行评分。",
    tools=[run_test_suite],
    max_iterations=5,
)

from deepagents import RubricMiddleware
from langchain.tools import tool


@tool
def run_test_suite(code: str) -> dict:
    """运行针对Python源代码的find_duplicates测试套件。"""
    namespace: dict = {"__builtins__": __builtins__}
    try:
        exec(code, namespace)
    except Exception as exc:
        return {"ok": False, "failures": [f"执行代码失败: {exc}"]}

    find_duplicates = namespace.get("find_duplicates")
    if find_duplicates is None:
        return {"ok": False, "failures": ["函数 find_duplicates 未定义"]}

    tests = [
        ("test_basic", [1, 2, 2, 3, 1], [2, 1]),
        ("test_empty", [], []),
        ("test_no_duplicates", [1, 2, 3], []),
        ("test_unhashable", [[1], [1], 2], [[1]]),
    ]
    failures: list[str] = []
    for name, args, expected in tests:
        try:
            actual = find_duplicates(args)
            if actual != expected:
                failures.append(f"{name}: 期望 {expected}, 得到 {actual}")
        except Exception as exc:
            failures.append(f"{name}: {exc}")

    return {"ok": not failures, "failures": failures}


rubric_middleware = RubricMiddleware(
    model="openrouter:z-ai/glm-5.2",
    system_prompt="您是一位代码评审员，正在根据评分标准对生成的代码进行评分。",
    tools=[run_test_suite],
    max_iterations=5,
)

from deepagents import RubricMiddleware
from langchain.tools import tool


@tool
def run_test_suite(code: str) -> dict:
    """运行针对Python源代码的find_duplicates测试套件。"""
    namespace: dict = {"__builtins__": __builtins__}
    try:
        exec(code, namespace)
    except Exception as exc:
        return {"ok": False, "failures": [f"执行代码失败: {exc}"]}

    find_duplicates = namespace.get("find_duplicates")
    if find_duplicates is None:
        return {"ok": False, "failures": ["函数 find_duplicates 未定义"]}

    tests = [
        ("test_basic", [1, 2, 2, 3, 1], [2, 1]),
        ("test_empty", [], []),
        ("test_no_duplicates", [1, 2, 3], []),
        ("test_unhashable", [[1], [1], 2], [[1]]),
    ]
    failures: list[str] = []
    for name, args, expected in tests:
        try:
            actual = find_duplicates(args)
            if actual != expected:
                failures.append(f"{name}: 期望 {expected}, 得到 {actual}")
        except Exception as exc:
            failures.append(f"{name}: {exc}")

    return {"ok": not failures, "failures": failures}


rubric_middleware = RubricMiddleware(
    model="fireworks:accounts/fireworks/models/glm-5p2",
    system_prompt="您是一位代码评审员，正在根据评分标准对生成的代码进行评分。",
    tools=[run_test_suite],
    max_iterations=5,
)

from deepagents import RubricMiddleware
from langchain.tools import tool


@tool
def run_test_suite(code: str) -> dict:
    """运行针对Python源代码的find_duplicates测试套件。"""
    namespace: dict = {"__builtins__": __builtins__}
    try:
        exec(code, namespace)
    except Exception as exc:
        return {"ok": False, "failures": [f"执行代码失败: {exc}"]}

    find_duplicates = namespace.get("find_duplicates")
    if find_duplicates is None:
        return {"ok": False, "failures": ["函数 find_duplicates 未定义"]}

    tests = [
        ("test_basic", [1, 2, 2, 3, 1], [2, 1]),
        ("test_empty", [], []),
        ("test_no_duplicates", [1, 2, 3], []),
        ("test_unhashable", [[1], [1], 2], [[1]]),
    ]
    failures: list[str] = []
    for name, args, expected in tests:
        try:
            actual = find_duplicates(args)
            if actual != expected:
                failures.append(f"{name}: 期望 {expected}, 得到 {actual}")
        except Exception as exc:
            failures.append(f"{name}: {exc}")

    return {"ok": not failures, "failures": failures}


rubric_middleware = RubricMiddleware(
    model="baseten:zai-org/GLM-5.2",
    system_prompt="您是一位代码评审员，正在根据评分标准对生成的代码进行评分。",
    tools=[run_test_suite],
    max_iterations=5,
)

from deepagents import RubricMiddleware
from langchain.tools import tool


@tool
def run_test_suite(code: str) -> dict:
    """运行针对Python源代码的find_duplicates测试套件。"""
    namespace: dict = {"__builtins__": __builtins__}
    try:
        exec(code, namespace)
    except Exception as exc:
        return {"ok": False, "failures": [f"执行代码失败: {exc}"]}

    find_duplicates = namespace.get("find_duplicates")
    if find_duplicates is None:
        return {"ok": False, "failures": ["函数 find_duplicates 未定义"]}

    tests = [
        ("test_basic", [1, 2, 2, 3, 1], [2, 1]),
        ("test_empty", [], []),
        ("test_no_duplicates", [1, 2, 3], []),
        ("test_unhashable", [[1], [1], 2], [[1]]),
    ]
    failures: list[str] = []
    for name, args, expected in tests:
        try:
            actual = find_duplicates(args)
            if actual != expected:
                failures.append(f"{name}: 期望 {expected}, 得到 {actual}")
        except Exception as exc:
            failures.append(f"{name}: {exc}")

    return {"ok": not failures, "failures": failures}


rubric_middleware = RubricMiddleware(
    model="ollama:north-mini-code-1.0",
    system_prompt="您是一位代码评审员，正在根据评分标准对生成的代码进行评分。",
    tools=[run_test_suite],
    max_iterations=5,
)

将其传递给深度智能体

智能体的 system_prompt 告诉它如何工作，而评分标准告诉评分器如何评判工作。

from deepagents import create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    system_prompt=(
        "你是一名谨慎的Python工程师。编写正确、可读的代码。"
        "严格遵守用户的指令。"
    ),
    middleware=[rubric_middleware],
    checkpointer=InMemorySaver(),
)

from deepagents import create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_deep_agent(
    model="openai:gpt-5.5",
    system_prompt=(
        "你是一名谨慎的Python工程师。编写正确、可读的代码。"
        "严格遵守用户的指令。"
    ),
    middleware=[rubric_middleware],
    checkpointer=InMemorySaver(),
)

from deepagents import create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    system_prompt=(
        "你是一名谨慎的Python工程师。编写正确、可读的代码。"
        "严格遵守用户的指令。"
    ),
    middleware=[rubric_middleware],
    checkpointer=InMemorySaver(),
)

from deepagents import create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_deep_agent(
    model="openrouter:z-ai/glm-5.2",
    system_prompt=(
        "你是一名谨慎的Python工程师。编写正确、可读的代码。"
        "严格遵守用户的指令。"
    ),
    middleware=[rubric_middleware],
    checkpointer=InMemorySaver(),
)

from deepagents import create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_deep_agent(
    model="fireworks:accounts/fireworks/models/glm-5p2",
    system_prompt=(
        "你是一名谨慎的Python工程师。编写正确、可读的代码。"
        "严格遵守用户的指令。"
    ),
    middleware=[rubric_middleware],
    checkpointer=InMemorySaver(),
)

from deepagents import create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_deep_agent(
    model="baseten:zai-org/GLM-5.2",
    system_prompt=(
        "你是一名谨慎的Python工程师。编写正确、可读的代码。"
        "严格遵守用户的指令。"
    ),
    middleware=[rubric_middleware],
    checkpointer=InMemorySaver(),
)

from deepagents import create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_deep_agent(
    model="ollama:north-mini-code-1.0",
    system_prompt=(
        "你是一名谨慎的Python工程师。编写正确、可读的代码。"
        "严格遵守用户的指令。"
    ),
    middleware=[rubric_middleware],
    checkpointer=InMemorySaver(),
)

使用用户消息和评分标准进行调用

在调用时，在 messages 中提供用户请求，并在 rubric 中提供一个换行符分隔的清单，评分器必须将其标记为已满足。如果在输入状态中未提供 rubric，中间件将不会运行。

from langchain.messages import HumanMessage

result = agent.invoke(
    {
        "messages": [
            HumanMessage(
                content=(
                    "编写一个 Python 函数 `find_duplicates(lst)`，该函数返回输入列表中所有出现超过一次的元素，按首次出现的顺序排列。"
                )
            )
        ],
        "rubric": (
            "- 所有测试在 run_test_suite 中通过\n"
            "- 函数名为 `find_duplicates` 并接受单个列表参数\n"
        ),
    },
    config={"configurable": {"thread_id": "code-generation-session"}},
)
print(result["messages"][-1].text)

智能体生成输出后，评分器接管并检查输出是否满足每个标准：例如，当输入包含不可哈希类型时，test_unhashable 是否抛出 TypeError 失败。如果有任何问题，评分器会提供此反馈，然后智能体修改其实现并返回给评分器。

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.

Get started

Deployment

Execution environment

Context management

Delegation

Steering

Frontend

Protocols

配置中间件

在调用时传递评分标准

评分标准裁决

观察迭代进度

评分器遍历事件

跨调用持久化评分标准

示例：生成经过审查的 Python 代码

​配置中间件

​在调用时传递评分标准

​评分标准裁决

​观察迭代进度

​评分器遍历事件

​跨调用持久化评分标准

​示例：生成经过审查的 Python 代码

配置中间件

在调用时传递评分标准

评分标准裁决

观察迭代进度

评分器遍历事件

跨调用持久化评分标准

示例：生成经过审查的 Python 代码