05 综合案例：真实 Agent 读取 Memory

相关资源

• 查看示例代码：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()