第五章：综合案例

本章把 wrapmodelcall、wraptoolcall 和 createdeepagent 放到同一条真实流程里。wrapmodelcall 观察模型层调用，并对临时上游错误做最小重试；wraptoolcall 观察并改写工具结果；Deep Agent 负责真实调用模型并决定工具调用。

全部笔记查看源文档

本章把 wrap_model_call、wrap_tool_call 和 create_deep_agent 放到同一条真实流程里。wrap_model_call 观察模型层调用，并对临时上游错误做最小重试；wrap_tool_call 观察并改写工具结果；Deep Agent 负责真实调用模型并决定工具调用。

最小代码

代码在 deepagent_src/middleware_teach/05_comprehensive_case.py。

@wrap_model_call
def record_model_call(request, handler):
    model_events.append(f"model_call:{len(request.state['messages'])}")
    for attempt in range(3):
        try:
            return handler(request)
        except Exception:
            if attempt == 2:
                raise

@wrap_tool_call
def audit_invoice_tool(request, handler):
    result = handler(request)
    return ToolMessage(
        content=f"AUDITED_TOOL:{result.content}",
        tool_call_id=request.tool_call["id"],
        name=request.tool_call["name"],
    )

模型层 middleware 不碰工具结果，只记录每次模型调用，并在模型服务临时不可用时重试同一个请求。工具层 middleware 不决定模型是否调用工具，只包住已经发生的工具执行。

运行命令

uv run python -m deepagent_src.middleware_teach.05_comprehensive_case

预期现象

输出类似：

model_events: ['model_call:1', 'model_call:3']
tool_events: ["before:lookup_invoice:{'invoice_id': 'mw-5001'}", 'after:invoice mw-5001 total is 88 USD']
tool_message: AUDITED_TOOL:invoice mw-5001 total is 88 USD
final: MIDDLEWARE_CASE_OK: AUDITED_TOOL:invoice mw-5001 total is 88 USD
middleware comprehensive real call ok

这里有两次模型调用：第一次模型决定调用工具，工具结果进入历史后，第二次模型生成最终回答。wrap_tool_call 只在工具节点执行时触发一次。

常见误区

不要把 middleware 当成万能业务框架。模型选择、工具审计、错误恢复、HITL 都能放 middleware，但每层只干自己的事：模型层管模型请求，工具层管工具执行，权限边界还要落在后端服务和工具实现里。艹，把所有逻辑塞进一个 middleware，就是换个地方写屎山。

五章串起来

before_model / after_model：看模型调用前后状态。
wrap_model_call：包住模型调用，可重试、fallback、短路或替换模型。
wrap_tool_call：包住工具执行，可审计、脱敏、错误转译。
Deep Agents 默认栈：理解你的 middleware 插在 Deep Agents harness 的哪个位置。
综合案例：真实 Deep Agent 中同时使用模型层和工具层 middleware。

本章触发真实 gpt-5.5 Deep Agent 调用，会产生最小模型调用成本。没有使用 mock、fake model 或内部函数冒充 Agent 运行。

本地验证时 gpt-5.5 曾连续返回上游 503 Service temporarily unavailable，所以示例在 wrap_model_call 里保留 3 次最小重试。生产环境不要裸捕全部异常，应按 provider 错误类型、重试预算和审计策略收窄。

相关资源

查看示例代码：deepagent_src/memory_teach/05_comprehensive_case.py

from __future__ import annotations

import os

from deepagents import FilesystemPermission, create_deep_agent
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
from deepagents.backends.utils import create_file_data
from deepagents.middleware.filesystem import _check_fs_permission
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

from deepagent_src.agent_output import stream_debug_trace
from deepagent_src.llms import get_gpt_model


USER_NAMESPACE = ("memory-teach-user",)
ORG_NAMESPACE = ("memory-teach-org",)
MEMORY_SOURCE = "/memories/AGENTS.md"
POLICY_SOURCE = "/policies/AGENTS.md"
ROUTED_STORE_KEY = "/AGENTS.md"


def main() -> None:
    os.environ["LANGSMITH_TRACING"] = "false"

    store = InMemoryStore()
    store.put(
        USER_NAMESPACE,
        ROUTED_STORE_KEY,
        create_file_data(
            """# User Memory

## Response style

- 用户喜欢中文解释。
- 用户希望 DeepAgent 教学尽量给可运行代码。
"""
        ),
    )
    store.put(
        ORG_NAMESPACE,
        ROUTED_STORE_KEY,
        create_file_data(
            """# Organization Policy

## Memory safety

- 不要把 API key、token、密码写进 memory。
- 共享 policy memory 只能读，不能让 Agent 修改。
"""
        ),
    )

    backend = CompositeBackend(
        default=StateBackend(),
        routes={
            "/memories/": StoreBackend(
                store=store,
                namespace=lambda _rt: USER_NAMESPACE,
            ),
            "/policies/": StoreBackend(
                store=store,
                namespace=lambda _rt: ORG_NAMESPACE,
            ),
        },
    )
    permissions = [
        FilesystemPermission(
            operations=["write"],
            paths=["/policies/**"],
            mode="deny",
        )
    ]

    assert backend.download_files([MEMORY_SOURCE])[0].error is None
    assert backend.download_files([POLICY_SOURCE])[0].error is None
    assert _check_fs_permission(permissions, "write", POLICY_SOURCE) == "deny"
    assert _check_fs_permission(permissions, "read", POLICY_SOURCE) == "allow"

    graph = create_deep_agent(
        model=get_gpt_model(),
        backend=backend,
        memory=[MEMORY_SOURCE, POLICY_SOURCE],
        permissions=permissions,
        store=store,
        checkpointer=InMemorySaver(),
    )

    messages = [
        HumanMessage(
            content=(
                "请根据已加载的长期 memory，说明我学习 DeepAgent memory 时应该注意什么。"
                "要求：用中文，提到用户偏好和组织安全规则。不要写文件。"
            )
        )
    ]
    stream_debug_trace(
        graph,
        {"messages": messages},
        {"configurable": {"thread_id": "memory-comprehensive-case"}},
    )


if __name__ == "__main__":
    main()

查看示例代码：deepagent_src/middleware_teach/05_comprehensive_case.py

from __future__ import annotations

from collections.abc import Callable
from time import sleep

from deepagents import create_deep_agent
from langchain.agents.middleware import (
    ModelRequest,
    ModelResponse,
    ToolCallRequest,
    wrap_model_call,
    wrap_tool_call,
)
from langchain.messages import HumanMessage, ToolMessage
from langchain_core.tools import tool

from deepagent_src.llms import get_gpt_model

model_events: list[str] = []
tool_events: list[str] = []


@tool
def lookup_invoice(invoice_id: str) -> str:
    """Look up an invoice by invoice id."""
    if invoice_id != "mw-5001":
        return f"invoice {invoice_id} not found"
    return "invoice mw-5001 total is 88 USD"


@wrap_model_call(name="record_model_call")
def record_model_call(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    model_events.append(f"model_call:{len(request.state['messages'])}")
    for attempt in range(3):
        try:
            return handler(request)
        except Exception as exc:
            model_events.append(f"retry:{attempt + 1}:{type(exc).__name__}")
            if attempt == 2:
                raise
            sleep(1)
    raise AssertionError("unreachable")


@wrap_tool_call(name="audit_invoice_tool")
def audit_invoice_tool(
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolMessage],
) -> ToolMessage:
    tool_events.append(f"before:{request.tool_call['name']}:{request.tool_call['args']}")
    result = handler(request)
    tool_events.append(f"after:{result.content}")
    return ToolMessage(
        content=f"AUDITED_TOOL:{result.content}",
        tool_call_id=request.tool_call["id"],
        name=request.tool_call["name"],
    )


def main() -> None:
    model_events.clear()
    tool_events.clear()

    agent = create_deep_agent(
        model=get_gpt_model(disable_tool_streaming=True),
        tools=[lookup_invoice],
        middleware=[record_model_call, audit_invoice_tool],
        subagents=[],
        system_prompt=(
            "你是 middleware 综合案例助手。"
            "必须调用 lookup_invoice，invoice_id 必须是 mw-5001。"
            "看到工具结果后，只回复 `MIDDLEWARE_CASE_OK: <工具结果>`。"
        ),
    )
    state = agent.invoke({"messages": [HumanMessage(content="查询教学发票。")]})
    messages = state["messages"]
    tool_messages = [message for message in messages if message.type == "tool"]
    final_text = messages[-1].text

    print("model_events:", model_events)
    print("tool_events:", tool_events)
    print("tool_message:", tool_messages[-1].text)
    print("final:", final_text)

    assert len(model_events) >= 2, model_events
    assert tool_events == [
        "before:lookup_invoice:{'invoice_id': 'mw-5001'}",
        "after:invoice mw-5001 total is 88 USD",
    ], tool_events
    assert tool_messages[-1].text == "AUDITED_TOOL:invoice mw-5001 total is 88 USD"
    assert "MIDDLEWARE_CASE_OK" in final_text, final_text
    assert "mw-5001" in final_text and "88 USD" in final_text, final_text
    print("middleware comprehensive real call ok")


if __name__ == "__main__":
    main()

查看示例代码：deepagent_src/profiles_teach/05_comprehensive_case.py

from __future__ import annotations

from deepagents import (
    GeneralPurposeSubagentProfile,
    HarnessProfile,
    ProviderProfile,
    create_deep_agent,
    register_harness_profile,
    register_provider_profile,
)
from deepagents.profiles.provider.provider_profiles import apply_provider_profile
from langchain.tools import tool

from _model import MODEL_PROFILE_KEY, get_real_model
from deepagent_src.agent_output import invoke_and_pretty_print


@tool
def profile_summary_tool() -> str:
    """Return the profile summary marker."""
    return "profiles connect harness behavior and provider construction"


def main() -> None:
    register_provider_profile(
        "profiledemo",
        ProviderProfile(init_kwargs={"temperature": 0}),
    )
    register_provider_profile(
        "profiledemo:lesson",
        ProviderProfile(init_kwargs={"timeout": 8}),
    )
    provider_kwargs = apply_provider_profile(
        "profiledemo:lesson",
        {"timeout": 3},
        run_pre_init=False,
    )

    register_harness_profile(
        MODEL_PROFILE_KEY,
        HarnessProfile(
            system_prompt_suffix="Keep the final answer to one sentence.",
            tool_description_overrides={
                "profile_summary_tool": "Use this for the profiles lesson summary."
            },
            excluded_tools=frozenset({"execute"}),
            general_purpose_subagent=GeneralPurposeSubagentProfile(enabled=False),
        ),
    )

    agent = create_deep_agent(model=get_real_model(), tools=[profile_summary_tool])
    result = invoke_and_pretty_print(
        agent,
        {
            "messages": [
                {
                    "role": "user",
                    "content": (
                        "必须调用 profile_summary_tool，"
                        "然后用一句中文总结工具返回值。"
                    ),
                }
            ]
        },
    )

    tool_outputs = [
        getattr(message, "content", "")
        for message in result["messages"]
        if message.__class__.__name__ == "ToolMessage"
    ]

    assert provider_kwargs["temperature"] == 0
    assert provider_kwargs["timeout"] == 3
    assert any("profiles connect harness behavior" in output for output in tool_outputs)
    print("profiles comprehensive real agent ok")


if __name__ == "__main__":
    main()