From 7dfc9968fe5e93eb23c8ed3361532ae899e83387 Mon Sep 17 00:00:00 2001 From: 1445043649 <> Date: Sat, 27 Jun 2026 22:02:44 +0800 Subject: [PATCH] =?UTF-8?q?feat(apps):=20=E6=96=B0=E5=A2=9E=20DeerFlow=20?= =?UTF-8?q?=E5=BA=94=E7=94=A8=E8=84=9A=E6=89=8B=E6=9E=B6=EF=BC=8C=E5=B9=B6?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20store=20=E7=9A=84=20database=20=E5=9B=9E?= =?UTF-8?q?=E9=80=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit apps/：新增「基于 DeerFlow 的应用」目录，与 backend/、frontend/ 平级， 位于「app 消费 deerflow、不反向依赖」边界的正确侧。含两种集成示例： - examples/http-chat —— HTTP Gateway (REST+SSE)，含登录/CSRF/建线程/流式对话 - examples/embedded-chat —— 进程内直接调 DeerFlowClient README 说明边界规则、两种模式、鉴权流程及新建应用约定。 runtime/store：修复 make_store 缺失的 database 段回退。原先 store 工厂只读 legacy 的 checkpointer 段，导致仅配 database:postgres 时，checkpointer 走了 Postgres、但 store 仍回退 InMemoryStore（并打出误导性的「线程列表会丢失」告警， 实际线程在 threads_meta 表里、本就持久）。现对齐 checkpointer 工厂的优先级： checkpointer 段 → database 段 → InMemoryStore；postgres 分支同样剥掉 +asyncpg 方言前缀，使一个 DATABASE_URL 同时满足 SQLAlchemy 与 LangGraph 的 psycopg store。 告警文案也修正为「跨线程 store 数据会丢失」。 tests：新增 test_store_provider.py（3 例，TDD）覆盖 database→postgres 回退、 无配置时的内存回退、以及 checkpointer 段优先级。 Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/.gitignore | 9 ++ apps/README.md | 52 ++++++++ apps/examples/embedded-chat/app.py | 46 +++++++ apps/examples/http-chat/app.py | 125 ++++++++++++++++++ apps/examples/http-chat/requirements.txt | 1 + .../deerflow/runtime/store/async_provider.py | 81 ++++++++++-- backend/tests/test_store_provider.py | 89 +++++++++++++ 7 files changed, 394 insertions(+), 9 deletions(-) create mode 100644 apps/.gitignore create mode 100644 apps/README.md create mode 100644 apps/examples/embedded-chat/app.py create mode 100644 apps/examples/http-chat/app.py create mode 100644 apps/examples/http-chat/requirements.txt create mode 100644 backend/tests/test_store_provider.py diff --git a/apps/.gitignore b/apps/.gitignore new file mode 100644 index 00000000..5f5e2a21 --- /dev/null +++ b/apps/.gitignore @@ -0,0 +1,9 @@ +# 本地依赖 / 虚拟环境 +.venv/ +__pycache__/ +*.pyc + +# 鉴权 cookie / 本地产物 +jar.txt +*.local +.env diff --git a/apps/README.md b/apps/README.md new file mode 100644 index 00000000..f93eb3ba --- /dev/null +++ b/apps/README.md @@ -0,0 +1,52 @@ +# apps/ — 基于 DeerFlow 的应用 + +这个目录用于存放**消费 DeerFlow 智能体能力的上层应用**。每个应用一个子文件夹。 + +## 为什么放在这里 + +DeerFlow 的代码有一条严格的依赖方向（见根 `CLAUDE.md`）： + +``` +backend/packages/harness/deerflow/ ← 可发布的 Agent 框架（deerflow.*） +backend/app/ ← Gateway / IM 通道（app.*） +apps/ ← 你的应用（消费 deerflow，不反向依赖） ← 本目录 +``` + +规则：**app 可以依赖 deerflow，deerflow 不能依赖 app / apps**。本目录放在 `backend/` 之外、与 `frontend/` 平级，天然符合这条边界。 + +## 两种集成模式 + +| 模式 | 适用场景 | 怎么连 | 示例 | +|---|---|---|---| +| **HTTP Gateway**（REST+SSE） | 上层是别的服务 / 多语言 | 调 `http://localhost:2026/api/*` | [`examples/http-chat/`](examples/http-chat/) | +| **内嵌 DeerFlowClient** | 上层本身是 Python，进程内直接当 SDK 调 | `from deerflow.client import DeerFlowClient` | [`examples/embedded-chat/`](examples/embedded-chat/) | + +> 还有第三种：LangGraph SDK（`langgraph_sdk.get_client(url=".../api")`，graph id `lead_agent`），用于接入 LangGraph 生态工具链。需要的话照 HTTP 示例的鉴权流程拿 cookie 即可。 + +## 前置：先把 DeerFlow 跑起来 + +在**仓库根目录**： + +```bash +make dev # 起 Gateway(8001) + 前端(3000) + nginx(2026)，统一入口 http://localhost:2026 +``` + +确保 `config.yaml` 里至少配了一个可用模型 + API key。 + +## 鉴权（HTTP 模式必读） + +Gateway 是 **fail-closed** 的——除少数公开路径外所有请求都要带会话 cookie： + +1. `GET /api/v1/auth/setup-status` → 是否还没管理员 +2. 首次 `POST /api/v1/auth/initialize`（JSON `{email,password}`）建第一个管理员；之后 `POST /api/v1/auth/login/local`（**表单** `username`=邮箱 + `password`） +3. 成功后 Session 里有 `access_token`(HttpOnly) + `csrf_token` 两个 cookie +4. **所有写请求**（POST/PUT/DELETE/PATCH）必须带 `X-CSRF-Token` 头 = `csrf_token` 值 + +> 多租户：当前 `docs/multi-tenant-redesign` 分支的 API Key 鉴权中间件尚未接入，外部系统暂时只能走会话 cookie。等 `Authorization: Bearer dfk_live_...` 落地后再补无人值守接入。 + +## 新建一个应用 + +```bash +mkdir apps/my-app +# 放你的代码；HTTP 模式参照 examples/http-chat，内嵌模式参照 examples/embedded-chat +``` diff --git a/apps/examples/embedded-chat/app.py b/apps/examples/embedded-chat/app.py new file mode 100644 index 00000000..6c9d2d71 --- /dev/null +++ b/apps/examples/embedded-chat/app.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python3 +""" +内嵌模式示例：进程内直接把 DeerFlow 当 SDK 调，不起 HTTP。 + +必须在 backend 的 uv 环境里跑（这样才能 import deerflow.*）： + cd backend + uv run python ../apps/examples/embedded-chat/app.py + +依赖 config.yaml 里配好至少一个可用模型 + API key（路径解析见根 CLAUDE.md）。 + +API 对照：backend/packages/harness/deerflow/client.py +""" + +from deerflow.client import DeerFlowClient +from deerflow.runtime.checkpointer.provider import get_checkpointer + + +def main() -> None: + # checkpointer 提供跨轮状态持久化（sqlite/postgres 由 config.yaml 决定） + client = DeerFlowClient( + checkpointer=get_checkpointer(), + thinking_enabled=True, + ) + + thread_id = "embedded-demo-1" + + # ① 流式：stream() 产出 StreamEvent + print("👤 用一句话介绍你自己，然后心算 17 * 23。\n🤖 ", end="", flush=True) + for ev in client.stream("用一句话介绍你自己，然后心算 17 * 23。", thread_id=thread_id): + if ev.type == "messages-tuple" and ev.data.get("type") == "ai": + print(ev.data.get("content", ""), end="", flush=True) # AI 文本增量 + elif ev.type == "end": + print(f"\n[usage] {ev.data.get('usage')}") + + # ② 阻塞式：chat() 直接返回完整 AI 文本（复用 thread_id 即多轮） + print("\n👤 刚才结果再乘以 2 是多少？") + answer = client.chat("刚才结果再乘以 2 是多少？", thread_id=thread_id) + print(f"🤖 {answer}") + + # 其它能力：list_models() / list_skills() / get_memory() / upload_files() ... + models = client.list_models().get("models", []) + print(f"\n[已配置模型] {[m.get('name') for m in models]}") + + +if __name__ == "__main__": + main() diff --git a/apps/examples/http-chat/app.py b/apps/examples/http-chat/app.py new file mode 100644 index 00000000..9ef9b7df --- /dev/null +++ b/apps/examples/http-chat/app.py @@ -0,0 +1,125 @@ +#!/usr/bin/env python3 +""" +HTTP 模式示例：把 DeerFlow 当底层服务，通过 Gateway (REST + SSE) 对话。 + +运行： + pip install -r requirements.txt + python app.py # 前提：仓库根目录已 `make dev` + +字段 / 事件名均已对照后端源码核对： + 鉴权 backend/app/gateway/routers/auth.py + auth_middleware.py + csrf_middleware.py + 线程/运行 backend/app/gateway/routers/threads.py + thread_runs.py + SSE 事件名 backend/packages/harness/deerflow/runtime/runs/worker.py +""" + +import os +import json +import requests + +# 默认走 nginx(:2026);只起了 Gateway 时用 BASE=http://localhost:8001 覆盖 +BASE = os.environ.get("DF_BASE", "http://localhost:2026") +EMAIL = os.environ.get("DF_EMAIL", "admin@example.com") +PASSWORD = os.environ.get("DF_PASSWORD", "change-me-please-123") # 至少 8 位，避免弱口令 + + +def authenticate(s: requests.Session) -> None: + """首启则初始化管理员，否则登录。成功后 cookie 落在 session。""" + status = s.get(f"{BASE}/api/v1/auth/setup-status").json() + if status.get("needs_setup"): + print("→ 首次启动，创建管理员账号") + r = s.post(f"{BASE}/api/v1/auth/initialize", + json={"email": EMAIL, "password": PASSWORD}) + else: + print("→ 已有账号，登录") + # login/local 是 OAuth2 表单：字段名 username（填邮箱）+ password + r = s.post(f"{BASE}/api/v1/auth/login/local", + data={"username": EMAIL, "password": PASSWORD}) + r.raise_for_status() + print(" cookies:", list(s.cookies.keys())) + + +def _csrf(s: requests.Session) -> dict: + """双提交 cookie 模式：csrf_token cookie 的值放进 X-CSRF-Token 头。""" + token = s.cookies.get("csrf_token") + if not token: + raise RuntimeError("缺少 csrf_token cookie —— 鉴权可能失败") + return {"X-CSRF-Token": token} + + +def create_thread(s: requests.Session) -> str: + r = s.post(f"{BASE}/api/threads", json={}, headers=_csrf(s)) + r.raise_for_status() + tid = r.json()["thread_id"] + print(f"→ 线程已创建: {tid}") + return tid + + +_seen_text = "" # 简单状态，按需扩展为按 message-id 维护 + + +def stream_chat(s: requests.Session, thread_id: str, message: str) -> None: + body = { + "assistant_id": "lead_agent", # 见 backend/langgraph.json + "input": {"messages": [{"role": "user", "content": message}]}, + "stream_mode": ["messages-tuple", "values"], # 增量文本 + 全量状态 + } + headers = {**_csrf(s), "Accept": "text/event-stream"} + + with s.post(f"{BASE}/api/threads/{thread_id}/runs/stream", + json=body, headers=headers, stream=True) as resp: + resp.raise_for_status() + print(f"\n👤 {message}\n🤖 ", end="", flush=True) + + event, buf = None, [] + for raw in resp.iter_lines(decode_unicode=True): + if raw is None: + continue + line = raw.strip() + if line == "": # 一帧结束 + if event: + _handle(event, "\n".join(buf)) + event, buf = None, [] + elif line.startswith(":"): # 心跳注释 + continue + elif line.startswith("event:"): + event = line[6:].strip() + elif line.startswith("data:"): + buf.append(line[5:].strip()) + print() + + +def _handle(event: str, data: str) -> None: + if event == "end" or not data: + return + try: + payload = json.loads(data) + except json.JSONDecodeError: + return + if event == "messages": + # 形如 [chunk_dict, metadata_dict]；AI 文本是增量 + chunk = payload[0] if isinstance(payload, list) and payload else {} + if chunk.get("type") in ("ai", "AIMessageChunk"): + content = chunk.get("content") + text = content if isinstance(content, str) else _flatten(content) + if text: + print(text, end="", flush=True) + # event == "metadata" → {run_id, thread_id} + # event == "values" → 全量状态快照（title / messages / artifacts ...） + + +def _flatten(content) -> str: + if isinstance(content, list): + return "".join(b.get("text", "") for b in content if isinstance(b, dict)) + return "" + + +def main() -> None: + s = requests.Session() + authenticate(s) + tid = create_thread(s) + stream_chat(s, tid, "用一句话介绍你自己，然后心算 17 * 23。") + stream_chat(s, tid, "刚才结果再乘以 2 是多少？") # 复用 thread_id 即多轮 + + +if __name__ == "__main__": + main() diff --git a/apps/examples/http-chat/requirements.txt b/apps/examples/http-chat/requirements.txt new file mode 100644 index 00000000..535409c2 --- /dev/null +++ b/apps/examples/http-chat/requirements.txt @@ -0,0 +1 @@ +requests>=2.31 diff --git a/backend/packages/harness/deerflow/runtime/store/async_provider.py b/backend/packages/harness/deerflow/runtime/store/async_provider.py index bc1f07eb..729c3ee3 100644 --- a/backend/packages/harness/deerflow/runtime/store/async_provider.py +++ b/backend/packages/harness/deerflow/runtime/store/async_provider.py @@ -19,6 +19,7 @@ from __future__ import annotations import contextlib import logging +import re from collections.abc import AsyncIterator from langgraph.store.base import BaseStore @@ -80,6 +81,57 @@ async def _async_store(config) -> AsyncIterator[BaseStore]: raise ValueError(f"Unknown store backend type: {config.type!r}") +@contextlib.asynccontextmanager +async def _async_store_from_database(db_config) -> AsyncIterator[BaseStore]: + """Async context manager that constructs a Store from a unified DatabaseConfig. + + Mirrors :func:`deerflow.runtime.checkpointer.async_provider._async_checkpointer_from_database` + so the store and checkpointer share one ``database`` section. + """ + if db_config.backend == "memory": + from langgraph.store.memory import InMemoryStore + + logger.info("Store: using InMemoryStore (in-process, not persistent)") + yield InMemoryStore() + return + + if db_config.backend == "sqlite": + try: + from langgraph.store.sqlite.aio import AsyncSqliteStore + except ImportError as exc: + raise ImportError(SQLITE_STORE_INSTALL) from exc + + conn_str = db_config.sqlite_path + ensure_sqlite_parent_dir(conn_str) + async with AsyncSqliteStore.from_conn_string(conn_str) as store: + await store.setup() + logger.info("Store: using AsyncSqliteStore (%s)", conn_str) + yield store + return + + if db_config.backend == "postgres": + try: + from langgraph.store.postgres.aio import AsyncPostgresStore # type: ignore[import] + except ImportError as exc: + raise ImportError(POSTGRES_STORE_INSTALL) from exc + + if not db_config.postgres_url: + raise ValueError(POSTGRES_CONN_REQUIRED) + + # LangGraph's AsyncPostgresStore wraps psycopg and expects a libpq-style + # conninfo (`postgresql://...`). DeerFlow's SQLAlchemy engine uses the + # same URL with the `+asyncpg` dialect prefix — strip it so one + # DATABASE_URL satisfies both paths (same as the checkpointer factory). + lg_conn_str = re.sub(r"^postgresql\+\w+://", "postgresql://", db_config.postgres_url) + async with AsyncPostgresStore.from_conn_string(lg_conn_str) as store: + await store.setup() + logger.info("Store: using AsyncPostgresStore") + yield store + return + + raise ValueError(f"Unknown database backend: {db_config.backend!r}") + + # --------------------------------------------------------------------------- # Public async context manager # --------------------------------------------------------------------------- @@ -97,18 +149,29 @@ async def make_store(app_config: AppConfig | None = None) -> AsyncIterator[BaseS async with make_store(app_config) as store: app.state.store = store - Yields an :class:`~langgraph.store.memory.InMemoryStore` when no - ``checkpointer`` section is configured (emits a WARNING in that case). + Priority (mirrors the checkpointer factory): + 1. Legacy ``checkpointer:`` config section (backward compatible) + 2. Unified ``database:`` config section + 3. Default InMemoryStore (emits a WARNING) """ if app_config is None: app_config = get_app_config() - if app_config.checkpointer is None: - from langgraph.store.memory import InMemoryStore - - logger.warning("No 'checkpointer' section in config.yaml — using InMemoryStore for the store. Thread list will be lost on server restart. Configure a sqlite or postgres backend for persistence.") - yield InMemoryStore() + # Legacy: standalone checkpointer config takes precedence + if app_config.checkpointer is not None: + async with _async_store(app_config.checkpointer) as store: + yield store return - async with _async_store(app_config.checkpointer) as store: - yield store + # Unified database config + db_config = getattr(app_config, "database", None) + if db_config is not None and db_config.backend != "memory": + async with _async_store_from_database(db_config) as store: + yield store + return + + # Default: in-memory + from langgraph.store.memory import InMemoryStore + + logger.warning("No persistent store backend configured (no 'checkpointer' or 'database' section) — using InMemoryStore. Cross-thread store data will be lost on server restart. Configure a sqlite or postgres backend for persistence.") + yield InMemoryStore() diff --git a/backend/tests/test_store_provider.py b/backend/tests/test_store_provider.py new file mode 100644 index 00000000..93ecef63 --- /dev/null +++ b/backend/tests/test_store_provider.py @@ -0,0 +1,89 @@ +"""Tests for the async Store factory's backend selection. + +Mirrors the checkpointer factory: when no legacy ``checkpointer`` section +is configured but a unified ``database`` section is, the store must use +that database backend instead of silently falling back to InMemoryStore. +""" + +import sys +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from deerflow.config.database_config import DatabaseConfig +from deerflow.runtime.store.async_provider import make_store + + +@pytest.fixture +def anyio_backend() -> str: + return "asyncio" + + +class TestStoreDatabaseFallback: + @pytest.mark.anyio + async def test_postgres_store_from_database_when_no_checkpointer_section(self): + """make_store uses AsyncPostgresStore from the database section when + no legacy checkpointer section is present. The +asyncpg dialect prefix + is stripped so the same DATABASE_URL satisfies both SQLAlchemy and + LangGraph's psycopg-based store.""" + mock_config = MagicMock() + mock_config.checkpointer = None + mock_config.database = DatabaseConfig( + backend="postgres", + postgres_url="postgresql+asyncpg://postgres:pw@localhost:5432/deerflow", + ) + + mock_store = AsyncMock() + mock_cm = AsyncMock() + mock_cm.__aenter__.return_value = mock_store + mock_cm.__aexit__.return_value = False + + mock_store_cls = MagicMock() + mock_store_cls.from_conn_string.return_value = mock_cm + + mock_module = MagicMock() + mock_module.AsyncPostgresStore = mock_store_cls + + with ( + patch("deerflow.runtime.store.async_provider.get_app_config", return_value=mock_config), + patch.dict(sys.modules, {"langgraph.store.postgres.aio": mock_module}), + ): + async with make_store() as store: + assert store is mock_store + + # dialect prefix stripped to libpq conninfo + mock_store_cls.from_conn_string.assert_called_once_with("postgresql://postgres:pw@localhost:5432/deerflow") + mock_store.setup.assert_awaited_once() + + @pytest.mark.anyio + async def test_memory_when_no_checkpointer_and_no_database(self): + """With neither a checkpointer section nor a non-memory database, + the store falls back to InMemoryStore.""" + from langgraph.store.memory import InMemoryStore + + mock_config = MagicMock() + mock_config.checkpointer = None + mock_config.database = DatabaseConfig(backend="memory") + + with patch("deerflow.runtime.store.async_provider.get_app_config", return_value=mock_config): + async with make_store() as store: + assert isinstance(store, InMemoryStore) + + @pytest.mark.anyio + async def test_checkpointer_section_takes_precedence_over_database(self): + """A legacy checkpointer section still wins over the database section.""" + from deerflow.config.checkpointer_config import CheckpointerConfig + + mock_config = MagicMock() + mock_config.checkpointer = CheckpointerConfig(type="memory") + mock_config.database = DatabaseConfig( + backend="postgres", + postgres_url="postgresql+asyncpg://postgres:pw@localhost:5432/deerflow", + ) + + from langgraph.store.memory import InMemoryStore + + with patch("deerflow.runtime.store.async_provider.get_app_config", return_value=mock_config): + async with make_store() as store: + # checkpointer.type == memory → InMemoryStore, database ignored + assert isinstance(store, InMemoryStore)