INFO

RubricMiddleware 需要 deepagents>=0.6.5。它处于beta阶段；API 可能在未来发生变化。

某些智能体任务有一个清晰的"完成"定义，而工作模型本身无法在第一次尝试中可靠地达到：音节模式正确的俳句、所有测试都通过的重构，或包含所有必需部分的报告。RubricMiddleware 让你将完成的样子声明为评分准则，并让智能体自我评估和迭代，直到准则得到满足，或达到配置的最大迭代上限。

LLM-as-a-judge（LLM 作为评审） 是一种模式，其中一个语言模型根据定义的标准评估另一个模型的输出。在 LangSmith 评估中，LLM-as-a-judge 评估器在批量离线状态下对应用程序输出评分。RubricMiddleware 在运行时应用相同的模式：深度智能体产生输出后，一个专门的评审模型根据你的评分准则审查记录，并推动修订，直到每个标准都通过（或达到配置的迭代上限）。

当深度智能体完成推理时，LLM-as-a-judge 评审子智能体审查输出并返回判决。如果返回 needs_revision，按标准的反馈会被注入回对话中，智能体再次运行。循环在 satisfied、max_iterations_reached、failed 或 grader_error 时终止。

配置中间件

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

Google

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

OpenAI

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

Anthropic

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

OpenRouter

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

Fireworks

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

Baseten

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

Ollama

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	评审者在产生判决之前可以调用来收集证据（运行测试、统计 token、读取文件）的工具。如果没有，评审者只能根据记录推理。
max_iterations	否	3	每次准则尝试的最大评审迭代次数；必须是正整数。当达到上限而没有 satisfied 判决时，智能体以 max_iterations_reached 状态终止。
on_evaluation	否	None	可选回调，在每次评分迭代后使用每个 RubricEvaluation 调用，无论你使用 invoke()、stream() 还是 stream_events()。用于日志记录、自定义指标、评估数据集或 UI 更新。

在调用时传递准则

在调用状态中传递 rubric 字符串以启动自我评估循环。使用 invoke() 进行单次阻塞调用，或使用带 CustomTransformer 的 stream_events(..., version="v3") 在评分事件发生时通过 stream.custom 接收它们：

invoke()

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

stream_events()

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` | 评审者运行之前。 | <ul><li>`type`：事件名称</li><li>`grading_run_id`：一次准则尝试中的所有事件共享</li><li>`iteration`：当前评分运行的从零开始的索引</li></ul> |
    | `rubric_evaluation_end` | 评审者返回之后或评审异常之后。 | <ul><li>`type`：事件名称</li><li>`grading_run_id`：一次准则尝试中的所有事件共享</li><li>`iteration`：当前评审通过次数的从零开始的索引</li><li>`result`：本次通过的终态判决</li><li>`explanation`：来自评审者的摘要</li><li>`criteria`：按标准的判决</li></ul> |

准则判决

当深度智能体完成推理并产生输出时，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 追踪运行，那么它就是检查评分期间发生情况的主要方式。

Google

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"iteration {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("Write a one-sentence summary of photosynthesis.")],
        "rubric": (
            "- The answer is one sentence\n"
            "- The answer mentions light and chlorophyll"
        ),
    },
    config=config,
)

OpenAI

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"iteration {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("Write a one-sentence summary of photosynthesis.")],
        "rubric": (
            "- The answer is one sentence\n"
            "- The answer mentions light and chlorophyll"
        ),
    },
    config=config,
)

Anthropic

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"iteration {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("Write a one-sentence summary of photosynthesis.")],
        "rubric": (
            "- The answer is one sentence\n"
            "- The answer mentions light and chlorophyll"
        ),
    },
    config=config,
)

OpenRouter

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"iteration {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("Write a one-sentence summary of photosynthesis.")],
        "rubric": (
            "- The answer is one sentence\n"
            "- The answer mentions light and chlorophyll"
        ),
    },
    config=config,
)

Fireworks

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"iteration {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("Write a one-sentence summary of photosynthesis.")],
        "rubric": (
            "- The answer is one sentence\n"
            "- The answer mentions light and chlorophyll"
        ),
    },
    config=config,
)

Baseten

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"iteration {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("Write a one-sentence summary of photosynthesis.")],
        "rubric": (
            "- The answer is one sentence\n"
            "- The answer mentions light and chlorophyll"
        ),
    },
    config=config,
)

Ollama

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"iteration {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("Write a one-sentence summary of photosynthesis.")],
        "rubric": (
            "- The answer is one sentence\n"
            "- The answer mentions light and chlorophyll"
        ),
    },
    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"（评审者的判决）。该运行的终态状态是私有状态 _rubric_status 上的 max_iterations_reached，而不是评估记录上的。在 invoke 完成后检查 _rubric_status，或结合 _rubric_iterations 读取 _rubric_evaluations 中的最后一条条目，以便根据上限耗尽进行分支。
评审异常	以 result: "grader_error"、从异常派生的说明和空的 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 评审循环。配置评审模型、可选的自定义提示词、用于收集证据的工具和最大迭代上限。

Google

from deepagents import RubricMiddleware
from langchain.tools import tool

@tool
def run_test_suite(code: str) -> dict:
    """Run the find_duplicates test suite against Python source code."""
    namespace: dict = {"__builtins__": __builtins__}
    try:
        exec(code, namespace)
    except Exception as exc:
        return {"ok": False, "failures": [f"Failed to execute code: {exc}"]}

    find_duplicates = namespace.get("find_duplicates")
    if find_duplicates is None:
        return {"ok": False, "failures": ["Function find_duplicates is not defined"]}

    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 {expected}, got {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="You are a code reviewer grading generated code against a rubric.",
    tools=[run_test_suite],
    max_iterations=5,
)

OpenAI

from deepagents import RubricMiddleware
from langchain.tools import tool

@tool
def run_test_suite(code: str) -> dict:
    """Run the find_duplicates test suite against Python source code."""
    namespace: dict = {"__builtins__": __builtins__}
    try:
        exec(code, namespace)
    except Exception as exc:
        return {"ok": False, "failures": [f"Failed to execute code: {exc}"]}

    find_duplicates = namespace.get("find_duplicates")
    if find_duplicates is None:
        return {"ok": False, "failures": ["Function find_duplicates is not defined"]}

    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 {expected}, got {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="You are a code reviewer grading generated code against a rubric.",
    tools=[run_test_suite],
    max_iterations=5,
)

Anthropic

from deepagents import RubricMiddleware
from langchain.tools import tool

@tool
def run_test_suite(code: str) -> dict:
    """Run the find_duplicates test suite against Python source code."""
    namespace: dict = {"__builtins__": __builtins__}
    try:
        exec(code, namespace)
    except Exception as exc:
        return {"ok": False, "failures": [f"Failed to execute code: {exc}"]}

    find_duplicates = namespace.get("find_duplicates")
    if find_duplicates is None:
        return {"ok": False, "failures": ["Function find_duplicates is not defined"]}

    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 {expected}, got {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="You are a code reviewer grading generated code against a rubric.",
    tools=[run_test_suite],
    max_iterations=5,
)

OpenRouter

from deepagents import RubricMiddleware
from langchain.tools import tool

@tool
def run_test_suite(code: str) -> dict:
    """Run the find_duplicates test suite against Python source code."""
    namespace: dict = {"__builtins__": __builtins__}
    try:
        exec(code, namespace)
    except Exception as exc:
        return {"ok": False, "failures": [f"Failed to execute code: {exc}"]}

    find_duplicates = namespace.get("find_duplicates")
    if find_duplicates is None:
        return {"ok": False, "failures": ["Function find_duplicates is not defined"]}

    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 {expected}, got {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="You are a code reviewer grading generated code against a rubric.",
    tools=[run_test_suite],
    max_iterations=5,
)

Fireworks

from deepagents import RubricMiddleware
from langchain.tools import tool

@tool
def run_test_suite(code: str) -> dict:
    """Run the find_duplicates test suite against Python source code."""
    namespace: dict = {"__builtins__": __builtins__}
    try:
        exec(code, namespace)
    except Exception as exc:
        return {"ok": False, "failures": [f"Failed to execute code: {exc}"]}

    find_duplicates = namespace.get("find_duplicates")
    if find_duplicates is None:
        return {"ok": False, "failures": ["Function find_duplicates is not defined"]}

    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 {expected}, got {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="You are a code reviewer grading generated code against a rubric.",
    tools=[run_test_suite],
    max_iterations=5,
)

Baseten

from deepagents import RubricMiddleware
from langchain.tools import tool

@tool
def run_test_suite(code: str) -> dict:
    """Run the find_duplicates test suite against Python source code."""
    namespace: dict = {"__builtins__": __builtins__}
    try:
        exec(code, namespace)
    except Exception as exc:
        return {"ok": False, "failures": [f"Failed to execute code: {exc}"]}

    find_duplicates = namespace.get("find_duplicates")
    if find_duplicates is None:
        return {"ok": False, "failures": ["Function find_duplicates is not defined"]}

    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 {expected}, got {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="You are a code reviewer grading generated code against a rubric.",
    tools=[run_test_suite],
    max_iterations=5,
)

Ollama

from deepagents import RubricMiddleware
from langchain.tools import tool

@tool
def run_test_suite(code: str) -> dict:
    """Run the find_duplicates test suite against Python source code."""
    namespace: dict = {"__builtins__": __builtins__}
    try:
        exec(code, namespace)
    except Exception as exc:
        return {"ok": False, "failures": [f"Failed to execute code: {exc}"]}

    find_duplicates = namespace.get("find_duplicates")
    if find_duplicates is None:
        return {"ok": False, "failures": ["Function find_duplicates is not defined"]}

    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 {expected}, got {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="You are a code reviewer grading generated code against a rubric.",
    tools=[run_test_suite],
    max_iterations=5,
)

将其传递给深度智能体

智能体的 system_prompt 告诉它如何完成工作，而准则告诉评审者如何评判工作。

Google

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=(
        "You are a careful Python engineer. Write correct, readable code. "
        "Follow the user's instructions exactly."
    ),
    middleware=[rubric_middleware],
    checkpointer=InMemorySaver(),
)

OpenAI

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

agent = create_deep_agent(
    model="openai:gpt-5.5",
    system_prompt=(
        "You are a careful Python engineer. Write correct, readable code. "
        "Follow the user's instructions exactly."
    ),
    middleware=[rubric_middleware],
    checkpointer=InMemorySaver(),
)

Anthropic

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

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    system_prompt=(
        "You are a careful Python engineer. Write correct, readable code. "
        "Follow the user's instructions exactly."
    ),
    middleware=[rubric_middleware],
    checkpointer=InMemorySaver(),
)

OpenRouter

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=(
        "You are a careful Python engineer. Write correct, readable code. "
        "Follow the user's instructions exactly."
    ),
    middleware=[rubric_middleware],
    checkpointer=InMemorySaver(),
)

Fireworks

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=(
        "You are a careful Python engineer. Write correct, readable code. "
        "Follow the user's instructions exactly."
    ),
    middleware=[rubric_middleware],
    checkpointer=InMemorySaver(),
)

Baseten

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=(
        "You are a careful Python engineer. Write correct, readable code. "
        "Follow the user's instructions exactly."
    ),
    middleware=[rubric_middleware],
    checkpointer=InMemorySaver(),
)

Ollama

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=(
        "You are a careful Python engineer. Write correct, readable code. "
        "Follow the user's instructions exactly."
    ),
    middleware=[rubric_middleware],
    checkpointer=InMemorySaver(),
)

使用人类消息和准则调用

在调用时，在 messages 中提供用户请求，在 rubric 中提供评审者必须标记为满足的换行分隔的检查清单。当输入状态中没有提供 rubric 时，中间件不会运行。

from langchain.messages import HumanMessage

result = agent.invoke(
    {
        "messages": [
            HumanMessage(
                content=(
                    "Write a Python function `find_duplicates(lst)` that returns a list of "
                    "all elements that appear more than once in the input list, in the order "
                    "they first appear."
                )
            )
        ],
        "rubric": (
            "- All tests pass in run_test_suite\n"
            "- The function is named `find_duplicates` and accepts a single list argument\n"
        ),
    },
    config={"configurable": {"thread_id": "code-generation-session"}},
)
print(result["messages"][-1].text)

智能体产生输出后，评审者接管并针对每个标准检查输出：例如，当输入包含不可哈希类型时，test_unhashable 以 TypeError 失败。如果有任何问题，评审者会提供此反馈，然后智能体修订其实现并将其返回给评审者。