---
title: "deepagents学习打卡day07"
source: https://www.zbhgis.com/note/note-20260926-1
author: zbhgis 浩瀚地学
created: 2026-09-26
tags: [Agent, Agent]
---

> 本文来自浩瀚地学，规范页面（含代码高亮与图片灯箱）：https://www.zbhgis.com/note/note-20260926-1
# deepagents学习打卡day07

个人主页：https://github.com/zbhgis/

# 打卡记录

## deepagents输出

### 跨会话记忆

```bash
[sanitizer] 已安装 openai JSON 兜底清理
正在初始化 DeepAgent（首次调用模型可能稍慢）...
初始化完成。

============================================================
DeepAgents 真实操作演示
============================================================
  1. 跨轮对话长期记忆验证
  2. 真实操作 + 人工审批（读密钥 / 写 / 删）
  q. 退出
请选择 > 1

============================================================
【跨轮对话长期记忆验证】
============================================================
机制：每轮对话使用全新 thread_id；Agent 只能通过
     持久化的 /memories/AGENTS.md 感知历史，验证记忆是否真正跨轮。


可选操作：
  [1] 让 Agent 记住一条新信息
  [2] 用全新 thread_id 询问 Agent（验证跨轮记忆）
  [3] 直接查看 /memories/AGENTS.md 的真实内容
  [q] 返回主菜单
请选择 > 1
请输入要 Agent 记住的内容: 我是zbhgis，以后每次回复我都称呼我zbhgis

[thread_id] 9aabb5be-8553-4589-a0ce-5bae647d76ba  (本轮会话)
[Agent]

好的 zbhgis，已经记住了！以后每次回复我都会称呼你为 **zbhgis**。

我已经把这条偏好写入了 `/memories/AGENTS.md`，后续对话也会一直生效。有什么需要帮忙的尽管说～

可选操作：
  [1] 让 Agent 记住一条新信息
  [2] 用全新 thread_id 询问 Agent（验证跨轮记忆）
  [3] 直接查看 /memories/AGENTS.md 的真实内容
  [q] 返回主菜单
请选择 > 2
请输入要询问的问题: 你是谁

[thread_id] 55483701-cf65-423d-95fe-d613e5b1c46e  ← 全新会话，无历史消息
[Agent]

zbhgis，你好！

我是**商量**（SenseNova），由商汤科技（SenseTime）自主研发的日日新融合模态大模型。

我具备多模态理解能力，可以处理文本、图像等多种信息形式，致力于成为一个**有用无害**的 AI 助手。

在当前这个环境里，我还额外配备了一套**带人工审批的本地运维工具**，可以帮你：

- 📂 **查看工作区文件**（`list_workspace` / `read_local_file`）—— 只读，直接执行
- 🔑 **读取本地密钥**（`get_secret`）—— 高敏感，需人工审批
- ✏️ **写入 / 覆盖文件**（`write_local_file`）—— 敏感，需人工审批
- 🗑️ **删除文件**（`delete_local_file`）—— 高风险，需人工审批

需要我帮你做点什么？

可选操作：
  [1] 让 Agent 记住一条新信息
  [2] 用全新 thread_id 询问 Agent（验证跨轮记忆）
  [3] 直接查看 /memories/AGENTS.md 的真实内容
  [q] 返回主菜单
请选择 >
```

### 权限审批：拒绝

```bash
============================================================
【真实操作 + 人工审批】
============================================================
工作区：/home/zbhgis/.deepagents_demo/workspace

可尝试的请求：
  1) 列出工作区里的文件，然后删除 old_backup.csv
  2) 读取 important.txt 的内容
  3) 获取 db_password 这个密钥的值
  4) 删除 old_backup.csv，然后给 notes.md 追加一行 '# 已归档'

请输入你的请求（或直接按回车用 1 号示例）: 删除 old_backup.csv，然后给 notes.md 追加一行 '# 已归档'

[thread_id] cb0378a9-743a-4233-b809-e54f3fc28814


[中断] 2 个待审批动作

--- 待审批 [1/2] ---
  工具 : delete_local_file
  参数 : {"path": "old_backup.csv"}
  允许 : ['approve', 'edit', 'reject']
  提示 : 将真实删除 old_backup.csv，可从工作区目录验证效果
决策 ['approve', 'edit', 'reject'] > reject
  拒绝理由（回车用默认）:
  → 决策：reject

--- 待审批 [2/2] ---
  工具 : write_local_file
  参数 : {"path": "notes.md", "content": "# 项目笔记\n- 周五上线\n- 备份数据库\n# 已归档\n"}
  允许 : ['approve', 'edit', 'reject']
  提示 : 将真实写入 notes.md，可用 edit 修改内容
决策 ['approve', 'edit', 'reject'] > reject
  拒绝理由（回车用默认）:
  → 决策：reject

[恢复执行] 提交 2 条决策 ...

[最终回复]


zbhgis，两个操作都被你拒绝了：

1. `delete_local_file(old_backup.csv)` —— 已拒绝
2. `write_local_file(notes.md, ...)` —— 已拒绝

我没有做任何改动，工作区保持原样（`old_backup.csv`、`notes.md`、`important.txt` 都还在）。

想确认一下你的意图，方便我按你的意思来：

- 是**两个都不做**，还是**只做其中一个**？
- 如果只想追加 `# 已归档`，你希望：
  - 追加在文件**末尾**（我刚才的做法），还是
  - 作为**新的一行**紧跟在最后一行 `- 备份数据库` 之后（不加空行）？
- 如果只想删 `old_backup.csv`，是否确认删除？

告诉我怎么调整，我再发起调用。

[验证] 工作区当前状态：
  [FILE] important.txt  (51 bytes)
  [FILE] notes.md  (48 bytes)
  [FILE] old_backup.csv  (37 bytes)

```

### 权限审批：允许

```bash
============================================================
DeepAgents 真实操作演示
============================================================
  1. 跨轮对话长期记忆验证
  2. 真实操作 + 人工审批（读密钥 / 写 / 删）
  q. 退出
请选择 > 2

============================================================
【真实操作 + 人工审批】
============================================================
工作区：/home/zbhgis/.deepagents_demo/workspace

可尝试的请求：
  1) 列出工作区里的文件，然后删除 old_backup.csv
  2) 读取 important.txt 的内容
  3) 获取 db_password 这个密钥的值
  4) 删除 old_backup.csv，然后给 notes.md 追加一行 '# 已归档'

请输入你的请求（或直接按回车用 1 号示例）: 获取 db_password 这个密钥的值

[thread_id] 198a747f-9c8d-48e6-9bf9-1f984d46dbcb


[中断] 1 个待审批动作

--- 待审批 [1/1] ---
  工具 : get_secret
  参数 : {"key": "db_password"}
  允许 : ['approve', 'reject']
  提示 : 将读取密钥 db_password（脱敏输出）
决策 ['approve', 'reject'] > approve
  → 决策：approve

[恢复执行] 提交 1 条决策 ...

[最终回复]


zbhgis，密钥 `db_password` 已读取，但出于安全考虑，我不会在对话中回显完整明文。以下是脱敏信息：

- **脱敏值**：`S3cr***********2024`
- **长度**：19 字符
- **来源**：`/home/zbhgis/.deepagents_demo/secrets.json`

如果你确实需要完整明文（例如用于本地调试），建议直接在终端里查看源文件，或者告诉我具体用途，我可以帮你把它注入到某个脚本/配置里，避免明文出现在聊天记录中。

[验证] 工作区当前状态：
  [FILE] important.txt  (51 bytes)
  [FILE] notes.md  (48 bytes)
  [FILE] old_backup.csv  (37 bytes)

============================================================
DeepAgents 真实操作演示
============================================================
  1. 跨轮对话长期记忆验证
  2. 真实操作 + 人工审批（读密钥 / 写 / 删）
  q. 退出
```

### 权限审批：修改

```bash
============================================================
DeepAgents 真实操作演示
============================================================
  1. 跨轮对话长期记忆验证
  2. 真实操作 + 人工审批（读密钥 / 写 / 删）
  q. 退出
请选择 > 2

============================================================
【真实操作 + 人工审批】
============================================================
工作区：/home/zbhgis/.deepagents_demo/workspace

可尝试的请求：
  1) 列出工作区里的文件，然后删除 old_backup.csv
  2) 读取 important.txt 的内容
  3) 获取 db_password 这个密钥的值
  4) 删除 old_backup.csv，然后给 notes.md 追加一行 '# 已归档'

请输入你的请求（或直接按回车用 1 号示例）: 给 notes.md 追加一行 # 测试

[thread_id] 3fbb9073-6985-49d3-8a15-6aae93a1f166


[中断] 1 个待审批动作

--- 待审批 [1/1] ---
  工具 : write_local_file
  参数 : {"path": "notes.md", "content": "归档完成20260926\n# 已归档\n# # 测试\n"}
  允许 : ['approve', 'edit', 'reject']
  提示 : 将真实写入 notes.md，可用 edit 修改内容
决策 ['approve', 'edit', 'reject'] > edit
  输入要覆盖/新增的参数（JSON 对象），例如：
    {"path": "notes.md", "content": "..."}
  JSON > {"path": "notes.md", "content": "..."}^[[D^[[D
  解析失败：Extra data: line 1 column 39 (char 38)
决策 ['approve', 'edit', 'reject'] >
  无效，请输入 ['approve', 'edit', 'reject'] 之一。
决策 ['approve', 'edit', 'reject'] > edit
  输入要覆盖/新增的参数（JSON 对象），例如：
    {"path": "notes.md", "content": "..."}
  JSON > {"path": "notes.md", "content": "测试成功"}
  → 决策：edit

[恢复执行] 提交 1 条决策 ...

```

## langsmith记录

### 跨会话记忆

![image-20260926190608145](https://cdn.jsdelivr.net/gh/zbhgis/BlogImg2@main/blog/202609261906944.png)

![image-20260926190637706](https://cdn.jsdelivr.net/gh/zbhgis/BlogImg2@main/blog/202609261906782.png)

### 权限审批：拒绝

![image-20260926191259175](https://cdn.jsdelivr.net/gh/zbhgis/BlogImg2@main/blog/202609261913654.png)

### 权限审批：允许

![image-20260926191114885](https://cdn.jsdelivr.net/gh/zbhgis/BlogImg2@main/blog/202609261911264.png)

### 权限审批：修改

![image-20260926192145944](https://cdn.jsdelivr.net/gh/zbhgis/BlogImg2@main/blog/202609261921543.png)

## 完整代码

config.py

```Python
"""模型与全局配置。"""
import os
from langchain_openai import ChatOpenAI


def get_model() -> ChatOpenAI:
    return ChatOpenAI(
        model=os.environ.get("LLM_MODEL_NAME"),
        api_key=os.environ.get("LLM_API_KEY"),
        base_url=os.environ.get("LLM_BASE_URL"),
        temperature=0,
    )

```

tools.py

```Python
"""真实工具集。低敏感工具自动执行，高敏感工具走 HITL。"""
from langchain.tools import tool

from workspace import (
    WORKSPACE_DIR,
    human_size,
    list_secret_keys,
    read_secret,
    resolve_workspace_path,
)


# ---------- 只读 / 低风险 ----------

@tool
def list_workspace(subdir: str = "") -> str:
    """列出工作区（或子目录）内的真实文件。只读操作，自动执行。"""
    try:
        target = resolve_workspace_path(subdir) if subdir else WORKSPACE_DIR.resolve()
    except ValueError as e:
        return str(e)

    if not target.exists():
        return f"目录不存在：{target}"
    if not target.is_dir():
        return f"不是目录：{target}"

    lines = []
    for p in sorted(target.iterdir()):
        if p.is_dir():
            lines.append(f"  [DIR ] {p.name}/")
        else:
            lines.append(f"  [FILE] {p.name}  ({human_size(p.stat().st_size)})")
    body = "\n".join(lines) if lines else "  (空目录)"
    return f"工作区 {target}：\n{body}"


@tool
def read_local_file(path: str) -> str:
    """读取工作区内真实文件内容。只读操作，自动执行。"""
    try:
        p = resolve_workspace_path(path)
    except ValueError as e:
        return str(e)

    if not p.exists() or not p.is_file():
        return f"文件不存在：{p}"
    try:
        content = p.read_text(encoding="utf-8")
    except UnicodeDecodeError:
        return f"非 UTF-8 文本，拒绝读取：{p}"

    truncated = ""
    if len(content) > 4000:
        content = content[:4000]
        truncated = "\n... (内容已截断至 4000 字符)"
    return f"=== {p} ===\n{content}{truncated}"


# ---------- 高敏感：真实读取本地密钥 ----------

@tool
def get_secret(key: str) -> str:
    """从本地密钥库 ~/.deepagents_demo/secrets.json 读取密钥。

    高敏感操作，需要人工审批。
    """
    value = read_secret(key)
    if value is None:
        available = ", ".join(list_secret_keys()) or "(密钥库为空)"
        return f"未找到密钥 '{key}'。可用密钥名：{available}"

    # 只输出脱敏后的形式，避免完整密钥泄露到对话中
    if len(value) <= 8:
        masked = "*" * len(value)
    else:
        masked = f"{value[:4]}{'*' * (len(value) - 8)}{value[-4:]}"
    return (
        f"已读取密钥 '{key}'：\n"
        f"  脱敏值 = {masked}\n"
        f"  长度   = {len(value)} 字符\n"
        f"  来源   = {WORKSPACE_DIR.parent / 'secrets.json'}"
    )


# ---------- 高敏感：真实写 / 删 ----------

@tool
def write_local_file(path: str, content: str) -> str:
    """在工作区内写入或覆盖真实文件。敏感操作，需要人工审批。"""
    try:
        p = resolve_workspace_path(path)
    except ValueError as e:
        return str(e)

    if p.is_dir():
        return f"拒绝写入：{p} 是目录"
    p.parent.mkdir(parents=True, exist_ok=True)
    existed = p.exists()
    p.write_text(content, encoding="utf-8")
    verb = "覆盖" if existed else "新建"
    return f"已{verb}文件：{p}\n写入 {len(content)} 字符，落盘大小 {human_size(p.stat().st_size)}"


@tool
def delete_local_file(path: str) -> str:
    """删除工作区内真实文件。高风险操作，需要人工审批。"""
    try:
        p = resolve_workspace_path(path)
    except ValueError as e:
        return str(e)

    if not p.exists():
        return f"文件不存在：{p}"
    if p.is_dir():
        return f"拒绝删除目录（只允许删除文件）：{p}"

    size = p.stat().st_size
    p.unlink()  # 真实删除
    return f"已真实删除文件：{p}  (释放 {human_size(size)})"
```

workspace.py

```Python
"""真实本地工作区与密钥库。

- ~/.deepagents_demo/secrets.json   ← 本地密钥库（真实文件，权限 0o600）
- ~/.deepagents_demo/workspace/     ← 文件操作沙箱（真实目录）
"""
import json
import os
from pathlib import Path


ROOT = Path.home() / ".deepagents_demo"
SECRETS_FILE = ROOT / "secrets.json"
WORKSPACE_DIR = ROOT / "workspace"


def init_environment(force: bool = False) -> None:
    """创建目录、写入示例密钥和工作区文件。幂等。"""
    ROOT.mkdir(mode=0o700, exist_ok=True)
    WORKSPACE_DIR.mkdir(mode=0o700, exist_ok=True)

    if force or not SECRETS_FILE.exists():
        demo_secrets = {
            "db_password": "S3cr3t-DB-P@ss-2024",
            "openai_api_key": "sk-demo-openai-abcdef1234567890",
            "aws_secret_key": "aws-demo-key-xyz-789",
            "jwt_signing_key": "jwt-sign-thisisasecret",
        }
        SECRETS_FILE.write_text(
            json.dumps(demo_secrets, indent=2, ensure_ascii=False),
            encoding="utf-8",
        )
        try:
            os.chmod(SECRETS_FILE, 0o600)
        except (OSError, NotImplementedError):
            pass  # Windows 上可能不支持

    # 工作区里的示例文件（用于删除演示）
    samples = {
        "old_backup.csv": "id,name,amount\n1,alice,100\n2,bob,200\n",
        "important.txt": "⚠️  这个文件很重要，请不要删除。\n",
        "notes.md": "# 项目笔记\n- 周五上线\n- 备份数据库\n",
    }
    for name, content in samples.items():
        p = WORKSPACE_DIR / name
        if not p.exists():
            p.write_text(content, encoding="utf-8")


# ---------- 密钥库真实读写 ----------

def read_secret(key: str) -> str | None:
    if not SECRETS_FILE.exists():
        return None
    data = json.loads(SECRETS_FILE.read_text(encoding="utf-8"))
    return data.get(key)


def list_secret_keys() -> list[str]:
    if not SECRETS_FILE.exists():
        return []
    return list(json.loads(SECRETS_FILE.read_text(encoding="utf-8")).keys())


# ---------- 工作区路径解析（沙箱） ----------

def resolve_workspace_path(rel_or_abs: str) -> Path:
    """把用户提供的路径解析到工作区内，越界抛 ValueError。"""
    raw = (rel_or_abs or "").strip()
    if not raw:
        raise ValueError("路径不能为空")
    p = Path(raw)
    if not p.is_absolute():
        p = WORKSPACE_DIR / p
    p = p.resolve()
    try:
        p.relative_to(WORKSPACE_DIR.resolve())
    except ValueError:
        raise ValueError(
            f"路径越界：只允许操作工作区 {WORKSPACE_DIR} 内的文件（收到 {p}）"
        )
    return p


def human_size(n: int) -> str:
    for unit in ("B", "KB", "MB", "GB"):
        if n < 1024:
            return f"{n:.0f}{unit}"
        n /= 1024
    return f"{n:.1f}TB"
```

demo_memory.py

```Python
"""演示一：跨轮对话长期记忆验证（真实 store 内容可查）。"""
from uuid import uuid4

from agent_setup import MEMORY_NAMESPACE, MEMORY_PATH


def run_memory_demo(agent, store):
    print("\n" + "=" * 60)
    print("【跨轮对话长期记忆验证】")
    print("=" * 60)
    print("机制：每轮对话使用全新 thread_id；Agent 只能通过")
    print(f"     持久化的 {MEMORY_PATH} 感知历史，验证记忆是否真正跨轮。")
    print()

    while True:
        print("\n可选操作：")
        print("  [1] 让 Agent 记住一条新信息")
        print("  [2] 用全新 thread_id 询问 Agent（验证跨轮记忆）")
        print("  [3] 直接查看 /memories/AGENTS.md 的真实内容")
        print("  [q] 返回主菜单")
        choice = input("请选择 > ").strip().lower()

        if choice == "q":
            return
        elif choice == "1":
            info = input("请输入要 Agent 记住的内容: ").strip()
            if not info:
                print("内容为空，已取消。")
                continue
            _remember(agent, info)
        elif choice == "2":
            q = input("请输入要询问的问题: ").strip()
            if not q:
                print("问题为空，已取消。")
                continue
            _ask(agent, q)
        elif choice == "3":
            _show_memory(store)
        else:
            print("无效选项。")


def _remember(agent, info: str):
    thread_id = str(uuid4())
    cfg = {"configurable": {"thread_id": thread_id}}
    result = agent.invoke(
        {"messages": [{"role": "user", "content": f"请记住：{info}"}]},
        config=cfg,
    )
    print(f"\n[thread_id] {thread_id}  (本轮会话)")
    print(f"[Agent] {result['messages'][-1].content}")


def _ask(agent, question: str):
    thread_id = str(uuid4())
    cfg = {"configurable": {"thread_id": thread_id}}
    result = agent.invoke(
        {"messages": [{"role": "user", "content": question}]},
        config=cfg,
    )
    print(f"\n[thread_id] {thread_id}  ← 全新会话，无历史消息")
    print(f"[Agent] {result['messages'][-1].content}")


def _show_memory(store):
    item = store.get(MEMORY_NAMESPACE, MEMORY_PATH)
    print(f"\n[持久化存储中的 {MEMORY_PATH}]")
    if item is None:
        print("  (尚未写入)")
        return
    value = item.value
    if isinstance(value, dict) and "content" in value:
        content = value["content"]
        if isinstance(content, list):
            for line in content:
                print(f"  {line}")
        else:
            print(f"  {content}")
    else:
        print(f"  {value}")
```

demo_hitl.py

```Python
"""演示二：真实操作的审批流（approve / edit / reject）+ 效果验证。"""
import json
from uuid import uuid4

from langgraph.types import Command

from workspace import WORKSPACE_DIR, resolve_workspace_path


SUGGESTED = [
    "列出工作区里的文件，然后删除 old_backup.csv",
    "读取 important.txt 的内容",
    "获取 db_password 这个密钥的值",
    "删除 old_backup.csv，然后给 notes.md 追加一行 '# 已归档'",
]


def run_hitl_demo(agent):
    print("\n" + "=" * 60)
    print("【真实操作 + 人工审批】")
    print("=" * 60)
    print(f"工作区：{WORKSPACE_DIR}")
    print("\n可尝试的请求：")
    for i, s in enumerate(SUGGESTED, 1):
        print(f"  {i}) {s}")
    print()

    user_input = input("请输入你的请求（或直接按回车用 1 号示例）: ").strip()
    if not user_input:
        user_input = SUGGESTED[0]
        print(f"→ 使用默认请求：{user_input}")

    thread_id = str(uuid4())
    cfg = {"configurable": {"thread_id": thread_id}}
    print(f"\n[thread_id] {thread_id}\n")

    result = agent.invoke(
        {"messages": [{"role": "user", "content": user_input}]},
        config=cfg,
    )

    while True:
        interrupt = result.get("__interrupt__")
        if not interrupt:
            break

        action_requests = interrupt[0].value["action_requests"]
        review_configs = interrupt[0].value["review_configs"]

        print(f"\n[中断] {len(action_requests)} 个待审批动作")
        decisions = []
        for i, action in enumerate(action_requests):
            allowed = review_configs[i].get(
                "allowed_decisions", ["approve", "edit", "reject"]
            )
            print(f"\n--- 待审批 [{i+1}/{len(action_requests)}] ---")
            print(f"  工具 : {action['name']}")
            print(f"  参数 : {json.dumps(action['args'], ensure_ascii=False)}")
            print(f"  允许 : {allowed}")
            print(f"  提示 : {_hint(action)}")
            decision = _prompt_decision(action, allowed)
            decisions.append(decision)
            print(f"  → 决策：{decision['type']}")

        print(f"\n[恢复执行] 提交 {len(decisions)} 条决策 ...")
        result = agent.invoke(
            Command(resume={"decisions": decisions}),
            config=cfg,
        )

    print(f"\n[最终回复]\n{result['messages'][-1].content}")
    print("\n[验证] 工作区当前状态：")
    _print_workspace()


def _hint(action) -> str:
    name = action["name"]
    if name == "delete_local_file":
        return f"将真实删除 {action['args'].get('path')}，可从工作区目录验证效果"
    if name == "get_secret":
        return f"将读取密钥 {action['args'].get('key')}（脱敏输出）"
    if name == "write_local_file":
        return f"将真实写入 {action['args'].get('path')}，可用 edit 修改内容"
    return ""


def _prompt_decision(action, allowed):
    while True:
        raw = input(f"决策 {allowed} > ").strip().lower()
        if raw not in allowed:
            print(f"  无效，请输入 {allowed} 之一。")
            continue

        if raw == "approve":
            return {"type": "approve"}

        if raw == "reject":
            reason = input("  拒绝理由（回车用默认）: ").strip()
            return {
                "type": "reject",
                "message": reason or "用户拒绝了本次操作",
            }

        if raw == "edit":
            print("  输入要覆盖/新增的参数（JSON 对象），例如：")
            print('    {"path": "notes.md", "content": "..."}')
            js = input("  JSON > ").strip()
            if not js:
                print("  为空，请重新选择。")
                continue
            try:
                new_args = json.loads(js)
                if not isinstance(new_args, dict):
                    raise ValueError("必须是 JSON 对象")
            except (json.JSONDecodeError, ValueError) as e:
                print(f"  解析失败：{e}")
                continue
            merged = {**action["args"], **new_args}
            return {
                "type": "edit",
                "edited_action": {"name": action["name"], "args": merged},
            }

        if raw == "respond":
            text = input("  respond 内容: ").strip()
            return {"type": "respond", "message": text or "已记录你的反馈"}


def _print_workspace():
    for p in sorted(WORKSPACE_DIR.iterdir()):
        mark = "DIR " if p.is_dir() else "FILE"
        size = "" if p.is_dir() else f"  ({p.stat().st_size} bytes)"
        print(f"  [{mark}] {p.name}{size}")
```

agent_setup.py

```Python
"""Agent 工厂：长期记忆 + HITL 审批（适配 deepagents 0.7）。

相对 0.6 的变更（0.7 移除了 backend factory）：
  1. backend 参数从 "工厂函数" 改为 "已初始化的实例"
  2. StateBackend() 不再传 runtime 参数
  3. StoreBackend() 不再传 runtime 参数（namespace 工厂仍保留）
"""
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
from deepagents.backends.utils import create_file_data

from config import get_model
from tools import (
    delete_local_file,
    get_secret,
    list_workspace,
    read_local_file,
    write_local_file,
)
from workspace import init_environment


# 固定命名空间 → 所有 thread_id 共享同一份长期记忆
MEMORY_NAMESPACE = ("my-agent",)
MEMORY_PATH = "/memories/AGENTS.md"

SYSTEM_PROMPT = (
    "你是一个带人工审批的本地运维助手。回答一律使用中文。\n"
    "\n"
    "【你的真实能力】\n"
    "  · list_workspace / read_local_file —— 探查工作区文件，只读，直接调用。\n"
    "  · get_secret —— 读取本地密钥库中的密钥（高敏感）。\n"
    "  · write_local_file —— 写入 / 覆盖工作区文件（敏感）。\n"
    "  · delete_local_file —— 删除工作区文件（高风险）。\n"
    "\n"
    "【工作流】\n"
    "  · 需要读密钥、写文件、删文件时，直接调用对应工具即可；\n"
    "    系统会自动暂停并请求人工审批，你不需要在文本中反复确认。\n"
    "  · 若人工拒绝或修改了你的调用，请认真阅读返回消息并按指示调整。\n"
    "  · 请参考 /memories/AGENTS.md 中的用户偏好来调整回答风格。\n"
    "  · 用户明确说“记住……”时，将内容写入 /memories/AGENTS.md。\n"
)


def build_agent(seed_memory: bool = True):
    """返回 (agent, store)。"""
    init_environment()

    # ---------- 1. 持久化存储 + 播种记忆 ----------
    store = InMemoryStore()
    if seed_memory:
        store.put(
            MEMORY_NAMESPACE,
            MEMORY_PATH,
            create_file_data(
                "## 用户偏好\n"
                "- 用户是后端工程师，偏好 Python 示例\n"
                "- 回答尽量简洁，先给结论\n"
            ),
        )

    # ---------- 2. 检查点（HITL 必需） ----------
    checkpointer = InMemorySaver()

    # ---------- 3. 后端：直接构造实例（0.7 写法） ----------
    # ⚠️ 0.6 写法（会报错）:
    #     backend=lambda rt: CompositeBackend(
    #         default=StateBackend(rt),
    #         routes={"/memories/": StoreBackend(rt, store=store,
    #                                           namespace=lambda _rt: MEMORY_NAMESPACE)},
    #     )
    # ✅ 0.7 写法:
    backend = CompositeBackend(
        default=StateBackend(),                     # ✅ 不再传 rt
        routes={
            "/memories/": StoreBackend(
                store=store,
                namespace=lambda _rt: MEMORY_NAMESPACE,   # namespace 工厂仍允许
            ),                                      # ✅ 不再传 rt
        },
    )

    # ---------- 4. 创建 Agent ----------
    agent = create_deep_agent(
        model=get_model(),
        system_prompt=SYSTEM_PROMPT,
        tools=[
            list_workspace,
            read_local_file,
            get_secret,
            write_local_file,
            delete_local_file,
        ],
        memory=[MEMORY_PATH],
        backend=backend,                            # ✅ 传实例，不是 lambda
        store=store,
        checkpointer=checkpointer,
        interrupt_on={
            # 高敏感：只允许 approve / reject（防止改 key 去读别的密钥）
            "get_secret": {"allowed_decisions": ["approve", "reject"]},
            # 高风险：允许全部三种决策
            "delete_local_file": {"allowed_decisions": ["approve", "edit", "reject"]},
            # 敏感：允许全部三种决策
            "write_local_file": {"allowed_decisions": ["approve", "edit", "reject"]},
            # 只读工具不声明 → 自动执行
        },
    )
    return agent, store

```

main.py

```python
# ch08/main.py
# ---------- 强制 UTF-8 标准流 ----------
import sys
for _s in (sys.stdin, sys.stdout, sys.stderr):
    try:
        _s.reconfigure(encoding="utf-8", errors="replace")   # ← 加上 errors
    except Exception:
        pass

# ---------- 兜底：拦截 openai 的 JSON 序列化，清理非法代理字符 ----------
def _strip_surrogates(o):
    if isinstance(o, str):
        return "".join(ch for ch in o if not (0xD800 <= ord(ch) <= 0xDFFF))
    if isinstance(o, dict):
        return {k: _strip_surrogates(v) for k, v in o.items()}
    if isinstance(o, (list, tuple)):
        return type(o)(_strip_surrogates(x) for x in o)
    return o


def _find_surrogates(o, path="$"):
    """递归找第一处含代理字符的字符串，返回 (path, snippet)。"""
    if isinstance(o, str):
        if any(0xD800 <= ord(c) <= 0xDFFF for c in o):
            i = next(j for j, c in enumerate(o) if 0xD800 <= ord(c) <= 0xDFFF)
            return path, o[max(0, i - 60):i + 60]
        return None
    if isinstance(o, dict):
        for k, v in o.items():
            r = _find_surrogates(v, f"{path}.{k}")
            if r:
                return r
    if isinstance(o, (list, tuple)):
        for i, v in enumerate(o):
            r = _find_surrogates(v, f"{path}[{i}]")
            if r:
                return r
    return None


def _install_openai_sanitizer():
    try:
        from openai._utils import _json as _oai_json
        import openai._base_client as _bc
    except ImportError as e:
        print(f"[sanitizer] 无法导入 openai 内部模块: {e}")
        return

    _orig = _oai_json.openapi_dumps

    def _safe_dumps(obj, **kw):
        hit = _find_surrogates(obj)
        if hit:
            path, snippet = hit
            print(f"[sanitizer] 检出非法代理字符 @ {path}")
            print(f"[sanitizer] 上下文: ...{snippet!r}...")
        return _orig(_strip_surrogates(obj), **kw)

    _oai_json.openapi_dumps = _safe_dumps
    _bc.openapi_dumps = _safe_dumps          # 重新绑定引用
    print("[sanitizer] 已安装 openai JSON 兜底清理")


_install_openai_sanitizer()

# ---------- 以下是原有内容 ----------
from agent_setup import build_agent
from demo_hitl import run_hitl_demo
from demo_memory import run_memory_demo
from workspace import init_environment


def main():
    init_environment()
    print("正在初始化 DeepAgent（首次调用模型可能稍慢）...")
    agent, store = build_agent(seed_memory=True)
    print("初始化完成。")

    while True:
        print("\n" + "=" * 60)
        print("DeepAgents 真实操作演示")
        print("=" * 60)
        print("  1. 跨轮对话长期记忆验证")
        print("  2. 真实操作 + 人工审批（读密钥 / 写 / 删）")
        print("  q. 退出")
        choice = input("请选择 > ").strip().lower()

        if choice == "1":
            run_memory_demo(agent, store)
        elif choice == "2":
            run_hitl_demo(agent)
        elif choice == "q":
            print("再见 👋")
            return
        else:
            print("无效选项。")


if __name__ == "__main__":
    main()

```



# 踩坑记录

1.创建环境变量中的base_url的时候，要记得加上前面的https

2.修改的权限给了agent之后，一定要更新agent接受的上下文，不然agent使用read_file工具进行检查的时候，就会误以为自己修改错了，然后回退修改。

3.在wsl的控制台中进行输入的时候，尽量使用全英文，避免一些奇葩报错

# 关键内容

1.Agent的两种记忆及其实现路径

![image-20260925212124057](https://cdn.jsdelivr.net/gh/zbhgis/BlogImg2@main/blog/202609252121724.png)

2.四种记忆的实用场景

用户偏好记忆，自我改进的agent，知识库累积，研究项目持续推进

3.记忆的6个维度

![image-20260925212418283](https://cdn.jsdelivr.net/gh/zbhgis/BlogImg2@main/blog/202609252124052.png)

4.human-in-the-loop的流程

![image-20260925222317965](https://cdn.jsdelivr.net/gh/zbhgis/BlogImg2@main/blog/202609252223493.png)



# 参考教程

第8/9章

https://datawhalechina.github.io/deepagents-in-action/chapters/ch08-long-term-memory/

https://datawhalechina.github.io/deepagents-in-action/chapters/ch09-human-in-the-loop/