deepagents学习打卡day03

============================================================
正在测试后端: STATE
============================================================

======================================================================
【state】 测试报告
======================================================================
状态: ✅ 成功
输出预览:

研究已完成，报告已保存到 `/workspace/report.md`。

## 报告要点速览

**Deep Agents** 是 LangChain AI 团队构建在 LangChain + LangGraph 之上的开源 agent harness 框架，其 **虚拟文件系统（VFS）**...
文件列表: (空)

物理路径:
  存储方式: 内存（Graph State），无物理文件

============================================================
正在测试后端: STORE
============================================================

======================================================================
【store】 测试报告
======================================================================
状态: ✅ 成功
输出预览:

已完成研究并将报告保存到 `/workspace/report.md`。

**报告概要：**

1. **背景与定位**：Deep Agents 是 LangChain 的开源 agent harness，其虚拟文件系统是该项目的核心抽象之一。

2. **核心架构**：
   - **前端*...
文件列表: (空)

物理路径:
  存储方式: InMemoryStore（内存版 Store，进程结束即丢）
  提示: 如需持久化，替换为 Redis/Postgres 的 Store 实例

============================================================
正在测试后端: FILESYSTEM
============================================================

======================================================================
【filesystem】 测试报告
======================================================================
状态: ✅ 成功
输出预览:

抱歉，刚才误触发了一次编辑。文件已在上一步成功保存。以下是本次研究的执行摘要：

## ✅ 研究完成

**已保存文件**：`/workspace/report.md`

### 执行步骤回顾
1. **网络检索**：围绕 Deep Agents 的虚拟文件系统进行了 5 组关键词搜索，覆盖官方...
文件列表 (1 项):
  - /workspace/report.md [15870 bytes] (来源: disk)

物理路径:
  根目录: /home/zbhgis/projects/research_deepagent/ch03/test_workspace
  虚拟路径 /workspace/report.md 会落在: /home/zbhgis/projects/research_deepagent/ch03/test_workspace/workspace/report.md

============================================================
正在测试后端: COMPOSITE
============================================================

======================================================================
【composite】 测试报告
======================================================================
状态: ✅ 成功
输出预览:

报告已保存到 `/workspace/report.md`。以下是研究要点摘要：

## 核心发现

**Deep Agents 的虚拟文件系统是一套"路径 = 存储后端"的可插拔抽象层**，让 Agent 只需面对统一的一套文件工具（`ls` / `read_file` / `write_fi...
文件列表 (1 项):
  - /report.md [21832 bytes] (来源: disk)

物理路径:
  /workspace/ 路由根目录: /home/zbhgis/projects/research_deepagent/ch03/test_workspace_composite
  虚拟路径 /workspace/report.md 会落在: /home/zbhgis/projects/research_deepagent/ch03/test_workspace_composite/workspace/report.md
  默认路由: StateBackend（内存）
  /memories/ 路由: StoreBackend（内存）

"""
共享模块：模型、搜索工具、系统提示词、文件提取工具函数。
所有后端测试脚本都从这里导入。
"""

import os
from typing import Literal
from dotenv import load_dotenv

from tavily import TavilyClient
from langchain_openai import ChatOpenAI

load_dotenv()

# ===== 路径基准 =====
BASE_DIR = os.path.dirname(os.path.abspath(__file__))

# ===== 1. 配置模型 =====
model = ChatOpenAI(
    model=os.environ.get("LLM_MODEL_NAME"),
    api_key=os.environ.get("LLM_API_KEY"),
    base_url=os.environ.get("LLM_BASE_URL"),
)

# ===== 2. 初始化搜索客户端 =====
tavily_client = TavilyClient(api_key=os.environ.get("TAVILY_API_KEY"))


# ===== 3. 定义搜索工具 =====
def internet_search(
    query: str,
    max_results: int = 5,
    topic: Literal["general", "news", "finance"] = "general",
    include_raw_content: bool = False,
):
    """Run a web search for the given query."""
    return tavily_client.search(
        query,
        max_results=max_results,
        include_raw_content=include_raw_content,
        topic=topic,
    )


# ===== 4. 定义系统提示词 =====
research_instructions = """你是一位专业的研究员。
你的工作是进行深入研究，然后撰写一份完整的研究报告。
你可以使用 internet_search 工具搜索互联网获取信息。
另外，请务必使用 write_file 工具将你的研究报告保存到 /workspace/report.md 文件中。
"""

# ===== 5. 默认测试任务 =====
DEFAULT_QUERY = (
    "请研究一下 Deep Agents 的虚拟文件系统，并将研究报告保存到 /workspace/report.md"
)


# ===== 6. 工具函数：给 LangSmith trace 打标签的 config =====
def make_run_config(name: str, query: str) -> dict:
    """构造 invoke 时传给 LangSmith 的 config。"""
    return {
        "run_name": f"backend-compare::{name}",
        "tags": ["backend-compare", f"backend:{name}"],
        "metadata": {
            "backend": name,
            "model": getattr(model, "model_name", "unknown"),
            "query": query,
        },
    }


# ===== 7. 工具函数：磁盘扫描 =====
def scan_disk(scan_dir: str) -> list:
    """扫描目录下的所有文件，返回 [{path, size, source='disk'}]。"""
    files = []
    if not os.path.exists(scan_dir):
        return files
    for root, dirs, filenames in os.walk(scan_dir):
        for fn in filenames:
            fp = os.path.join(root, fn)
            rel = os.path.relpath(fp, scan_dir)
            files.append(
                {
                    "path": "/" + rel.replace(os.sep, "/"),
                    "size": os.path.getsize(fp),
                    "source": "disk",
                }
            )
    return files


# ===== 8. 工具函数：去重（disk > store > state） =====
def dedupe_files(collected: list) -> list:
    priority = {"disk": 0, "store": 1, "state": 2}
    by_path = {}
    for item in collected:
        p = item["path"]
        if (
            p not in by_path
            or priority[item["source"]] < priority[by_path[p]["source"]]
        ):
            by_path[p] = item
    return sorted(by_path.values(), key=lambda x: x["path"])


# ===== 9. 工具函数：从 StateBackend 提取文件 =====
def extract_state_files(result: dict) -> list:
    files = []
    file_dict = result.get("files", {})
    if isinstance(file_dict, dict):
        for k, v in file_dict.items():
            size = len(v) if isinstance(v, str) else 0
            if size > 0:
                files.append({"path": k, "size": size, "source": "state"})
    return files


# ===== 10. 工具函数：从 StoreBackend 的 ls() 提取文件 =====
def extract_store_files(backend) -> list:
    files = []
    try:
        ls_result = backend.ls("/")
        entries = getattr(ls_result, "entries", ls_result) or []
        for e in entries:
            if isinstance(e, dict):
                size = e.get("size", 0)
                is_dir = e.get("is_dir", False)
                if size > 0 and not is_dir:
                    files.append(
                        {
                            "path": e.get("path", "?"),
                            "size": size,
                            "source": "store",
                        }
                    )
    except Exception as e:
        print(f"  [警告] StoreBackend ls() 失败: {e}")
    return files


# ===== 11. 统一打印报告 =====
def print_report(name: str, result: dict, extra_paths: dict = None):
    """
    打印单个后端的测试报告。
    - result: test_backend 的返回结果
    - extra_paths: 额外打印物理路径信息，如 {"磁盘路径": "/xxx"}
    """
    print("\n" + "=" * 70)
    print(f"【{name}】 测试报告")
    print("=" * 70)

    status = "✅ 成功" if result["status"] == "success" else "❌ 失败"
    print(f"状态: {status}")

    if result["status"] == "success":
        print(f"输出预览: {result['output_preview']}...")
        if result["files"]:
            print(f"文件列表 ({len(result['files'])} 项):")
            for f in result["files"]:
                print(f"  - {f['path']} [{f['size']} bytes] (来源: {f['source']})")
        else:
            print("文件列表: (空)")
    else:
        print(f"错误: {result.get('error', 'Unknown')}")

    if extra_paths:
        print("\n物理路径:")
        for label, path in extra_paths.items():
            print(f"  {label}: {path}")

"""
测试 StateBackend（默认内存后端）。
"""

from deepagents import create_deep_agent
from deepagents.backends import StateBackend

from common import (
    model,
    internet_search,
    research_instructions,
    DEFAULT_QUERY,
    make_run_config,
    extract_state_files,
    dedupe_files,
    print_report,
)

BACKEND_NAME = "state"


def test_backend(query: str = DEFAULT_QUERY) -> dict:
    print(f"\n{'='*60}")
    print(f"正在测试后端: {BACKEND_NAME.upper()}")
    print(f"{'='*60}")

    try:
        backend = StateBackend()

        agent = create_deep_agent(
            model=model,
            tools=[internet_search],
            system_prompt=research_instructions,
            backend=backend,
        )

        result = agent.invoke(
            {"messages": [{"role": "user", "content": query}]},
            config=make_run_config(BACKEND_NAME, query),
        )

        output = result["messages"][-1].content

        # StateBackend 的文件只存在于 result["files"] 中
        files = dedupe_files(extract_state_files(result))

        return {
            "backend": BACKEND_NAME,
            "status": "success",
            "output_preview": output[:150],
            "files": files,
        }

    except Exception as e:
        return {
            "backend": BACKEND_NAME,
            "status": "failed",
            "output_preview": "",
            "files": [],
            "error": str(e),
        }


if __name__ == "__main__":
    res = test_backend()
    print_report(
        BACKEND_NAME,
        res,
        extra_paths={
            "存储方式": "内存（Graph State），无物理文件",
        },
    )

"""
测试 StoreBackend（LangGraph Store 后端，使用 InMemoryStore）。
"""

from langgraph.store.memory import InMemoryStore
from deepagents import create_deep_agent
from deepagents.backends import StoreBackend

from common import (
    model,
    internet_search,
    research_instructions,
    DEFAULT_QUERY,
    make_run_config,
    extract_store_files,
    dedupe_files,
    print_report,
)

BACKEND_NAME = "store"


def test_backend(query: str = DEFAULT_QUERY) -> dict:
    print(f"\n{'='*60}")
    print(f"正在测试后端: {BACKEND_NAME.upper()}")
    print(f"{'='*60}")

    try:
        backend = StoreBackend(
            store=InMemoryStore(), namespace=lambda _runtime: ("filesystem",)
        )

        agent = create_deep_agent(
            model=model,
            tools=[internet_search],
            system_prompt=research_instructions,
            backend=backend,
        )

        result = agent.invoke(
            {"messages": [{"role": "user", "content": query}]},
            config=make_run_config(BACKEND_NAME, query),
        )

        output = result["messages"][-1].content

        # StoreBackend 的文件需要调用 ls() 读取
        files = dedupe_files(extract_store_files(backend))

        return {
            "backend": BACKEND_NAME,
            "status": "success",
            "output_preview": output[:150],
            "files": files,
        }

    except Exception as e:
        return {
            "backend": BACKEND_NAME,
            "status": "failed",
            "output_preview": "",
            "files": [],
            "error": str(e),
        }


if __name__ == "__main__":
    res = test_backend()
    print_report(
        BACKEND_NAME,
        res,
        extra_paths={
            "存储方式": "InMemoryStore（内存版 Store，进程结束即丢）",
            "提示": "如需持久化，替换为 Redis/Postgres 的 Store 实例",
        },
    )

"""
测试 FilesystemBackend（本地磁盘后端）。
"""

import os
from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend

from common import (
    BASE_DIR,
    model,
    internet_search,
    research_instructions,
    DEFAULT_QUERY,
    make_run_config,
    scan_disk,
    extract_state_files,
    dedupe_files,
    print_report,
)

BACKEND_NAME = "filesystem"

# 独立目录
WORKSPACE_DIR = os.path.join(BASE_DIR, "test_workspace")


def test_backend(query: str = DEFAULT_QUERY) -> dict:
    print(f"\n{'='*60}")
    print(f"正在测试后端: {BACKEND_NAME.upper()}")
    print(f"{'='*60}")

    os.makedirs(WORKSPACE_DIR, exist_ok=True)

    try:
        backend = FilesystemBackend(
            root_dir=WORKSPACE_DIR,
            virtual_mode=True,
        )

        agent = create_deep_agent(
            model=model,
            tools=[internet_search],
            system_prompt=research_instructions,
            backend=backend,
        )

        result = agent.invoke(
            {"messages": [{"role": "user", "content": query}]},
            config=make_run_config(BACKEND_NAME, query),
        )

        output = result["messages"][-1].content

        # FilesystemBackend 的文件真实落盘，扫描磁盘即可
        collected = scan_disk(WORKSPACE_DIR)
        files = dedupe_files(collected)

        return {
            "backend": BACKEND_NAME,
            "status": "success",
            "output_preview": output[:150],
            "files": files,
        }

    except Exception as e:
        return {
            "backend": BACKEND_NAME,
            "status": "failed",
            "output_preview": "",
            "files": [],
            "error": str(e),
        }


if __name__ == "__main__":
    res = test_backend()
    print_report(
        BACKEND_NAME,
        res,
        extra_paths={
            "根目录": WORKSPACE_DIR,
            "虚拟路径 /workspace/report.md 会落在": os.path.join(
                WORKSPACE_DIR, "workspace", "report.md"
            ),
        },
    )

"""
测试 CompositeBackend（混合路由后端）。
/workspace/  → FilesystemBackend（落盘）
/memories/   → StoreBackend（内存 Store）
默认          → StateBackend（内存 State）
"""

import os
from langgraph.store.memory import InMemoryStore
from deepagents import create_deep_agent
from deepagents.backends import (
    StateBackend,
    FilesystemBackend,
    StoreBackend,
    CompositeBackend,
)

from common import (
    BASE_DIR,
    model,
    internet_search,
    research_instructions,
    DEFAULT_QUERY,
    make_run_config,
    scan_disk,
    extract_state_files,
    extract_store_files,
    dedupe_files,
    print_report,
)

BACKEND_NAME = "composite"

# 独立目录，避免与其他后端的 WORKSPACE_DIR 污染
COMPOSITE_ROUTE_DIR = os.path.join(BASE_DIR, "test_workspace_composite")


def test_backend(query: str = DEFAULT_QUERY) -> dict:
    print(f"\n{'='*60}")
    print(f"正在测试后端: {BACKEND_NAME.upper()}")
    print(f"{'='*60}")

    os.makedirs(COMPOSITE_ROUTE_DIR, exist_ok=True)

    try:
        backend = CompositeBackend(
            default=StateBackend(),
            routes={
                "/workspace/": FilesystemBackend(
                    root_dir=COMPOSITE_ROUTE_DIR,
                    virtual_mode=True,
                ),
                "/memories/": StoreBackend(
                    store=InMemoryStore(), namespace=lambda _runtime: ("filesystem",)
                ),
            },
        )

        agent = create_deep_agent(
            model=model,
            tools=[internet_search],
            system_prompt=research_instructions,
            backend=backend,
        )

        result = agent.invoke(
            {"messages": [{"role": "user", "content": query}]},
            config=make_run_config(BACKEND_NAME, query),
        )

        output = result["messages"][-1].content

        # CompositeBackend 的文件散落在多个后端，需要分别收集
        collected = []
        # 1) StateBackend 部分
        collected.extend(extract_state_files(result))
        # 2) 磁盘部分（/workspace/ 路由）
        collected.extend(scan_disk(COMPOSITE_ROUTE_DIR))
        # 3) Store 部分（/memories/ 路由）
        collected.extend(extract_store_files(backend.routes.get("/memories/")))

        files = dedupe_files(collected)

        return {
            "backend": BACKEND_NAME,
            "status": "success",
            "output_preview": output[:150],
            "files": files,
        }

    except Exception as e:
        return {
            "backend": BACKEND_NAME,
            "status": "failed",
            "output_preview": "",
            "files": [],
            "error": str(e),
        }


if __name__ == "__main__":
    res = test_backend()
    print_report(
        BACKEND_NAME,
        res,
        extra_paths={
            "/workspace/ 路由根目录": COMPOSITE_ROUTE_DIR,
            "虚拟路径 /workspace/report.md 会落在": os.path.join(
                COMPOSITE_ROUTE_DIR, "workspace", "report.md"
            ),
            "默认路由": "StateBackend（内存）",
            "/memories/ 路由": "StoreBackend（内存）",
        },
    )

cp ~/projects/research_deepagent/search_agent.py /mnt/c/Users/lenovo2026/Desktop/

export LLM_BASE_URL="https://token.sensenova.cn/v1"
export LLM_API_KEY=""
export LLM_MODEL_NAME="sensenova-6.8-flash-lite"