This commit is contained in:
吉鑫升 committed 2026-06-08 18:02:14 +08:00
commit fb692a0555
30 files changed
+3613 -1

No files matched your search

@@ -0,0 +1,158 @@
"use client";
import { BookMarked } from "lucide-react";
import { useState } from "react";
import { useNavigate } from "react-router-dom";
import { toast } from "sonner";
import { Button } from "@/components/ui/button";
import {
Dialog,
DialogContent,
DialogDescription,
DialogFooter,
DialogHeader,
DialogTitle,
} from "@/components/ui/dialog";
import { Input } from "@/components/ui/input";
import {
Select,
SelectContent,
SelectItem,
SelectTrigger,
SelectValue,
} from "@/components/ui/select";
import { useKnowledgeActions } from "@/core/knowledge/hooks";
import { useThread } from "./messages/context";
import { Tooltip } from "./tooltip";
export function KnowledgeSaveTrigger({ threadId }: { threadId: string }) {
const { thread } = useThread();
const navigate = useNavigate();
const { ingestThread } = useKnowledgeActions();
const [open, setOpen] = useState(false);
const [title, setTitle] = useState("");
const [tags, setTags] = useState("");
const [status, setStatus] = useState<"approved" | "draft">("approved");
const [mode, setMode] = useState<"summary" | "full">("summary");
const [includeSources, setIncludeSources] = useState(true);
if (thread.messages.length === 0) {
return null;
}
const handleSave = async () => {
try {
const result = await ingestThread.mutateAsync({
threadId,
payload: {
title: title.trim() || null,
tags: tags
.split(/[,,\s]+/)
.map((t) => t.trim())
.filter(Boolean),
status,
mode,
include_sources: includeSources,
},
});
toast.success("已保存到知识库", {
action: {
label: "查看",
onClick: () =>
navigate(`/page/workspace/knowledge/${result.note.id}`),
},
});
setOpen(false);
setTitle("");
setTags("");
} catch (error) {
toast.error(error instanceof Error ? error.message : "保存失败");
}
};
return (
<>
<Tooltip content="保存到知识库">
<Button
className="text-muted-foreground hover:text-foreground"
variant="ghost"
onClick={() => setOpen(true)}
>
<BookMarked />
保存
</Button>
</Tooltip>
<Dialog open={open} onOpenChange={setOpen}>
<DialogContent className="max-w-md">
<DialogHeader>
<DialogTitle>保存到知识库</DialogTitle>
<DialogDescription>
将本次对话沉淀为 obsidian-wiki 兼容的知识笔记。
</DialogDescription>
</DialogHeader>
<div className="flex flex-col gap-3">
<Input
placeholder="标题(留空自动生成)"
value={title}
onChange={(e) => setTitle(e.target.value)}
/>
<Input
placeholder="标签,逗号分隔"
value={tags}
onChange={(e) => setTags(e.target.value)}
/>
<div className="flex items-center gap-2">
<Select
value={status}
onValueChange={(v) => setStatus(v as "approved" | "draft")}
>
<SelectTrigger className="flex-1">
<SelectValue />
</SelectTrigger>
<SelectContent>
<SelectItem value="approved">正式</SelectItem>
<SelectItem value="draft">草稿</SelectItem>
</SelectContent>
</Select>
<Select
value={mode}
onValueChange={(v) => setMode(v as "summary" | "full")}
>
<SelectTrigger className="flex-1">
<SelectValue />
</SelectTrigger>
<SelectContent>
<SelectItem value="summary">摘要</SelectItem>
<SelectItem value="full">完整</SelectItem>
</SelectContent>
</Select>
</div>
<label className="flex items-center gap-2 text-sm">
<input
type="checkbox"
checked={includeSources}
onChange={(e) => setIncludeSources(e.target.checked)}
className="size-4"
/>
包含搜索 / 工具来源
</label>
</div>
<DialogFooter>
<Button variant="outline" onClick={() => setOpen(false)}>
取消
</Button>
<Button
onClick={() => void handleSave()}
disabled={ingestThread.isPending}
>
{ingestThread.isPending ? "保存中..." : "保存"}
</Button>
</DialogFooter>
</DialogContent>
</Dialog>
</>
);
}
+133
View File
@@ -0,0 +1,133 @@
import { apiFetch } from "@/core/api";
import { getBackendBaseURL } from "@/core/config";
import type {
BatchIngestPayload,
BatchIngestResult,
GraphExportResult,
IngestThreadPayload,
KnowledgeListParams,
KnowledgeListResponse,
KnowledgeNote,
KnowledgeSearchItem,
} from "./types";
const baseUrl = () => `${getBackendBaseURL()}/api/knowledge`;
async function parseResponse<T>(response: Response): Promise<T> {
if (!response.ok) {
const error = await response
.json()
.catch(() => ({ detail: "Request failed" }));
const detail = error.detail;
if (typeof detail === "string") throw new Error(detail);
throw new Error("请求失败");
}
return response.json() as Promise<T>;
}
export async function listKnowledgeNotes(params: KnowledgeListParams = {}) {
const qs = new URLSearchParams();
if (params.keyword) qs.set("keyword", params.keyword);
if (params.tag) qs.set("tag", params.tag);
if (params.source_type) qs.set("source_type", params.source_type);
if (params.status) qs.set("status", params.status);
qs.set("limit", String(params.limit ?? 20));
qs.set("offset", String(params.offset ?? 0));
return parseResponse<KnowledgeListResponse>(
await apiFetch(`${baseUrl()}/notes?${qs.toString()}`),
);
}
export async function getKnowledgeNote(noteId: string) {
return parseResponse<KnowledgeNote>(
await apiFetch(`${baseUrl()}/notes/${encodeURIComponent(noteId)}`),
);
}
export async function createKnowledgeNote(data: {
title: string;
content_md: string;
summary?: string | null;
tags?: string[];
status?: string;
}) {
const res = await parseResponse<{ note: KnowledgeNote }>(
await apiFetch(`${baseUrl()}/notes`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(data),
}),
);
return res.note;
}
export async function updateKnowledgeNote(
noteId: string,
data: Partial<{
title: string;
content_md: string;
summary: string | null;
tags: string[];
status: string;
}>,
) {
return parseResponse<KnowledgeNote>(
await apiFetch(`${baseUrl()}/notes/${encodeURIComponent(noteId)}`, {
method: "PUT",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(data),
}),
);
}
export async function deleteKnowledgeNote(noteId: string, hard = false) {
return parseResponse<{ ok: boolean; note_id: string }>(
await apiFetch(
`${baseUrl()}/notes/${encodeURIComponent(noteId)}${hard ? "?hard=true" : ""}`,
{ method: "DELETE" },
),
);
}
export async function searchKnowledge(query: string, limit = 8) {
const res = await parseResponse<{ items: KnowledgeSearchItem[] }>(
await apiFetch(`${baseUrl()}/search`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ query, mode: "keyword", limit }),
}),
);
return res.items;
}
export async function ingestThread(threadId: string, payload: IngestThreadPayload = {}) {
return parseResponse<{ note: KnowledgeNote; created: boolean }>(
await apiFetch(`${baseUrl()}/ingest/thread/${encodeURIComponent(threadId)}`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(payload),
}),
);
}
export async function ingestThreadsBatch(payload: BatchIngestPayload = {}) {
return parseResponse<BatchIngestResult>(
await apiFetch(`${baseUrl()}/ingest/threads/batch`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(payload),
}),
);
}
export async function exportKnowledgeGraph() {
return parseResponse<GraphExportResult>(
await apiFetch(`${baseUrl()}/export/graph`, { method: "POST" }),
);
}
/** Absolute URL of a served wiki-export file (for iframe / download). */
export function knowledgeExportFileUrl(filename: string) {
return `${baseUrl()}/export/files/${encodeURIComponent(filename)}`;
}
+89
View File
@@ -0,0 +1,89 @@
import {
keepPreviousData,
useMutation,
useQuery,
useQueryClient,
} from "@tanstack/react-query";
import {
createKnowledgeNote,
deleteKnowledgeNote,
exportKnowledgeGraph,
getKnowledgeNote,
ingestThread,
ingestThreadsBatch,
listKnowledgeNotes,
searchKnowledge,
updateKnowledgeNote,
} from "./api";
import type { KnowledgeListParams } from "./types";
const ROOT_KEY = ["knowledge"] as const;
export function useKnowledgeNotes(params: KnowledgeListParams = {}) {
return useQuery({
queryKey: [...ROOT_KEY, "list", params],
queryFn: () => listKnowledgeNotes(params),
placeholderData: keepPreviousData,
refetchOnWindowFocus: false,
});
}
export function useKnowledgeNote(noteId: string | undefined) {
return useQuery({
queryKey: [...ROOT_KEY, "note", noteId],
queryFn: () => getKnowledgeNote(noteId as string),
enabled: !!noteId,
refetchOnWindowFocus: false,
});
}
export function useKnowledgeSearch(query: string, enabled: boolean) {
return useQuery({
queryKey: [...ROOT_KEY, "search", query],
queryFn: () => searchKnowledge(query),
enabled: enabled && query.trim().length > 0,
placeholderData: keepPreviousData,
refetchOnWindowFocus: false,
});
}
export function useKnowledgeActions() {
const queryClient = useQueryClient();
const invalidate = () =>
queryClient.invalidateQueries({ queryKey: ROOT_KEY });
return {
create: useMutation({ mutationFn: createKnowledgeNote, onSuccess: invalidate }),
update: useMutation({
mutationFn: ({
noteId,
data,
}: {
noteId: string;
data: Parameters<typeof updateKnowledgeNote>[1];
}) => updateKnowledgeNote(noteId, data),
onSuccess: invalidate,
}),
remove: useMutation({
mutationFn: ({ noteId, hard }: { noteId: string; hard?: boolean }) =>
deleteKnowledgeNote(noteId, hard),
onSuccess: invalidate,
}),
ingestThread: useMutation({
mutationFn: ({
threadId,
payload,
}: {
threadId: string;
payload?: Parameters<typeof ingestThread>[1];
}) => ingestThread(threadId, payload),
onSuccess: invalidate,
}),
batchIngest: useMutation({
mutationFn: ingestThreadsBatch,
onSuccess: invalidate,
}),
exportGraph: useMutation({ mutationFn: exportKnowledgeGraph }),
};
}
+102
View File
@@ -0,0 +1,102 @@
export type KnowledgeStatus = "draft" | "approved" | "archived";
export type KnowledgeSourceType =
| "thread"
| "search"
| "tool"
| "file"
| "manual"
| "report";
export interface KnowledgeSource {
id: string;
note_id: string;
source_type: string;
thread_id?: string | null;
message_id?: string | null;
tool_name?: string | null;
title?: string | null;
url?: string | null;
snippet?: string | null;
created_at?: string | null;
}
export interface KnowledgeNote {
id: string;
title: string;
summary?: string | null;
content_md: string;
vault_path?: string | null;
source_type: string;
source_id?: string | null;
status: KnowledgeStatus;
confidence?: number;
created_by?: string | null;
updated_by?: string | null;
created_at?: string | null;
updated_at?: string | null;
tags: string[];
sources?: KnowledgeSource[];
}
export interface KnowledgeListResponse {
items: KnowledgeNote[];
total: number;
}
export interface KnowledgeListParams {
keyword?: string;
tag?: string;
source_type?: string;
status?: string;
limit?: number;
offset?: number;
}
export interface KnowledgeSearchItem {
note_id: string;
chunk_id?: string | null;
title?: string | null;
summary?: string | null;
snippet?: string | null;
score: number;
tags: string[];
source_type?: string | null;
updated_at?: string | null;
}
export interface IngestThreadPayload {
mode?: "summary" | "full";
title?: string | null;
status?: KnowledgeStatus;
tags?: string[];
include_sources?: boolean;
}
export interface BatchIngestPayload {
limit?: number;
offset?: number;
status?: KnowledgeStatus;
include_sources?: boolean;
skip_existing?: boolean;
mode?: "summary" | "full";
}
export interface BatchIngestResult {
ok: boolean;
processed: number;
created: number;
skipped: number;
failed: number;
items: Array<{ thread_id: string; note_id: string | null; status: string }>;
}
export interface GraphExportResult {
ok: boolean;
files: {
graph_json: string;
graph_html: string;
graph_graphml: string;
cypher: string;
};
stats: { nodes: number; edges: number };
}
@@ -9,6 +9,7 @@
*/
import type { ElementType } from "react";
import {
BookOpen,
Bot,
Brain,
CalendarClock,
@@ -262,6 +263,12 @@ export const sidebarMenuItems: MenuItem[] = [
icon: Network,
path: "/page/strategy/resource-management/resource-network-actor",
},
{
id: "resource-knowledge-native",
label: "知识库",
icon: BookOpen,
path: "/page/workspace/knowledge",
},
{
id: "resource-knowledge-base",
label: "知识库资源管理",
+2
View File
@@ -15,6 +15,7 @@ import {
useThreadChat,
} from "@/components/workspace/chats";
import { ExportTrigger } from "@/components/workspace/export-trigger";
import { KnowledgeSaveTrigger } from "@/components/workspace/knowledge-save-trigger";
import { InputBox } from "@/components/workspace/input-box";
import {
MessageList,
@@ -409,6 +410,7 @@ export default function ChatPage() {
messages={thread.messages}
/>
<ExportTrigger threadId={threadId} />
<KnowledgeSaveTrigger threadId={threadId} />
<ArtifactTrigger />
</div>
</header>
@@ -0,0 +1,597 @@
import {
BookOpen,
ExternalLink,
Loader2,
Network,
Pencil,
Plus,
RefreshCw,
Save,
Search,
Trash2,
X,
} from "lucide-react";
import { useEffect, useMemo, useState } from "react";
import { useNavigate, useParams } from "react-router-dom";
import { toast } from "sonner";
import { Badge } from "@/components/ui/badge";
import { Button } from "@/components/ui/button";
import {
Dialog,
DialogContent,
DialogDescription,
DialogFooter,
DialogHeader,
DialogTitle,
} from "@/components/ui/dialog";
import { Input } from "@/components/ui/input";
import { ScrollArea } from "@/components/ui/scroll-area";
import {
Select,
SelectContent,
SelectItem,
SelectTrigger,
SelectValue,
} from "@/components/ui/select";
import { Textarea } from "@/components/ui/textarea";
import {
useKnowledgeActions,
useKnowledgeNote,
useKnowledgeNotes,
} from "@/core/knowledge/hooks";
import type { KnowledgeNote } from "@/core/knowledge/types";
import { cn } from "@/lib/utils";
import MarkdownRenderer from "@/strategy-components/MarkdownRenderer";
const SOURCE_TYPE_LABELS: Record<string, string> = {
thread: "对话",
search: "检索",
tool: "工具",
file: "文件",
manual: "手动",
report: "报告",
};
const STATUS_LABELS: Record<string, string> = {
draft: "草稿",
approved: "已批准",
archived: "已归档",
};
function useDebounced<T>(value: T, delay = 350): T {
const [debounced, setDebounced] = useState(value);
useEffect(() => {
const id = window.setTimeout(() => setDebounced(value), delay);
return () => window.clearTimeout(id);
}, [value, delay]);
return debounced;
}
function NoteListItem({
note,
active,
onClick,
}: {
note: KnowledgeNote;
active: boolean;
onClick: () => void;
}) {
return (
<button
type="button"
onClick={onClick}
className={cn(
"flex w-full flex-col gap-1 rounded-md border p-3 text-left transition-colors",
active
? "border-primary bg-primary/5"
: "border-border/70 hover:bg-muted/50",
)}
>
<div className="flex items-center justify-between gap-2">
<span className="line-clamp-1 text-sm font-medium">{note.title}</span>
<Badge variant="outline" className="shrink-0 text-[10px]">
{SOURCE_TYPE_LABELS[note.source_type] ?? note.source_type}
</Badge>
</div>
{note.summary ? (
<span className="text-muted-foreground line-clamp-2 text-xs">
{note.summary}
</span>
) : null}
<div className="mt-1 flex flex-wrap items-center gap-1">
{note.tags.slice(0, 4).map((tag) => (
<span
key={tag}
className="bg-muted text-muted-foreground rounded px-1.5 py-0.5 text-[10px]"
>
#{tag}
</span>
))}
<span className="text-muted-foreground ml-auto text-[10px]">
{note.updated_at?.slice(0, 10)}
</span>
</div>
</button>
);
}
function CreateNoteDialog({
open,
onOpenChange,
}: {
open: boolean;
onOpenChange: (v: boolean) => void;
}) {
const { create } = useKnowledgeActions();
const navigate = useNavigate();
const [title, setTitle] = useState("");
const [tags, setTags] = useState("");
const [content, setContent] = useState("");
const reset = () => {
setTitle("");
setTags("");
setContent("");
};
const handleSubmit = async () => {
if (!title.trim()) {
toast.error("请输入标题");
return;
}
try {
const note = await create.mutateAsync({
title: title.trim(),
content_md: content || `# ${title.trim()}\n\n`,
tags: tags
.split(/[,,\s]+/)
.map((t) => t.trim())
.filter(Boolean),
status: "approved",
});
toast.success("知识已创建");
onOpenChange(false);
reset();
navigate(`/page/workspace/knowledge/${note.id}`);
} catch (error) {
toast.error(error instanceof Error ? error.message : "创建失败");
}
};
return (
<Dialog open={open} onOpenChange={onOpenChange}>
<DialogContent className="max-w-2xl">
<DialogHeader>
<DialogTitle>新建知识</DialogTitle>
<DialogDescription>手动创建一条知识笔记,写入共享知识库 vault。</DialogDescription>
</DialogHeader>
<div className="flex flex-col gap-3">
<Input
placeholder="标题"
value={title}
onChange={(e) => setTitle(e.target.value)}
/>
<Input
placeholder="标签,使用逗号分隔"
value={tags}
onChange={(e) => setTags(e.target.value)}
/>
<Textarea
placeholder="Markdown 正文"
rows={12}
value={content}
onChange={(e) => setContent(e.target.value)}
className="font-mono text-sm"
/>
</div>
<DialogFooter>
<Button variant="outline" onClick={() => onOpenChange(false)}>
取消
</Button>
<Button onClick={() => void handleSubmit()} disabled={create.isPending}>
{create.isPending ? "创建中..." : "创建"}
</Button>
</DialogFooter>
</DialogContent>
</Dialog>
);
}
function BatchImportDialog({
open,
onOpenChange,
}: {
open: boolean;
onOpenChange: (v: boolean) => void;
}) {
const { batchIngest } = useKnowledgeActions();
const [limit, setLimit] = useState(50);
const [skipExisting, setSkipExisting] = useState(true);
const handleRun = async () => {
try {
const result = await batchIngest.mutateAsync({
limit,
skip_existing: skipExisting,
status: "approved",
include_sources: true,
});
toast.success(
`批量导入完成:新增 ${result.created},跳过 ${result.skipped},失败 ${result.failed}`,
);
onOpenChange(false);
} catch (error) {
toast.error(error instanceof Error ? error.message : "批量导入失败");
}
};
return (
<Dialog open={open} onOpenChange={onOpenChange}>
<DialogContent className="max-w-md">
<DialogHeader>
<DialogTitle>批量导入历史线程</DialogTitle>
<DialogDescription>
扫描你最近的对话线程并沉淀为知识库内容。已沉淀的线程可自动跳过。
</DialogDescription>
</DialogHeader>
<div className="flex flex-col gap-4">
<label className="flex items-center justify-between gap-2 text-sm">
处理数量上限
<Input
type="number"
min={1}
max={500}
value={limit}
onChange={(e) => setLimit(Number(e.target.value) || 1)}
className="w-28"
/>
</label>
<label className="flex items-center justify-between gap-2 text-sm">
跳过已沉淀线程
<input
type="checkbox"
checked={skipExisting}
onChange={(e) => setSkipExisting(e.target.checked)}
className="size-4"
/>
</label>
</div>
<DialogFooter>
<Button variant="outline" onClick={() => onOpenChange(false)}>
取消
</Button>
<Button onClick={() => void handleRun()} disabled={batchIngest.isPending}>
{batchIngest.isPending ? "导入中..." : "开始导入"}
</Button>
</DialogFooter>
</DialogContent>
</Dialog>
);
}
function NoteDetail({ noteId }: { noteId: string }) {
const navigate = useNavigate();
const { data: note, isLoading } = useKnowledgeNote(noteId);
const { update, remove } = useKnowledgeActions();
const [editing, setEditing] = useState(false);
const [draftTitle, setDraftTitle] = useState("");
const [draftTags, setDraftTags] = useState("");
const [draftContent, setDraftContent] = useState("");
const [draftStatus, setDraftStatus] = useState("approved");
useEffect(() => {
setEditing(false);
}, [noteId]);
useEffect(() => {
if (note) {
setDraftTitle(note.title);
setDraftTags(note.tags.join(", "));
setDraftContent(note.content_md);
setDraftStatus(note.status);
}
}, [note]);
if (isLoading || !note) {
return (
<div className="text-muted-foreground flex h-full items-center justify-center text-sm">
{isLoading ? "加载中..." : "知识不存在"}
</div>
);
}
const handleSave = async () => {
try {
await update.mutateAsync({
noteId: note.id,
data: {
title: draftTitle.trim() || note.title,
content_md: draftContent,
tags: draftTags
.split(/[,,\s]+/)
.map((t) => t.trim())
.filter(Boolean),
status: draftStatus,
},
});
toast.success("已保存");
setEditing(false);
} catch (error) {
toast.error(error instanceof Error ? error.message : "保存失败");
}
};
const handleArchive = async () => {
try {
await remove.mutateAsync({ noteId: note.id });
toast.success("已归档");
navigate("/page/workspace/knowledge");
} catch (error) {
toast.error(error instanceof Error ? error.message : "归档失败");
}
};
return (
<div className="flex h-full flex-col">
<div className="flex items-center justify-between gap-2 border-b p-3">
<div className="flex items-center gap-2">
<Badge variant="outline">
{SOURCE_TYPE_LABELS[note.source_type] ?? note.source_type}
</Badge>
<Badge variant="secondary">{STATUS_LABELS[note.status] ?? note.status}</Badge>
</div>
<div className="flex items-center gap-2">
{editing ? (
<>
<Button size="sm" variant="ghost" onClick={() => setEditing(false)}>
<X className="size-4" /> 取消
</Button>
<Button size="sm" onClick={() => void handleSave()} disabled={update.isPending}>
<Save className="size-4" /> 保存
</Button>
</>
) : (
<>
<Button size="sm" variant="ghost" onClick={() => setEditing(true)}>
<Pencil className="size-4" /> 编辑
</Button>
<Button
size="sm"
variant="ghost"
className="text-destructive"
onClick={() => void handleArchive()}
disabled={remove.isPending}
>
<Trash2 className="size-4" /> 归档
</Button>
</>
)}
</div>
</div>
<ScrollArea className="min-h-0 flex-1">
<div className="p-4">
{editing ? (
<div className="flex flex-col gap-3">
<Input
value={draftTitle}
onChange={(e) => setDraftTitle(e.target.value)}
placeholder="标题"
/>
<div className="flex items-center gap-2">
<Input
value={draftTags}
onChange={(e) => setDraftTags(e.target.value)}
placeholder="标签,逗号分隔"
/>
<Select value={draftStatus} onValueChange={setDraftStatus}>
<SelectTrigger className="w-32">
<SelectValue />
</SelectTrigger>
<SelectContent>
<SelectItem value="draft">草稿</SelectItem>
<SelectItem value="approved">已批准</SelectItem>
<SelectItem value="archived">已归档</SelectItem>
</SelectContent>
</Select>
</div>
<Textarea
rows={20}
value={draftContent}
onChange={(e) => setDraftContent(e.target.value)}
className="font-mono text-sm"
/>
</div>
) : (
<>
<h1 className="mb-2 text-xl font-semibold">{note.title}</h1>
<div className="mb-3 flex flex-wrap gap-1">
{note.tags.map((tag) => (
<span
key={tag}
className="bg-muted text-muted-foreground rounded px-1.5 py-0.5 text-xs"
>
#{tag}
</span>
))}
</div>
<MarkdownRenderer content={note.content_md} />
{note.sources && note.sources.length > 0 ? (
<div className="mt-6 border-t pt-4">
<h2 className="mb-2 text-sm font-semibold">来源</h2>
<ul className="flex flex-col gap-2">
{note.sources.map((src) => (
<li key={src.id} className="text-sm">
{src.url ? (
<a
href={src.url}
target="_blank"
rel="noreferrer"
className="text-primary inline-flex items-center gap-1 hover:underline"
>
{src.title || src.url}
<ExternalLink className="size-3" />
</a>
) : src.thread_id ? (
<button
type="button"
className="text-primary hover:underline"
onClick={() =>
navigate(`/page/workspace/chats/${src.thread_id}`)
}
>
{src.title || `对话 ${src.thread_id.slice(0, 8)}`}
</button>
) : (
<span>{src.title || src.tool_name || "来源"}</span>
)}
{src.snippet ? (
<p className="text-muted-foreground mt-0.5 line-clamp-2 text-xs">
{src.snippet}
</p>
) : null}
</li>
))}
</ul>
</div>
) : null}
<div className="text-muted-foreground mt-6 border-t pt-3 text-xs">
创建人 {note.created_by ?? "—"} · 更新人 {note.updated_by ?? "—"} · 更新于{" "}
{note.updated_at?.slice(0, 19).replace("T", " ")}
</div>
</>
)}
</div>
</ScrollArea>
</div>
);
}
export default function KnowledgeBasePage() {
const navigate = useNavigate();
const { noteId } = useParams<{ noteId: string }>();
const [keyword, setKeyword] = useState("");
const debouncedKeyword = useDebounced(keyword);
const [sourceType, setSourceType] = useState("all");
const [status, setStatus] = useState("all");
const [createOpen, setCreateOpen] = useState(false);
const [batchOpen, setBatchOpen] = useState(false);
const params = useMemo(
() => ({
keyword: debouncedKeyword || undefined,
source_type: sourceType === "all" ? undefined : sourceType,
status: status === "all" ? undefined : status,
limit: 50,
}),
[debouncedKeyword, sourceType, status],
);
const { data, isLoading } = useKnowledgeNotes(params);
const notes = data?.items ?? [];
return (
<div className="flex size-full min-h-0 flex-col">
<header className="flex flex-wrap items-center gap-2 border-b p-3">
<BookOpen className="size-5" />
<h1 className="mr-2 text-base font-semibold">知识库</h1>
<div className="relative">
<Search className="text-muted-foreground absolute left-2 top-1/2 size-4 -translate-y-1/2" />
<Input
value={keyword}
onChange={(e) => setKeyword(e.target.value)}
placeholder="搜索标题 / 摘要 / 正文 / 标签"
className="w-64 pl-8"
/>
</div>
<Select value={sourceType} onValueChange={setSourceType}>
<SelectTrigger className="w-28">
<SelectValue placeholder="来源" />
</SelectTrigger>
<SelectContent>
<SelectItem value="all">全部来源</SelectItem>
{Object.entries(SOURCE_TYPE_LABELS).map(([v, label]) => (
<SelectItem key={v} value={v}>
{label}
</SelectItem>
))}
</SelectContent>
</Select>
<Select value={status} onValueChange={setStatus}>
<SelectTrigger className="w-28">
<SelectValue placeholder="状态" />
</SelectTrigger>
<SelectContent>
<SelectItem value="all">全部状态</SelectItem>
<SelectItem value="draft">草稿</SelectItem>
<SelectItem value="approved">已批准</SelectItem>
<SelectItem value="archived">已归档</SelectItem>
</SelectContent>
</Select>
<div className="ml-auto flex items-center gap-2">
<Button variant="outline" size="sm" onClick={() => setBatchOpen(true)}>
<RefreshCw className="size-4" /> 批量导入
</Button>
<Button
variant="outline"
size="sm"
onClick={() => navigate("/page/workspace/knowledge/graph")}
>
<Network className="size-4" /> 图谱
</Button>
<Button size="sm" onClick={() => setCreateOpen(true)}>
<Plus className="size-4" /> 新建
</Button>
</div>
</header>
<div className="grid min-h-0 flex-1 grid-cols-1 lg:grid-cols-[360px_minmax(0,1fr)]">
<div className="flex min-h-0 flex-col border-r">
<ScrollArea className="min-h-0 flex-1">
<div className="flex flex-col gap-2 p-3">
{isLoading ? (
<div className="text-muted-foreground flex items-center gap-2 p-4 text-sm">
<Loader2 className="size-4 animate-spin" /> 加载中...
</div>
) : notes.length === 0 ? (
<div className="text-muted-foreground rounded-md border border-dashed p-6 text-center text-sm">
暂无知识。点击「新建」或在对话页「保存到知识库」。
</div>
) : (
notes.map((note) => (
<NoteListItem
key={note.id}
note={note}
active={note.id === noteId}
onClick={() => navigate(`/page/workspace/knowledge/${note.id}`)}
/>
))
)}
</div>
</ScrollArea>
{data ? (
<div className="text-muted-foreground border-t p-2 text-center text-xs">
共 {data.total} 条知识
</div>
) : null}
</div>
<div className="min-h-0">
{noteId ? (
<NoteDetail noteId={noteId} />
) : (
<div className="text-muted-foreground flex h-full items-center justify-center text-sm">
从左侧选择一条知识查看详情
</div>
)}
</div>
</div>
<CreateNoteDialog open={createOpen} onOpenChange={setCreateOpen} />
<BatchImportDialog open={batchOpen} onOpenChange={setBatchOpen} />
</div>
);
}
@@ -0,0 +1,79 @@
import { ArrowLeft, Loader2, RefreshCw } from "lucide-react";
import { useCallback, useEffect, useState } from "react";
import { useNavigate } from "react-router-dom";
import { toast } from "sonner";
import { Button } from "@/components/ui/button";
import { knowledgeExportFileUrl } from "@/core/knowledge/api";
import { useKnowledgeActions } from "@/core/knowledge/hooks";
export default function KnowledgeGraphPage() {
const navigate = useNavigate();
const { exportGraph } = useKnowledgeActions();
const [src, setSrc] = useState<string | null>(null);
const [stats, setStats] = useState<{ nodes: number; edges: number } | null>(null);
const regenerate = useCallback(async () => {
try {
const result = await exportGraph.mutateAsync();
setStats(result.stats);
// Cache-bust the iframe so it reflects the freshly generated graph.
setSrc(`${knowledgeExportFileUrl("graph.html")}?t=${Date.now()}`);
} catch (error) {
toast.error(error instanceof Error ? error.message : "生成图谱失败");
}
// eslint-disable-next-line react-hooks/exhaustive-deps
}, []);
useEffect(() => {
void regenerate();
}, [regenerate]);
return (
<div className="flex size-full min-h-0 flex-col">
<header className="flex items-center gap-2 border-b p-3">
<Button
variant="ghost"
size="sm"
onClick={() => navigate("/page/workspace/knowledge")}
>
<ArrowLeft className="size-4" /> 返回
</Button>
<h1 className="text-base font-semibold">知识图谱</h1>
{stats ? (
<span className="text-muted-foreground text-xs">
{stats.nodes} 个节点 · {stats.edges} 条关系
</span>
) : null}
<Button
variant="outline"
size="sm"
className="ml-auto"
onClick={() => void regenerate()}
disabled={exportGraph.isPending}
>
{exportGraph.isPending ? (
<Loader2 className="size-4 animate-spin" />
) : (
<RefreshCw className="size-4" />
)}
重新生成
</Button>
</header>
<div className="min-h-0 flex-1">
{src ? (
<iframe
src={src}
title="知识图谱"
className="size-full border-0"
sandbox="allow-scripts allow-same-origin"
/>
) : (
<div className="text-muted-foreground flex h-full items-center justify-center text-sm">
<Loader2 className="mr-2 size-4 animate-spin" /> 生成图谱中...
</div>
)}
</div>
</div>
);
}
@@ -24,6 +24,8 @@ import AdminScheduledTasksPage from './AdminScheduledTasksPage';
import AdminSkillsPage from './AdminSkillsPage';
import ScheduledTaskRunDetailPage from './ScheduledTaskRunDetailPage';
import ScheduledTasksPage from './ScheduledTasksPage';
import KnowledgeBasePage from './KnowledgeBasePage';
import KnowledgeGraphPage from './KnowledgeGraphPage';
import LlmMetricsPage from './LlmMetricsPage';
import CheckpointMigrationPage from './CheckpointMigrationPage';
import MemoryPage from './MemoryPage';
@@ -335,6 +337,30 @@ export default function WorkspaceRoutes({
</WorkspaceLayout>
}
/>
<Route
path="knowledge"
element={
<WorkspaceLayout {...layoutProps}>
<KnowledgeBasePage />
</WorkspaceLayout>
}
/>
<Route
path="knowledge/graph"
element={
<WorkspaceLayout {...layoutProps}>
<KnowledgeGraphPage />
</WorkspaceLayout>
}
/>
<Route
path="knowledge/:noteId"
element={
<WorkspaceLayout {...layoutProps}>
<KnowledgeBasePage />
</WorkspaceLayout>
}
/>
<Route
path="scheduled"
element={<Navigate to="/page/workspace/scheduled/tasks" replace />}
+11 -1
View File
@@ -250,7 +250,7 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA
| **Scheduled Tasks** (`/api/scheduled-tasks`) | Per-user cron-like agent tasks. `GET/POST /` - list/create owned tasks; `PATCH/DELETE /{id}` - update/delete (owner); `POST /{id}/{pause,resume,trigger}`; `GET /{id}/runs`, `GET /runs/{rid}`, run-file preview/edit/download; `GET /runs/{rid}/messages` - agent messages for the run's underlying agent run (used to inspect a stuck/running run; authorized via `_load_run`); `GET /runs/{rid}/html` - generated HTML document for an `html_page` run; `DELETE /runs/{rid}` - owner-scoped, with an admin fallback so admins can delete any user's run; publish/subscribe under `/{id}/{publish,unpublish,subscribe}` and `GET /published`. **Admin (cross-user):** `GET /admin/all` - every task; `PATCH/DELETE /admin/{id}` - edit/delete any task; `POST /admin/{id}/trigger`; `GET /admin/{id}/runs`. Admin routes require `system_role == "admin"`. |
| **HTML Page Favorites** (`/api/html-page-favorites`) | Bookmarked generated HTML pages, owner-scoped. `GET /` - list; `POST /` `{page_id, title?, description?, tags?}` - bookmark a `scheduled_html_pages` row (extracts a reusable style/layout summary via GLM5 at favorite time); a page can only be favorited once per user (`DuplicateFavoriteError` → HTTP 409). `GET/DELETE /{id}` - read/delete (the create-task reference picker deletes favorites here). A favorite can be selected as a style reference when creating a new `html_page` task. |
| **Public Scheduled Tasks** (`/api/public/scheduled-tasks`) | **Unauthenticated** mirror for embedding. Adds `GET /runs/{rid}/html` and `GET /{task_id}/latest-html` for the `html_page` preview; the run result keeps `result.html` (only file paths are redacted). |
| **Public Leaderboard** (`/api/public/leaderboard`) | **Unauthenticated** read-only user leaderboard snapshot powering the public publish page (`#/public/leaderboard`). `GET /` returns the same `AdminLeaderboardResponse` as the admin endpoint, reusing `_load_or_schedule_daily_leaderboard` and honoring the system-wide leaderboard settings (excluded users, scheduled/admin/failed inclusion). No admin controls, no mutations. Router: `app/gateway/routers/public_leaderboard.py`. |
| **Knowledge** (`/api/knowledge`) | Product knowledge base, **global/shared** (no user isolation; writes record `created_by`/`updated_by` for audit). `POST /ingest/thread/{thread_id}` - sediment one thread into a note (reads messages from the checkpointer, `threads:read` owner-checked); `POST /ingest/threads/batch` - batch-sediment the caller's recent threads (`skip_existing` via `.manifest.json` dedup); `POST /ingest/search` - sediment search/tool results as a reference note; `POST /notes` - manual note; `GET /notes` (keyword/tag/source_type/status + pagination); `GET/PUT/DELETE /notes/{id}` (DELETE is soft = archive, `?hard=true` removes the vault file); `POST /search` - keyword/frontmatter/Markdown content search (vector/hybrid fall back to keyword in v1); `POST /export/graph` + `GET /export/files/{graph.json\|graph.html\|graph.graphml\|cypher.txt}` - obsidian-wiki `wiki-export` artifacts (self-contained `graph.html` for iframe). Backed by the `deerflow.knowledge` module (see below). |
Proxied through nginx: `/api/langgraph/*` → LangGraph, all other `/api/*` → Gateway.
@@ -274,6 +274,16 @@ Proxied through nginx: `/api/langgraph/*` → LangGraph, all other `/api/*` →
**HTML skills (tools the render step uses).** Two offline skills under `skills/public/` shape the generated HTML and are toggled in the skills settings page (enabled by default): `html-page-builder` (a self-contained inline-CSS design system — colors/cards/tables/KPI/responsive) and `html-charts` (inline-SVG charts that render in the no-`allow-scripts` preview iframe; optional local ECharts documented in its `assets/README.md`). A skill opts into HTML injection via `metadata.kind: html_page` in its SKILL.md frontmatter; `_collect_html_skill_instructions()` reads the bodies of **enabled** such skills and `build_generation_prompt(skill_instructions=…)` injects them, so the (raw-model) render step honors the user's skill selection. The data-gathering step and the render/repair/style passes all use the task's selected `model_name` (no silent fallback to the default model — and there is **no hardcoded GLM-5**; an unset selection falls back to config `models[0]`). To stop auxiliary LLM calls from leaking onto the default model, **`TitleMiddleware` is skipped when `is_scheduled_run` is set** (`_build_middlewares` in `lead_agent/agent.py`) — scheduled/HTML runs use throwaway threads whose titles are never shown, and `config.title.model_name` is usually unset (→ would call `models[0]`, not the chosen model). The completeness gate (`validate_html`) also requires a non-empty `<title>` and forbids external stylesheets, on top of the doctype/html/head/body + non-empty-body + no-script checks.
### Knowledge Base (`packages/harness/deerflow/knowledge/`)
Self-contained product knowledge base. **All knowledge capability lives in this one module** (harness layer, `deerflow.*` only — no `app.*` imports). Wired as `app.state.knowledge_service` in `deps.py` via `make_knowledge_service(sf, config.knowledge)` (returns `None` on the memory backend or when `knowledge.enabled=false` → router 503). Surfaced by `app/gateway/routers/knowledge.py` (`/api/knowledge`). Config: `config.yaml` → `knowledge` (`deerflow/config/knowledge_config.py`).
**Two stores, kept in sync on every write:**
- **DB mirror** — tables `knowledge_notes` / `knowledge_sources` / `knowledge_tags` / `knowledge_note_tags` (`knowledge/models.py`, registered in `persistence/models/__init__.py`, auto-created by `Base.metadata.create_all`; portable types `PortableLongText`/`BeijingDateTime`). Powers listing/filter/search/audit. Global/shared — no `user_id` column; `created_by`/`updated_by` are audit-only (resolved via `get_effective_user_id()`).
- **Markdown vault** — obsidian-wiki compatible, default `{$DEER_FLOW_HOME}/data/knowledge/obsidian-vault`. `ObsidianWikiAdapter` (`obsidian_adapter.py`) writes wiki-capture frontmatter+body, and after **every** write syncs `index.md` / `log.md` (CAPTURE/INGEST/UPDATE/ARCHIVE/EXPORT) / `hot.md` / `.manifest.json` (source-hash dedup for `skip_existing`). `markdown_vault.py` handles frontmatter build/parse, slugging, wikilink extraction and **path-traversal-proof** resolution.
**Module layout:** `models.py` (ORM) · `repository.py` (DB only, no LLM) · `schemas.py` (NoteDraft/SourceDraft DTOs) · `skill_templates.py` (wiki-capture/query/export constants) · `markdown_vault.py` · `manifest.py` · `extractor.py` (rule-based thread/search → NoteDraft; turn splitting, greeting filter, 5-section declarative body, ≥2 wikilinks; LLM extractor is a future drop-in) · `sources.py` (reference-batch / search-result → SourceDraft) · `exporter.py` (`graph.json`/`graph.graphml`/`cypher.txt` + self-contained vanilla-JS `graph.html`, no CDN) · `search.py` (keyword/frontmatter/content scoring; vector/hybrid reserved for phase 2) · `obsidian_adapter.py` · `service.py` (`KnowledgeService` orchestrator — the only thing the router touches). Tests: `tests/test_knowledge.py`. Frontend: `frontend-web/src/core/knowledge/*` + `pages/KnowledgeBasePage.tsx` / `KnowledgeGraphPage.tsx` + chat `knowledge-save-trigger.tsx`; spec at `frontend-web/docs/product-knowledge-base-dev.md`.
### Custom Agent Persistence
Custom agent metadata lives in the `agents` ORM table (`deerflow.persistence.agents`). `agents.id` is the stable runtime id and the filesystem directory name under `.deer-flow/agents/{id}/`; `agents.name` is display-only, can contain Chinese characters, and is not unique. `user_id IS NULL` marks built-in agents, user-owned agents are visible only to their owner unless `published = true`, and update/delete operations require ownership. The runtime accepts `agent_id` in context/configurable, and non-default `assistant_id` values are treated as agent ids after visibility checks.
@@ -38,6 +38,7 @@ from app.gateway.routers import (
feedback,
html_page_favorites,
intent,
knowledge,
llm_metrics,
mcp,
memory,
@@ -1252,6 +1253,9 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
app.include_router(public_embed.router)
app.include_router(html_page_favorites.router)
# Product knowledge base API is mounted at /api/knowledge
app.include_router(knowledge.router)
# System settings API is mounted at /api/system-settings
app.include_router(system_settings.router)
@@ -159,6 +159,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
app.state.memory_config_store = make_user_memory_config_store(sf)
from deerflow.knowledge import make_knowledge_service
app.state.knowledge_service = make_knowledge_service(sf, config.knowledge)
from deerflow.persistence.user_prompts import make_user_prompt_store
app.state.user_prompt_store = make_user_prompt_store(sf)
@@ -372,6 +376,14 @@ def get_memory_config_store(request: Request):
return val
def get_knowledge_service(request: Request):
"""Return the knowledge base service (503 when disabled / memory backend)."""
val = getattr(request.app.state, "knowledge_service", None)
if val is None:
raise HTTPException(status_code=503, detail="Knowledge base not available")
return val
def get_run_context(request: Request) -> RunContext:
"""Build a :class:`RunContext` from ``app.state`` singletons.
@@ -0,0 +1,337 @@
"""Knowledge base API (``/api/knowledge``).
Sediments threads / searches into an obsidian-wiki compatible vault, mirrors
metadata into the database, and serves listing, detail, edit, delete, search
and graph export. The knowledge base is **global / shared** — no per-user
isolation; writes record ``created_by`` / ``updated_by`` for audit.
All obsidian-wiki specifics live in :mod:`deerflow.knowledge`; this router only
adapts HTTP ↔ the :class:`~deerflow.knowledge.service.KnowledgeService`.
"""
from __future__ import annotations
import logging
import mimetypes
from typing import Any
from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field
from app.gateway.authz import require_auth, require_permission
from app.gateway.deps import get_checkpointer, get_knowledge_service, get_thread_store
from app.gateway.utils import sanitize_log_param
from deerflow.knowledge.service import EmptyThreadError
from deerflow.runtime import serialize_channel_values
from deerflow.runtime.references import build_reference_batches
from deerflow.runtime.user_context import get_effective_user_id
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/knowledge", tags=["knowledge"])
# --------------------------------------------------------------------------- #
# Request / response models
# --------------------------------------------------------------------------- #
class IngestThreadRequest(BaseModel):
mode: str = Field(default="summary", description="summary | full")
title: str | None = None
status: str = Field(default="approved", description="draft | approved")
tags: list[str] = Field(default_factory=list)
include_sources: bool = True
class BatchIngestRequest(BaseModel):
limit: int = Field(default=100, ge=1, le=500)
offset: int = Field(default=0, ge=0)
status: str = "approved"
include_sources: bool = True
skip_existing: bool = True
mode: str = "summary"
class SearchResultItem(BaseModel):
title: str | None = None
url: str | None = None
snippet: str | None = None
tool_name: str | None = None
score: float | None = None
class IngestSearchRequest(BaseModel):
query: str
results: list[SearchResultItem] = Field(default_factory=list)
thread_id: str | None = None
message_id: str | None = None
title: str | None = None
tags: list[str] = Field(default_factory=list)
status: str = "approved"
class CreateNoteRequest(BaseModel):
title: str
content_md: str
summary: str | None = None
tags: list[str] = Field(default_factory=list)
status: str = "approved"
class UpdateNoteRequest(BaseModel):
title: str | None = None
content_md: str | None = None
summary: str | None = None
tags: list[str] | None = None
status: str | None = None
class SearchRequest(BaseModel):
query: str
mode: str = Field(default="keyword", description="keyword | vector | hybrid (vector/hybrid fall back to keyword in v1)")
limit: int = Field(default=8, ge=1, le=50)
# --------------------------------------------------------------------------- #
# Helpers
# --------------------------------------------------------------------------- #
async def _read_thread_values(checkpointer, thread_id: str) -> dict[str, Any]:
"""Return serialized channel values for a thread, or raise 404/500."""
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
try:
checkpoint_tuple = await checkpointer.aget_tuple(config)
except Exception:
logger.exception("Failed to read thread %s for knowledge ingest", sanitize_log_param(thread_id))
raise HTTPException(status_code=500, detail="Failed to read thread state")
if checkpoint_tuple is None:
raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found")
checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {}
return serialize_channel_values(checkpoint.get("channel_values", {}) or {})
def _messages_and_meta(values: dict[str, Any]) -> tuple[list[dict[str, Any]], str | None, list[dict[str, Any]]]:
messages = values.get("messages")
if not isinstance(messages, list):
messages = []
title = values.get("title") if isinstance(values.get("title"), str) else None
batches = build_reference_batches(messages) if messages else []
return messages, title, batches
# --------------------------------------------------------------------------- #
# Ingest
# --------------------------------------------------------------------------- #
@router.post("/ingest/thread/{thread_id}")
@require_auth
@require_permission("threads", "read", owner_check=True)
async def ingest_thread(thread_id: str, body: IngestThreadRequest, request: Request) -> dict[str, Any]:
"""Sediment a single thread into a knowledge note."""
service = get_knowledge_service(request)
checkpointer = get_checkpointer(request)
values = await _read_thread_values(checkpointer, thread_id)
messages, title, batches = _messages_and_meta(values)
if not messages:
raise HTTPException(status_code=400, detail="Thread has no messages to sediment")
try:
result = await service.capture_thread(
thread_id=thread_id,
messages=messages,
thread_title=title,
mode=body.mode,
title=body.title,
status=body.status,
tags=body.tags,
include_sources=body.include_sources,
reference_batches=batches,
created_by=get_effective_user_id(),
)
except EmptyThreadError:
raise HTTPException(status_code=400, detail="Thread has no sediment-able content")
return {"note": result["note"], "created": result["created"]}
@router.post("/ingest/threads/batch")
@require_auth
async def ingest_threads_batch(body: BatchIngestRequest, request: Request) -> dict[str, Any]:
"""Batch-sediment the caller's recent threads into knowledge notes."""
service = get_knowledge_service(request)
checkpointer = get_checkpointer(request)
thread_store = get_thread_store(request)
threads = await thread_store.search(limit=body.limit, offset=body.offset)
created = skipped = failed = 0
items: list[dict[str, Any]] = []
for meta in threads:
thread_id = meta.get("thread_id")
if not thread_id:
continue
try:
values = await _read_thread_values(checkpointer, thread_id)
messages, title, batches = _messages_and_meta(values)
if not messages:
skipped += 1
items.append({"thread_id": thread_id, "note_id": None, "status": "skipped"})
continue
result = await service.capture_thread(
thread_id=thread_id,
messages=messages,
thread_title=title or meta.get("display_name"),
mode=body.mode,
status=body.status,
include_sources=body.include_sources,
reference_batches=batches,
created_by=get_effective_user_id(),
skip_existing=body.skip_existing,
)
if result["created"]:
created += 1
items.append({"thread_id": thread_id, "note_id": result["note"]["id"], "status": "created"})
else:
skipped += 1
items.append({"thread_id": thread_id, "note_id": result["note"]["id"], "status": "skipped"})
except EmptyThreadError:
skipped += 1
items.append({"thread_id": thread_id, "note_id": None, "status": "skipped"})
except HTTPException:
failed += 1
items.append({"thread_id": thread_id, "note_id": None, "status": "failed"})
except Exception:
logger.exception("Batch ingest failed for thread %s", sanitize_log_param(str(thread_id)))
failed += 1
items.append({"thread_id": thread_id, "note_id": None, "status": "failed"})
return {"ok": True, "processed": len(items), "created": created, "skipped": skipped, "failed": failed, "items": items}
@router.post("/ingest/search")
@require_auth
async def ingest_search(body: IngestSearchRequest, request: Request) -> dict[str, Any]:
"""Sediment search/tool results as a reference note."""
service = get_knowledge_service(request)
note = await service.capture_search(
query=body.query,
results=[r.model_dump() for r in body.results],
thread_id=body.thread_id,
message_id=body.message_id,
title=body.title,
tags=body.tags,
status=body.status,
created_by=get_effective_user_id(),
)
return {"note": note}
# --------------------------------------------------------------------------- #
# CRUD
# --------------------------------------------------------------------------- #
@router.post("/notes")
@require_auth
async def create_note(body: CreateNoteRequest, request: Request) -> dict[str, Any]:
service = get_knowledge_service(request)
note = await service.create_manual_note(
title=body.title,
content_md=body.content_md,
summary=body.summary,
tags=body.tags,
status=body.status,
created_by=get_effective_user_id(),
)
return {"note": note}
@router.get("/notes")
@require_auth
async def list_notes(
request: Request,
keyword: str | None = None,
tag: str | None = None,
source_type: str | None = None,
status: str | None = None,
limit: int = 20,
offset: int = 0,
) -> dict[str, Any]:
service = get_knowledge_service(request)
limit = max(1, min(limit, 100))
offset = max(0, offset)
return await service.list_notes(keyword=keyword, tag=tag, source_type=source_type, status=status, limit=limit, offset=offset)
@router.get("/notes/{note_id}")
@require_auth
async def get_note(note_id: str, request: Request) -> dict[str, Any]:
service = get_knowledge_service(request)
note = await service.get_note(note_id)
if note is None:
raise HTTPException(status_code=404, detail="Knowledge note not found")
return note
@router.put("/notes/{note_id}")
@require_auth
async def update_note(note_id: str, body: UpdateNoteRequest, request: Request) -> dict[str, Any]:
service = get_knowledge_service(request)
note = await service.update_note(
note_id,
title=body.title,
content_md=body.content_md,
summary=body.summary,
tags=body.tags,
status=body.status,
updated_by=get_effective_user_id(),
)
if note is None:
raise HTTPException(status_code=404, detail="Knowledge note not found")
return note
@router.delete("/notes/{note_id}")
@require_auth
async def delete_note(note_id: str, request: Request, hard: bool = False) -> dict[str, Any]:
service = get_knowledge_service(request)
ok = await service.delete_note(note_id, updated_by=get_effective_user_id(), hard=hard)
if not ok:
raise HTTPException(status_code=404, detail="Knowledge note not found")
return {"ok": True, "note_id": note_id}
# --------------------------------------------------------------------------- #
# Search & graph
# --------------------------------------------------------------------------- #
@router.post("/search")
@require_auth
async def search(body: SearchRequest, request: Request) -> dict[str, Any]:
service = get_knowledge_service(request)
items = await service.search(body.query, mode=body.mode, limit=body.limit)
return {"items": items}
@router.post("/export/graph")
@require_auth
async def export_graph(request: Request) -> dict[str, Any]:
service = get_knowledge_service(request)
result = await service.export_graph()
base = "/api/knowledge/export/files"
return {
"ok": True,
"files": {
"graph_json": f"{base}/graph.json",
"graph_html": f"{base}/graph.html",
"graph_graphml": f"{base}/graph.graphml",
"cypher": f"{base}/cypher.txt",
},
"stats": result["stats"],
}
@router.get("/export/files/{filename}")
@require_auth
async def export_file(filename: str, request: Request) -> FileResponse:
service = get_knowledge_service(request)
path = service.export_file_path(filename)
if path is None:
# Lazily build the export on first access so the iframe never 404s.
await service.export_graph()
path = service.export_file_path(filename)
if path is None:
raise HTTPException(status_code=404, detail="Export file not found")
media_type = mimetypes.guess_type(filename)[0] or "application/octet-stream"
return FileResponse(path, media_type=media_type)
@@ -342,3 +342,31 @@ log_system:
enabled: true
url: "https://your-log-system.example.com"
label: "日志系统"
# 产品内知识库 —— obsidian-wiki 兼容 vault。知识元数据写入数据库,正文
# content_md 写入 Markdown vault(index.md/log.md/hot.md/.manifest.json/frontmatter/
# wikilinks/wiki-export 图谱产物)。知识库为全局共享,不做 user_id 隔离,
# 写入时记录 created_by / updated_by 作为审计。
knowledge:
enabled: true
provider: obsidian_wiki
vault_dir: data/knowledge/obsidian-vault
project_name: deerflow
auto_ingest_enabled: false
auto_ingest_min_chars: 300
max_thread_messages: 80
max_source_items: 20
rag_enabled: false
rag_limit: 5
export:
enabled: true
graph_html_enabled: true
graph_json_enabled: true
query:
use_hot_md: true
use_index_md: true
use_frontmatter_scan: true
embedding:
enabled: false
provider: local
model: ""
@@ -16,6 +16,7 @@ from deerflow.config.database_config import DatabaseConfig
from deerflow.config.extensions_config import ExtensionsConfig
from deerflow.config.gateway_config import GatewayConfig
from deerflow.config.guardrails_config import GuardrailsConfig, load_guardrails_config_from_dict
from deerflow.config.knowledge_config import KnowledgeConfig
from deerflow.config.log_system_config import LogSystemConfig
from deerflow.config.login_config import AuthLoginConfig
from deerflow.config.memory_config import MemoryConfig, load_memory_config_from_dict
@@ -135,6 +136,7 @@ class AppConfig(BaseModel):
checkpointer: CheckpointerConfig | None = Field(default=None, description="Checkpointer configuration")
stream_bridge: StreamBridgeConfig | None = Field(default=None, description="Stream bridge configuration")
browser_context: BrowserContextConfig = Field(default_factory=BrowserContextConfig, description="Optional browser page capture integration")
knowledge: KnowledgeConfig = Field(default_factory=KnowledgeConfig, description="Product knowledge base (obsidian-wiki compatible vault) configuration")
@classmethod
def resolve_config_path(cls, config_path: str | None = None) -> Path:
@@ -0,0 +1,55 @@
"""Configuration for the product knowledge base (obsidian-wiki compatible)."""
from pydantic import BaseModel, Field
class KnowledgeExportConfig(BaseModel):
"""Knowledge graph export switches (``wiki-export`` compatible output)."""
enabled: bool = Field(default=True, description="Master switch for graph export.")
graph_html_enabled: bool = Field(default=True, description="Generate wiki-export/graph.html.")
graph_json_enabled: bool = Field(default=True, description="Generate wiki-export/graph.json.")
class KnowledgeQueryConfig(BaseModel):
"""Layered query strategy flags (``wiki-query`` style)."""
use_hot_md: bool = Field(default=True, description="Consult hot.md when searching.")
use_index_md: bool = Field(default=True, description="Consult index.md when searching.")
use_frontmatter_scan: bool = Field(default=True, description="Scan frontmatter when searching.")
class KnowledgeEmbeddingConfig(BaseModel):
"""Vector embedding config (phase 2 — disabled in the first release)."""
enabled: bool = Field(default=False, description="Enable vector embeddings.")
provider: str = Field(default="local", description="Embedding provider.")
model: str = Field(default="", description="Embedding model name.")
class KnowledgeConfig(BaseModel):
"""Product knowledge base configuration.
The first release ships a global, shared, obsidian-wiki compatible vault.
Knowledge metadata is mirrored into the database for fast listing/search,
while ``content_md`` is written to a Markdown vault that stays compatible
with ``ar9av/obsidian-wiki`` (index.md / log.md / hot.md / .manifest.json,
frontmatter, wikilinks, wiki-export graph artifacts).
"""
enabled: bool = Field(default=True, description="Enable the knowledge base feature.")
provider: str = Field(default="obsidian_wiki", description="Knowledge organization provider.")
vault_dir: str = Field(
default="data/knowledge/obsidian-vault",
description="Vault directory (relative to DEER_FLOW_HOME, or absolute).",
)
project_name: str = Field(default="deerflow", description="Project namespace for projects/<name>/ pages.")
auto_ingest_enabled: bool = Field(default=False, description="Auto-sediment conversations in the background (phase 3).")
auto_ingest_min_chars: int = Field(default=300, description="Minimum characters before auto-ingest considers a turn.")
max_thread_messages: int = Field(default=80, description="Maximum thread messages to scan when capturing.")
max_source_items: int = Field(default=20, description="Maximum source items recorded per note.")
rag_enabled: bool = Field(default=False, description="Enable RAG injection before agent runs (phase 2).")
rag_limit: int = Field(default=5, description="Maximum knowledge snippets injected per query.")
export: KnowledgeExportConfig = Field(default_factory=KnowledgeExportConfig, description="Graph export configuration.")
query: KnowledgeQueryConfig = Field(default_factory=KnowledgeQueryConfig, description="Layered query configuration.")
embedding: KnowledgeEmbeddingConfig = Field(default_factory=KnowledgeEmbeddingConfig, description="Embedding configuration.")
@@ -0,0 +1,41 @@
"""Product knowledge base module (obsidian-wiki compatible).
Self-contained: knowledge ingestion, storage (DB mirror + Markdown vault),
search and graph export all live here. The Gateway only depends on
:class:`KnowledgeService` via :func:`make_knowledge_service`.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from deerflow.knowledge.obsidian_adapter import ObsidianWikiAdapter
from deerflow.knowledge.repository import KnowledgeRepository
from deerflow.knowledge.service import EmptyThreadError, KnowledgeService
if TYPE_CHECKING:
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.config.knowledge_config import KnowledgeConfig
__all__ = [
"EmptyThreadError",
"KnowledgeRepository",
"KnowledgeService",
"ObsidianWikiAdapter",
"make_knowledge_service",
]
def make_knowledge_service(
session_factory: async_sessionmaker[AsyncSession] | None,
config: KnowledgeConfig,
) -> KnowledgeService | None:
"""Build the knowledge service.
Returns ``None`` when persistence is memory-only (no session factory) or
the feature is disabled — the router then responds 503.
"""
if session_factory is None or not config.enabled:
return None
return KnowledgeService(KnowledgeRepository(session_factory), config)
@@ -0,0 +1,190 @@
"""obsidian-wiki ``wiki-export`` compatible graph generation.
Builds a knowledge graph from the notes (nodes) plus their wikilinks and shared
tags (edges) and writes the four native artifacts into ``<vault>/wiki-export/``:
``graph.json``, ``graph.graphml``, ``cypher.txt`` and a self-contained
``graph.html`` (no external CDN, safe to iframe offline).
"""
from __future__ import annotations
import html
import json
from pathlib import Path
from typing import Any
from deerflow.knowledge.markdown_vault import extract_wikilinks
def _vault_key(vault_path: str | None) -> str | None:
if not vault_path:
return None
key = vault_path.replace("\\", "/")
if key.endswith(".md"):
key = key[:-3]
return key
def build_graph(notes: list[dict[str, Any]]) -> dict[str, Any]:
"""Return ``{nodes, edges}`` from note dicts (id, title, tags, vault_path, content_md)."""
nodes = []
by_key: dict[str, str] = {}
for n in notes:
nodes.append({"id": n["id"], "type": "note", "label": n.get("title") or n["id"], "category": n.get("source_type")})
key = _vault_key(n.get("vault_path"))
if key:
by_key[key] = n["id"]
edges: list[dict[str, Any]] = []
edge_seen: set[tuple[str, str, str]] = set()
def add_edge(a: str, b: str, label: str, weight: float) -> None:
if a == b:
return
key = (min(a, b), max(a, b), label)
if key in edge_seen:
return
edge_seen.add(key)
edges.append({"id": f"rel_{len(edges) + 1}", "source": a, "target": b, "label": label, "weight": weight})
# Wikilink edges.
for n in notes:
for target in extract_wikilinks(n.get("content_md") or ""):
tid = by_key.get(target.replace("\\", "/").removesuffix(".md"))
if tid:
add_edge(n["id"], tid, "links", 2)
# Shared-tag edges.
tag_to_notes: dict[str, list[str]] = {}
for n in notes:
for tag in n.get("tags") or []:
tag_to_notes.setdefault(tag.lower(), []).append(n["id"])
for tag, ids in tag_to_notes.items():
if len(ids) < 2 or len(ids) > 50: # skip ubiquitous tags (e.g. knowledge-base)
continue
for i in range(len(ids)):
for j in range(i + 1, len(ids)):
add_edge(ids[i], ids[j], "related", 1)
return {"nodes": nodes, "edges": edges}
def to_graphml(graph: dict[str, Any]) -> str:
lines = [
'<?xml version="1.0" encoding="UTF-8"?>',
'<graphml xmlns="http://graphml.graphdrawing.org/xmlns">',
' <key id="label" for="node" attr.name="label" attr.type="string"/>',
' <key id="elabel" for="edge" attr.name="label" attr.type="string"/>',
' <graph edgedefault="undirected">',
]
for n in graph["nodes"]:
lines.append(f' <node id="{html.escape(str(n["id"]))}"><data key="label">{html.escape(str(n["label"]))}</data></node>')
for e in graph["edges"]:
lines.append(
f' <edge id="{html.escape(str(e["id"]))}" source="{html.escape(str(e["source"]))}" '
f'target="{html.escape(str(e["target"]))}"><data key="elabel">{html.escape(str(e["label"]))}</data></edge>'
)
lines += [" </graph>", "</graphml>"]
return "\n".join(lines)
def to_cypher(graph: dict[str, Any]) -> str:
lines = []
for n in graph["nodes"]:
label = str(n["label"]).replace("'", "\\'")
lines.append(f"CREATE (:Note {{id:'{n['id']}', title:'{label}'}});")
for e in graph["edges"]:
rel = str(e["label"]).upper()
lines.append(f"MATCH (a:Note {{id:'{e['source']}'}}),(b:Note {{id:'{e['target']}'}}) CREATE (a)-[:{rel} {{weight:{e['weight']}}}]->(b);")
return "\n".join(lines)
def to_html(graph: dict[str, Any]) -> str:
"""Self-contained force-directed graph (canvas + vanilla JS, no CDN)."""
data_json = json.dumps(graph, ensure_ascii=False)
return _HTML_TEMPLATE.replace("__GRAPH_DATA__", data_json)
def export_all(
vault_root: Path | str,
notes: list[dict[str, Any]],
*,
write_html: bool = True,
write_json: bool = True,
) -> dict[str, Any]:
"""Generate graph artifacts under ``<vault>/wiki-export/`` and return stats."""
out_dir = Path(vault_root).resolve() / "wiki-export"
out_dir.mkdir(parents=True, exist_ok=True)
graph = build_graph(notes)
written: list[str] = []
if write_json:
(out_dir / "graph.json").write_text(json.dumps(graph, ensure_ascii=False, indent=2), encoding="utf-8")
written.append("graph.json")
(out_dir / "graph.graphml").write_text(to_graphml(graph), encoding="utf-8")
written.append("graph.graphml")
(out_dir / "cypher.txt").write_text(to_cypher(graph), encoding="utf-8")
written.append("cypher.txt")
if write_html:
(out_dir / "graph.html").write_text(to_html(graph), encoding="utf-8")
written.append("graph.html")
return {
"files": written,
"stats": {"nodes": len(graph["nodes"]), "edges": len(graph["edges"])},
"graph": graph,
}
_HTML_TEMPLATE = """<!DOCTYPE html>
<html lang="zh">
<head>
<meta charset="utf-8"/>
<meta name="viewport" content="width=device-width, initial-scale=1"/>
<title>Knowledge Graph</title>
<style>
html,body{margin:0;height:100%;background:#0b1020;color:#e7ecff;font-family:system-ui,Segoe UI,Arial,sans-serif;overflow:hidden}
#hud{position:fixed;top:10px;left:12px;font-size:12px;opacity:.8;z-index:2}
#tip{position:fixed;padding:4px 8px;background:#111a36;border:1px solid #2a3a66;border-radius:4px;font-size:12px;pointer-events:none;display:none;z-index:3}
canvas{display:block}
#empty{position:fixed;inset:0;display:flex;align-items:center;justify-content:center;opacity:.6}
</style>
</head>
<body>
<div id="hud"></div>
<div id="tip"></div>
<canvas id="c"></canvas>
<script>
const GRAPH = __GRAPH_DATA__;
const canvas = document.getElementById('c'), ctx = canvas.getContext('2d');
const hud = document.getElementById('hud'), tip = document.getElementById('tip');
let W=0,H=0;
function resize(){W=canvas.width=innerWidth;H=canvas.height=innerHeight;}
addEventListener('resize',resize);resize();
hud.textContent = GRAPH.nodes.length + ' 个知识节点 · ' + GRAPH.edges.length + ' 条关系';
if(!GRAPH.nodes.length){document.body.insertAdjacentHTML('beforeend','<div id=\"empty\">暂无知识图谱数据</div>');}
const idx={};GRAPH.nodes.forEach((n,i)=>{idx[n.id]=n;n.x=W/2+Math.cos(i)*200*Math.random()+(Math.random()-.5)*200;n.y=H/2+Math.sin(i)*200*Math.random()+(Math.random()-.5)*200;n.vx=0;n.vy=0;});
const edges=GRAPH.edges.map(e=>({s:idx[e.source],t:idx[e.target],w:e.w||e.weight||1})).filter(e=>e.s&&e.t);
let cam={x:0,y:0,k:1},drag=null,pan=null;
function step(){
for(const n of GRAPH.nodes){for(const m of GRAPH.nodes){if(n===m)continue;let dx=n.x-m.x,dy=n.y-m.y,d2=dx*dx+dy*dy+0.01,f=1200/d2;n.vx+=dx*f*0.0006;n.vy+=dy*f*0.0006;}}
for(const e of edges){let dx=e.t.x-e.s.x,dy=e.t.y-e.s.y,d=Math.sqrt(dx*dx+dy*dy)||1,f=(d-110)*0.01;e.s.vx+=dx/d*f;e.s.vy+=dy/d*f;e.t.vx-=dx/d*f;e.t.vy-=dy/d*f;}
for(const n of GRAPH.nodes){n.vx+=(W/2-n.x)*0.0008;n.vy+=(H/2-n.y)*0.0008;n.x+=n.vx*=0.85;n.y+=n.vy*=0.85;}
}
function draw(){
ctx.setTransform(cam.k,0,0,cam.k,cam.x,cam.y);ctx.clearRect(-cam.x/cam.k,-cam.y/cam.k,W/cam.k,H/cam.k);
ctx.strokeStyle='rgba(120,150,255,.25)';
for(const e of edges){ctx.lineWidth=e.w*0.6;ctx.beginPath();ctx.moveTo(e.s.x,e.s.y);ctx.lineTo(e.t.x,e.t.y);ctx.stroke();}
for(const n of GRAPH.nodes){const r=6;ctx.fillStyle='#5b8cff';ctx.beginPath();ctx.arc(n.x,n.y,r,0,7);ctx.fill();
ctx.fillStyle='rgba(231,236,255,.85)';ctx.font='11px system-ui';ctx.fillText((n.label||'').slice(0,18),n.x+9,n.y+3);}
}
function loop(){step();draw();requestAnimationFrame(loop);}loop();
function toWorld(x,y){return {x:(x-cam.x)/cam.k,y:(y-cam.y)/cam.k};}
canvas.addEventListener('mousedown',ev=>{const p=toWorld(ev.clientX,ev.clientY);let hit=null;
for(const n of GRAPH.nodes){if((n.x-p.x)**2+(n.y-p.y)**2<100){hit=n;break;}}if(hit)drag=hit;else pan={x:ev.clientX,y:ev.clientY,cx:cam.x,cy:cam.y};});
addEventListener('mousemove',ev=>{const p=toWorld(ev.clientX,ev.clientY);if(drag){drag.x=p.x;drag.y=p.y;drag.vx=drag.vy=0;}else if(pan){cam.x=pan.cx+ev.clientX-pan.x;cam.y=pan.cy+ev.clientY-pan.y;}
let hov=null;for(const n of GRAPH.nodes){if((n.x-p.x)**2+(n.y-p.y)**2<100){hov=n;break;}}
if(hov){tip.style.display='block';tip.style.left=(ev.clientX+12)+'px';tip.style.top=(ev.clientY+12)+'px';tip.textContent=hov.label;}else{tip.style.display='none';}});
addEventListener('mouseup',()=>{drag=null;pan=null;});
canvas.addEventListener('wheel',ev=>{ev.preventDefault();const s=ev.deltaY<0?1.1:0.9;cam.k*=s;cam.x=ev.clientX-(ev.clientX-cam.x)*s;cam.y=ev.clientY-(ev.clientY-cam.y)*s;},{passive:false});
</script>
</body>
</html>"""
@@ -0,0 +1,250 @@
"""Extract knowledge notes from threads / searches / tool results.
First-release strategy is rule-based (wiki-capture aligned, §11 of the dev doc):
keep high-value content (user questions, final assistant answers, cited
sources), drop noise (empty/greeting/intermediate reasoning), and write a
*declarative* note rather than a chat transcript. An LLM extractor can later
replace :func:`extract_thread` while keeping the same :class:`NoteDraft` output.
"""
from __future__ import annotations
import re
from typing import Any
from deerflow.knowledge.schemas import NoteDraft, SourceDraft
# Greeting / low-value lines filtered from extracted turns.
_GREETING_RE = re.compile(r"^(你好|您好|hi|hello|hey|thanks?|谢谢|好的|ok|okay)[\s!!。.,,~]*$", re.IGNORECASE)
_WS_RE = re.compile(r"\s+")
def _one_line(text: str, limit: int = 300) -> str:
"""Collapse whitespace and clip ``text`` to a single line."""
collapsed = _WS_RE.sub(" ", text or "").strip()
return collapsed[:limit]
def message_text(msg: dict[str, Any]) -> str:
"""Return the plain-text content of a serialized LangChain message."""
content = msg.get("content")
if isinstance(content, str):
return content.strip()
if isinstance(content, list):
parts: list[str] = []
for block in content:
if isinstance(block, str):
parts.append(block)
elif isinstance(block, dict):
if block.get("type") in (None, "text") and isinstance(block.get("text"), str):
parts.append(block["text"])
return "\n".join(p for p in parts if p).strip()
return ""
def _msg_type(msg: dict[str, Any]) -> str:
return str(msg.get("type") or msg.get("role") or "").lower()
def split_turns(messages: list[dict[str, Any]], *, max_messages: int = 80) -> list[dict[str, str]]:
"""Split messages into ``{user, ai}`` turns keyed by human messages.
For each turn we keep the last substantial AI message (with text body).
"""
msgs = messages[-max_messages:] if max_messages and len(messages) > max_messages else list(messages)
turns: list[dict[str, str]] = []
current: dict[str, str] | None = None
for msg in msgs:
mtype = _msg_type(msg)
text = message_text(msg)
if mtype == "human":
if current and (current.get("user") or current.get("ai")):
turns.append(current)
current = {"user": text, "ai": ""}
elif mtype == "ai" and text:
if current is None:
current = {"user": "", "ai": text}
else:
current["ai"] = text # last substantial AI message wins
if current and (current.get("user") or current.get("ai")):
turns.append(current)
# Drop turns that are pure greetings with no answer.
cleaned = []
for t in turns:
user = (t.get("user") or "").strip()
ai = (t.get("ai") or "").strip()
if not ai and _GREETING_RE.match(user):
continue
if not user and not ai:
continue
cleaned.append({"user": user, "ai": ai})
return cleaned
def _auto_title(thread_title: str | None, turns: list[dict[str, str]]) -> str:
if thread_title and thread_title.strip() and thread_title.strip().lower() != "untitled":
return thread_title.strip()[:200]
for t in turns:
if t.get("user"):
first = t["user"].splitlines()[0].strip()
return (first[:40] + ("…" if len(first) > 40 else "")) or "知识笔记"
return "知识笔记"
def _summary(turns: list[dict[str, str]], *, limit: int = 280) -> str:
"""Simple rule-based summary: the first answer, clipped."""
for t in turns:
if t.get("ai"):
text = re.sub(r"\s+", " ", t["ai"]).strip()
return text[:limit] + ("…" if len(text) > limit else "")
for t in turns:
if t.get("user"):
text = re.sub(r"\s+", " ", t["user"]).strip()
return text[:limit] + ("…" if len(text) > limit else "")
return ""
def related_links(project_name: str | None) -> list[str]:
"""At least two wikilinks per note (wiki-capture requirement)."""
links = ["[[index]]", "[[_meta/taxonomy]]"]
if project_name:
links.append(f"[[projects/{project_name}/synthesis/product-knowledge-base]]")
return links
# Backwards-compatible internal alias.
_related_links = related_links
def _build_body(
*,
title: str,
turns: list[dict[str, str]],
mode: str,
project_name: str | None,
sources: list[SourceDraft],
) -> str:
"""Render the wiki-capture 5-section declarative body."""
full = mode == "full"
answer_clip = 4000 if full else 800
context_lines = []
for t in turns:
if t.get("user"):
q = re.sub(r"\s+", " ", t["user"]).strip()
context_lines.append(f"- {q[:300]}")
context = "\n".join(context_lines[:8]) or "- 来自一次对话沉淀。^[inferred]"
finding_lines = []
for i, t in enumerate(turns, 1):
if t.get("ai"):
ans = t["ai"].strip()
ans = ans if full else (ans[:answer_clip] + ("…" if len(ans) > answer_clip else ""))
finding_lines.append(f"### 结论 {i}\n\n{ans}")
finding = "\n\n".join(finding_lines) or "_暂无可提炼的结论。_"
reasoning = "本页由对话自动沉淀而成,结论来自助手最终回答;过程性推理已省略。判断与归纳部分标记为 ^[inferred]。"
implications = "- 后续可在相关问答中参考本页知识。\n- 如内容过期或冲突,请编辑或归档。"
related = "\n".join(f"- {link}" for link in _related_links(project_name))
if sources:
related += "\n\n### Sources\n\n"
for s in sources[:20]:
label = s.title or s.url or "来源"
related += f"- {label}" + (f" — {s.url}" if s.url else "") + "\n"
return (
f"# {title}\n\n"
f"## Context\n\n{context}\n\n"
f"## Finding / Decision\n\n{finding}\n\n"
f"## Reasoning\n\n{reasoning}\n\n"
f"## Implications\n\n{implications}\n\n"
f"## Related\n\n{related}\n"
)
def extract_thread(
messages: list[dict[str, Any]],
*,
thread_id: str,
thread_title: str | None = None,
mode: str = "summary",
title: str | None = None,
tags: list[str] | None = None,
project_name: str | None = "deerflow",
sources: list[SourceDraft] | None = None,
max_messages: int = 80,
) -> NoteDraft | None:
"""Build a :class:`NoteDraft` from thread messages, or ``None`` if empty."""
turns = split_turns(messages, max_messages=max_messages)
if not turns:
return None
final_title = (title or "").strip() or _auto_title(thread_title, turns)
summary = _summary(turns)
sources = sources or []
body = _build_body(title=final_title, turns=turns, mode=mode, project_name=project_name, sources=sources)
base_tags = list(tags or [])
for default in (project_name or "deerflow", "knowledge-base"):
if default and default not in base_tags:
base_tags.append(default)
return NoteDraft(
title=final_title,
summary=summary,
content_md=body,
category="synthesis",
source_type="thread",
source_id=thread_id,
tags=base_tags,
confidence=0.6,
related=_related_links(project_name),
sources=sources,
source_key=f"thread:{thread_id}",
)
def extract_search(
query: str,
sources: list[SourceDraft],
*,
thread_id: str | None = None,
project_name: str | None = "deerflow",
title: str | None = None,
tags: list[str] | None = None,
) -> NoteDraft:
"""Build a reference-category :class:`NoteDraft` from search/tool results."""
final_title = (title or "").strip() or f"检索:{query[:40]}"
body_lines = [f"# {final_title}\n", "## Context\n", f"- 检索关键词:{query}\n", "## Finding / Decision\n"]
for s in sources:
label = s.title or s.url or "来源"
line = f"- **{label}**"
if s.url:
line += f" — {s.url}"
if s.snippet:
line += "\n - " + _one_line(s.snippet)
body_lines.append(line)
body_lines += [
"\n## Reasoning\n",
"本页汇总外部检索来源,供后续问答引用。来源可信度需结合原文判断。^[ambiguous]\n",
"## Implications\n",
"- 引用前请核对原始来源。\n",
"## Related\n",
*[f"- {link}" for link in _related_links(project_name)],
]
base_tags = list(tags or [])
for default in (project_name or "deerflow", "knowledge-base", "reference"):
if default and default not in base_tags:
base_tags.append(default)
return NoteDraft(
title=final_title,
summary=f"检索“{query}”的来源汇总,共 {len(sources)} 条。",
content_md="\n".join(body_lines),
category="references",
source_type="search",
source_id=f"search:{query[:120]}",
tags=base_tags,
confidence=0.4,
related=_related_links(project_name),
sources=sources,
source_key=None,
)
@@ -0,0 +1,74 @@
""".manifest.json read/write for the vault.
The manifest records, per page, the source key, content hash, vault path and
update time — so batch ingest can skip already-sedimented threads (delta/dedup)
exactly like obsidian-wiki's native ``.manifest.json``.
"""
from __future__ import annotations
import hashlib
import json
from pathlib import Path
from typing import Any
def content_hash(text: str) -> str:
"""Stable sha256 hex digest of ``text`` (used for dedup)."""
return hashlib.sha256((text or "").encode("utf-8")).hexdigest()
class Manifest:
"""JSON manifest stored at ``<vault>/.manifest.json``."""
def __init__(self, root: Path | str) -> None:
self.path = Path(root).resolve() / ".manifest.json"
def load(self) -> dict[str, Any]:
if not self.path.exists():
return {"version": 1, "updated": None, "pages": {}, "sources": {}}
try:
data = json.loads(self.path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
return {"version": 1, "updated": None, "pages": {}, "sources": {}}
data.setdefault("pages", {})
data.setdefault("sources", {})
return data
def save(self, data: dict[str, Any]) -> None:
self.path.parent.mkdir(parents=True, exist_ok=True)
self.path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
def record_page(
self,
*,
note_id: str,
vault_path: str,
source_key: str | None,
hash_value: str,
updated: str,
) -> None:
data = self.load()
data["pages"][note_id] = {
"vault_path": vault_path,
"source": source_key,
"hash": hash_value,
"updated": updated,
}
if source_key:
data["sources"][source_key] = {"note_id": note_id, "hash": hash_value, "updated": updated}
data["updated"] = updated
self.save(data)
def remove_page(self, note_id: str) -> None:
data = self.load()
page = data["pages"].pop(note_id, None)
if page and page.get("source"):
data["sources"].pop(page["source"], None)
self.save(data)
def has_source(self, source_key: str) -> bool:
return source_key in self.load().get("sources", {})
def get_source(self, source_key: str) -> dict[str, Any] | None:
return self.load().get("sources", {}).get(source_key)
@@ -0,0 +1,196 @@
"""obsidian-wiki compatible Markdown vault writer/reader.
Responsibilities (dev doc §10 ``markdown_vault.py``):
- Generate obsidian-wiki compatible frontmatter.
- Produce safe, path-traversal-proof file names/paths.
- Write/update/read Markdown note files.
- Parse frontmatter + extract ``[[wikilinks]]``.
The vault layout follows §8 of the dev doc and stays compatible with
``ar9av/obsidian-wiki`` (concepts/ references/ synthesis/ journal/ projects/…,
plus index.md / log.md / hot.md / .manifest.json / _meta/taxonomy.md /
wiki-export/).
"""
from __future__ import annotations
import re
import unicodedata
from datetime import datetime
from pathlib import Path
from typing import Any
import yaml
_WIKILINK_RE = re.compile(r"\[\[([^\]|]+)(?:\|[^\]]+)?\]\]")
_FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n?(.*)$", re.DOTALL)
_UNSAFE_CHARS_RE = re.compile(r"[^\w一-鿿-]+")
# Top-level vault directories that must exist (obsidian-wiki native layout).
_VAULT_DIRS = ("concepts", "entities", "skills", "references", "synthesis", "journal", "projects", "_meta", "wiki-export")
def slugify(title: str, *, max_len: int = 60) -> str:
"""Return a filesystem-safe slug (ASCII + CJK), never empty."""
text = unicodedata.normalize("NFKC", (title or "").strip()).lower()
text = text.replace(" ", "-")
text = _UNSAFE_CHARS_RE.sub("-", text).strip("-")
text = re.sub(r"-{2,}", "-", text)
if len(text) > max_len:
text = text[:max_len].rstrip("-")
return text or "note"
def extract_wikilinks(text: str) -> list[str]:
"""Return the targets of every ``[[wikilink]]`` in ``text`` (de-duplicated)."""
seen: list[str] = []
for match in _WIKILINK_RE.finditer(text or ""):
target = match.group(1).strip()
if target and target not in seen:
seen.append(target)
return seen
class VaultPathError(ValueError):
"""Raised when a requested vault path escapes the vault root."""
class MarkdownVault:
"""Reads/writes Markdown files inside a single vault root directory."""
def __init__(self, root: Path | str) -> None:
self.root = Path(root).resolve()
# -- path safety ------------------------------------------------------
def resolve(self, rel_path: str) -> Path:
"""Resolve ``rel_path`` under the vault root, blocking traversal."""
rel = str(rel_path).replace("\\", "/").lstrip("/")
candidate = (self.root / rel).resolve()
try:
candidate.relative_to(self.root)
except ValueError as exc:
raise VaultPathError(f"Path escapes vault root: {rel_path!r}") from exc
return candidate
def note_relpath(self, *, category: str, project_name: str | None, slug: str, note_id: str) -> str:
"""Build a vault-relative path for a note.
Project knowledge goes under ``projects/<project>/<category>/`` per
wiki-capture; a short id suffix guarantees uniqueness.
"""
from deerflow.knowledge.skill_templates import category_to_dir
cat_dir = category_to_dir(category)
safe_slug = slugify(slug)
suffix = note_id.split("_")[-1][:8]
filename = f"{safe_slug}-{suffix}.md"
if project_name:
return f"projects/{slugify(project_name)}/{cat_dir}/{filename}"
return f"{cat_dir}/{filename}"
# -- scaffold ---------------------------------------------------------
def ensure_scaffold(self, *, project_name: str | None = None) -> None:
"""Create the vault directory tree + base files if missing."""
self.root.mkdir(parents=True, exist_ok=True)
for d in _VAULT_DIRS:
(self.root / d).mkdir(parents=True, exist_ok=True)
if project_name:
base = self.root / "projects" / slugify(project_name)
for d in ("concepts", "references", "synthesis", "journal"):
(base / d).mkdir(parents=True, exist_ok=True)
# Base vault files.
index = self.root / "index.md"
if not index.exists():
index.write_text("# Knowledge Index\n\n_页面索引。每次写入新页面后自动更新。_\n\n", encoding="utf-8")
log = self.root / "log.md"
if not log.exists():
log.write_text("# Activity Log\n\n_CAPTURE / INGEST / QUERY / EXPORT 记录。_\n\n", encoding="utf-8")
hot = self.root / "hot.md"
if not hot.exists():
hot.write_text("# Hot\n\n_近期活动摘要。_\n\n", encoding="utf-8")
taxonomy = self.root / "_meta" / "taxonomy.md"
if not taxonomy.exists():
taxonomy.write_text(
"# Taxonomy\n\n- synthesis — 多步骤分析、方案、结论\n- concepts — 概念、框架、模型\n"
"- references — 外部来源、搜索结果、文章资料\n- decision — 架构或设计决策\n- journal — 完整会话摘要\n",
encoding="utf-8",
)
# -- frontmatter ------------------------------------------------------
@staticmethod
def build_frontmatter(
*,
note_id: str,
title: str,
category: str,
tags: list[str],
sources: list[str],
summary: str,
created: str,
updated: str,
base_confidence: float,
lifecycle: str,
provenance: dict[str, float] | None = None,
relationships: list[dict[str, str]] | None = None,
) -> dict[str, Any]:
"""Build an obsidian-wiki / wiki-capture compatible frontmatter dict."""
created_date = (created or "")[:10]
return {
"id": note_id,
"title": title,
"category": category,
"tags": list(tags or []),
"sources": list(sources or []),
"created": created,
"updated": updated,
"summary": summary or "",
"provenance": provenance or {"extracted": 0.7, "inferred": 0.2, "ambiguous": 0.1},
"base_confidence": round(float(base_confidence or 0.0), 2),
"lifecycle": lifecycle,
"lifecycle_changed": created_date,
"relationships": relationships or [],
}
# -- IO ---------------------------------------------------------------
def write_note(self, rel_path: str, frontmatter: dict[str, Any], body: str) -> Path:
"""Write ``frontmatter`` + ``body`` to ``rel_path`` (creates dirs)."""
path = self.resolve(rel_path)
path.parent.mkdir(parents=True, exist_ok=True)
fm = yaml.safe_dump(frontmatter, allow_unicode=True, sort_keys=False, default_flow_style=False)
path.write_text(f"---\n{fm}---\n\n{body.strip()}\n", encoding="utf-8")
return path
def read_note(self, rel_path: str) -> tuple[dict[str, Any], str] | None:
"""Return ``(frontmatter, body)`` for a note, or ``None`` if missing."""
path = self.resolve(rel_path)
if not path.exists():
return None
return self.parse(path.read_text(encoding="utf-8"))
@staticmethod
def parse(text: str) -> tuple[dict[str, Any], str]:
"""Split raw Markdown into ``(frontmatter dict, body)``."""
match = _FRONTMATTER_RE.match(text or "")
if not match:
return {}, text or ""
try:
fm = yaml.safe_load(match.group(1)) or {}
except yaml.YAMLError:
fm = {}
if not isinstance(fm, dict):
fm = {}
return fm, match.group(2) or ""
def delete_note(self, rel_path: str) -> None:
"""Remove a note file if present (used on hard delete; archive keeps it)."""
path = self.resolve(rel_path)
if path.exists():
path.unlink()
def now_iso_with_tz(dt: datetime) -> str:
"""ISO-8601 string with timezone offset (frontmatter ``created``/``updated``)."""
if dt.tzinfo is None:
return dt.isoformat()
return dt.isoformat(timespec="seconds")
@@ -0,0 +1,79 @@
"""ORM models for the product knowledge base.
Tables mirror the obsidian-wiki vault into the database so the product can
list, filter, search and audit notes quickly without re-parsing Markdown. The
vault Markdown remains the source-of-truth for ``content_md`` / frontmatter /
graph; these rows are an index + audit layer.
The knowledge base is **global / shared** — there is no ``user_id`` isolation.
Writes record ``created_by`` / ``updated_by`` for auditing only.
"""
from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import Float, ForeignKey, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime, PortableLongText
class KnowledgeNoteRow(Base):
"""Knowledge note main table (mirrors ``knowledge_notes``)."""
__tablename__ = "knowledge_notes"
id: Mapped[str] = mapped_column(String(64), primary_key=True)
title: Mapped[str] = mapped_column(String(512))
summary: Mapped[str | None] = mapped_column(Text)
content_md: Mapped[str] = mapped_column(PortableLongText())
vault_path: Mapped[str | None] = mapped_column(String(1024), index=True)
source_type: Mapped[str] = mapped_column(String(64), index=True)
source_id: Mapped[str | None] = mapped_column(String(256), index=True)
status: Mapped[str] = mapped_column(String(32), default="approved", index=True)
confidence: Mapped[float] = mapped_column(Float, default=0.0)
created_by: Mapped[str | None] = mapped_column(String(128), index=True)
updated_by: Mapped[str | None] = mapped_column(String(128))
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), index=True)
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC), index=True)
class KnowledgeSourceRow(Base):
"""Knowledge source table (mirrors ``knowledge_sources``)."""
__tablename__ = "knowledge_sources"
id: Mapped[str] = mapped_column(String(64), primary_key=True)
note_id: Mapped[str] = mapped_column(String(64), ForeignKey("knowledge_notes.id", ondelete="CASCADE"), index=True)
source_type: Mapped[str] = mapped_column(String(64))
thread_id: Mapped[str | None] = mapped_column(String(128), index=True)
message_id: Mapped[str | None] = mapped_column(String(128))
tool_name: Mapped[str | None] = mapped_column(String(128))
title: Mapped[str | None] = mapped_column(String(512))
url: Mapped[str | None] = mapped_column(Text)
snippet: Mapped[str | None] = mapped_column(Text)
raw_json: Mapped[str | None] = mapped_column(PortableLongText())
created_by: Mapped[str | None] = mapped_column(String(128))
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
class KnowledgeTagRow(Base):
"""Tag table (mirrors ``knowledge_tags``)."""
__tablename__ = "knowledge_tags"
id: Mapped[str] = mapped_column(String(64), primary_key=True)
name: Mapped[str] = mapped_column(String(128), unique=True, index=True)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
class KnowledgeNoteTagRow(Base):
"""Note ↔ tag join table (mirrors ``knowledge_note_tags``)."""
__tablename__ = "knowledge_note_tags"
__table_args__ = (UniqueConstraint("note_id", "tag_id", name="uq_knowledge_note_tags"),)
note_id: Mapped[str] = mapped_column(String(64), ForeignKey("knowledge_notes.id", ondelete="CASCADE"), primary_key=True)
tag_id: Mapped[str] = mapped_column(String(64), ForeignKey("knowledge_tags.id", ondelete="CASCADE"), primary_key=True)
@@ -0,0 +1,240 @@
"""ObsidianWikiAdapter — the obsidian-wiki vault facade.
Owns the Markdown vault: writes notes with wiki-capture frontmatter/body, keeps
``index.md`` / ``log.md`` / ``hot.md`` / ``.manifest.json`` in sync after every
write, and produces wiki-export graph artifacts. Output stays compatible with
``ar9av/obsidian-wiki`` without ever invoking its CLI.
This class is storage-only (no DB, no LLM). The :class:`KnowledgeService`
combines it with the repository + extractor.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
from deerflow.knowledge import exporter
from deerflow.knowledge.manifest import Manifest, content_hash
from deerflow.knowledge.markdown_vault import MarkdownVault
from deerflow.knowledge.schemas import NoteDraft
_INDEX_PAGES_HEADER = "## Pages"
_HOT_HEADER = "## Recent"
_HOT_MAX = 20
class ObsidianWikiAdapter:
"""Read/write an obsidian-wiki compatible vault."""
def __init__(self, vault_root: Path | str, *, project_name: str = "deerflow", export_html: bool = True, export_json: bool = True) -> None:
self.vault = MarkdownVault(vault_root)
self.root = self.vault.root
self.manifest = Manifest(self.root)
self.project_name = project_name
self.export_html = export_html
self.export_json = export_json
def ensure_ready(self) -> None:
self.vault.ensure_scaffold(project_name=self.project_name)
# -- writes -----------------------------------------------------------
def write_note(
self,
note_id: str,
draft: NoteDraft,
*,
created: str,
updated: str,
status: str,
) -> str:
"""Write the note file, sync vault metadata, return the vault path."""
self.ensure_ready()
rel = self.vault.note_relpath(category=draft.category, project_name=self.project_name, slug=draft.title, note_id=note_id)
sources_fm = self._frontmatter_sources(draft)
relationships = [{"target": link, "type": "related"} for link in (draft.related or [])[:8]]
frontmatter = self.vault.build_frontmatter(
note_id=note_id,
title=draft.title,
category=draft.category,
tags=draft.tags,
sources=sources_fm,
summary=draft.summary,
created=created,
updated=updated,
base_confidence=draft.confidence,
lifecycle=status,
relationships=relationships,
)
self.vault.write_note(rel, frontmatter, draft.content_md)
self.manifest.record_page(
note_id=note_id,
vault_path=rel,
source_key=draft.source_key,
hash_value=content_hash(draft.content_md),
updated=updated,
)
self._update_index(note_id, rel, draft.title, draft.summary)
self._append_log("CAPTURE" if draft.source_type in ("thread", "search", "tool") else "INGEST", note_id, draft.title, updated)
self._update_hot(draft.title, rel, updated)
return rel
def update_note_file(
self,
note_id: str,
vault_path: str | None,
*,
title: str,
content_md: str,
tags: list[str],
status: str,
summary: str,
category: str,
created: str,
updated: str,
) -> str:
"""Rewrite a note file in place (frontmatter refreshed), sync metadata."""
self.ensure_ready()
rel = vault_path
if not rel:
rel = self.vault.note_relpath(category=category, project_name=self.project_name, slug=title, note_id=note_id)
# Preserve original `sources`/`provenance`/`category` frontmatter when present.
existing = self.vault.read_note(rel)
sources_fm = existing[0].get("sources", []) if existing else []
provenance = existing[0].get("provenance") if existing else None
relationships = existing[0].get("relationships") if existing else []
effective_category = (existing[0].get("category") if existing else None) or category
frontmatter = self.vault.build_frontmatter(
note_id=note_id,
title=title,
category=effective_category,
tags=tags,
sources=sources_fm,
summary=summary,
created=created,
updated=updated,
base_confidence=float((existing[0].get("base_confidence") if existing else 0.0) or 0.0),
lifecycle=status,
provenance=provenance,
relationships=relationships,
)
self.vault.write_note(rel, frontmatter, content_md)
self.manifest.record_page(
note_id=note_id,
vault_path=rel,
source_key=self.manifest.load().get("pages", {}).get(note_id, {}).get("source"),
hash_value=content_hash(content_md),
updated=updated,
)
self._update_index(note_id, rel, title, summary)
self._append_log("UPDATE", note_id, title, updated)
self._update_hot(title, rel, updated)
return rel
def archive_note(self, note_id: str, vault_path: str | None, *, updated: str) -> None:
"""Mark a note archived in its frontmatter + log (soft delete)."""
if vault_path:
parsed = self.vault.read_note(vault_path)
if parsed:
fm, body = parsed
fm["lifecycle"] = "archived"
fm["updated"] = updated
self.vault.write_note(vault_path, fm, body)
self._remove_from_index(note_id)
self._append_log("ARCHIVE", note_id, "", updated)
def export_graph(self, notes: list[dict[str, Any]], *, updated: str) -> dict[str, Any]:
self.ensure_ready()
result = exporter.export_all(self.root, notes, write_html=self.export_html, write_json=self.export_json)
self._append_log("EXPORT", "graph", f"nodes={result['stats']['nodes']} edges={result['stats']['edges']}", updated)
return result
def export_file_path(self, filename: str) -> Path | None:
allowed = {"graph.json", "graph.graphml", "cypher.txt", "graph.html"}
if filename not in allowed:
return None
path = (self.root / "wiki-export" / filename).resolve()
try:
path.relative_to(self.root)
except ValueError:
return None
return path if path.exists() else None
# -- frontmatter source serialization ---------------------------------
@staticmethod
def _frontmatter_sources(draft: NoteDraft) -> list[str]:
out: list[str] = []
if draft.source_type == "thread" and draft.source_id:
out.append(f"thread:{draft.source_id}")
for s in draft.sources[:20]:
if s.url:
out.append(f"url:{s.url}")
elif s.thread_id:
out.append(f"thread:{s.thread_id}")
elif s.tool_name:
out.append(f"tool:{s.tool_name}")
# De-dup preserving order.
seen: set[str] = set()
deduped = []
for item in out:
if item not in seen:
seen.add(item)
deduped.append(item)
return deduped
# -- index.md ---------------------------------------------------------
def _index_path(self) -> Path:
return self.root / "index.md"
def _update_index(self, note_id: str, rel: str, title: str, summary: str | None) -> None:
path = self._index_path()
text = path.read_text(encoding="utf-8") if path.exists() else "# Knowledge Index\n\n"
link = rel[:-3] if rel.endswith(".md") else rel
summary_txt = (summary or "").strip().replace("\n", " ")
if len(summary_txt) > 120:
summary_txt = summary_txt[:120] + "…"
line = f"- [[{link}|{title}]]" + (f" — {summary_txt}" if summary_txt else "") + f" <!-- id:{note_id} -->"
lines = text.splitlines()
# Drop any existing line for this note id.
lines = [ln for ln in lines if f"<!-- id:{note_id} -->" not in ln]
if _INDEX_PAGES_HEADER not in text:
lines.append("")
lines.append(_INDEX_PAGES_HEADER)
lines.append("")
# Insert after the Pages header.
try:
insert_at = lines.index(_INDEX_PAGES_HEADER) + 1
except ValueError:
insert_at = len(lines)
lines.insert(insert_at, line)
path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")
def _remove_from_index(self, note_id: str) -> None:
path = self._index_path()
if not path.exists():
return
lines = [ln for ln in path.read_text(encoding="utf-8").splitlines() if f"<!-- id:{note_id} -->" not in ln]
path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")
# -- log.md -----------------------------------------------------------
def _append_log(self, action: str, note_id: str, label: str, updated: str) -> None:
path = self.root / "log.md"
text = path.read_text(encoding="utf-8") if path.exists() else "# Activity Log\n\n"
entry = f"- {updated} {action} {note_id}" + (f" — {label}" if label else "")
path.write_text(text.rstrip() + "\n" + entry + "\n", encoding="utf-8")
# -- hot.md -----------------------------------------------------------
def _update_hot(self, title: str, rel: str, updated: str) -> None:
path = self.root / "hot.md"
link = rel[:-3] if rel.endswith(".md") else rel
new_line = f"- {updated} [[{link}|{title}]]"
existing_lines: list[str] = []
if path.exists():
for ln in path.read_text(encoding="utf-8").splitlines():
if ln.startswith("- ") and "[[" in ln:
existing_lines.append(ln)
existing_lines = [ln for ln in existing_lines if f"[[{link}" not in ln]
existing_lines.insert(0, new_line)
existing_lines = existing_lines[:_HOT_MAX]
body = "# Hot\n\n_近期活动摘要。_\n\n" + _HOT_HEADER + "\n\n" + "\n".join(existing_lines) + "\n"
path.write_text(body, encoding="utf-8")
@@ -0,0 +1,315 @@
"""Database read/write for the knowledge base.
This layer never calls an LLM and never touches the Markdown vault — it only
persists/queries the relational mirror (notes, sources, tags). The vault and
extraction live in :mod:`deerflow.knowledge.markdown_vault` and
:mod:`deerflow.knowledge.extractor`; orchestration lives in the service.
"""
from __future__ import annotations
import uuid
from datetime import UTC, datetime
from typing import Any
from sqlalchemy import delete, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.knowledge.models import (
KnowledgeNoteRow,
KnowledgeNoteTagRow,
KnowledgeSourceRow,
KnowledgeTagRow,
)
from deerflow.knowledge.schemas import SourceDraft
def _new_id(prefix: str) -> str:
return f"{prefix}_{uuid.uuid4().hex[:24]}"
def _iso(value: Any) -> Any:
if isinstance(value, datetime):
return value.isoformat()
return value
class KnowledgeRepository:
"""SQLAlchemy-backed repository for knowledge notes/sources/tags."""
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
self._sf = session_factory
# -- serialization ----------------------------------------------------
@staticmethod
def _note_to_dict(row: KnowledgeNoteRow) -> dict[str, Any]:
d = row.to_dict()
d["created_at"] = _iso(d.get("created_at"))
d["updated_at"] = _iso(d.get("updated_at"))
return d
@staticmethod
def _source_to_dict(row: KnowledgeSourceRow) -> dict[str, Any]:
d = row.to_dict()
d["created_at"] = _iso(d.get("created_at"))
return d
# -- tags -------------------------------------------------------------
async def _get_or_create_tags(self, session: AsyncSession, names: list[str]) -> list[KnowledgeTagRow]:
cleaned = []
seen: set[str] = set()
for raw in names or []:
name = (raw or "").strip()
if not name or name.lower() in seen:
continue
seen.add(name.lower())
cleaned.append(name)
if not cleaned:
return []
existing = {
r.name.lower(): r
for r in (await session.execute(select(KnowledgeTagRow).where(func.lower(KnowledgeTagRow.name).in_([n.lower() for n in cleaned])))).scalars()
}
result: list[KnowledgeTagRow] = []
for name in cleaned:
row = existing.get(name.lower())
if row is None:
row = KnowledgeTagRow(id=_new_id("kbtag"), name=name, created_at=datetime.now(UTC))
session.add(row)
await session.flush()
existing[name.lower()] = row
result.append(row)
return result
async def _tags_for(self, session: AsyncSession, note_id: str) -> list[str]:
stmt = (
select(KnowledgeTagRow.name)
.join(KnowledgeNoteTagRow, KnowledgeNoteTagRow.tag_id == KnowledgeTagRow.id)
.where(KnowledgeNoteTagRow.note_id == note_id)
.order_by(KnowledgeTagRow.name)
)
return list((await session.execute(stmt)).scalars())
async def _tags_for_many(self, session: AsyncSession, note_ids: list[str]) -> dict[str, list[str]]:
if not note_ids:
return {}
stmt = (
select(KnowledgeNoteTagRow.note_id, KnowledgeTagRow.name)
.join(KnowledgeTagRow, KnowledgeNoteTagRow.tag_id == KnowledgeTagRow.id)
.where(KnowledgeNoteTagRow.note_id.in_(note_ids))
.order_by(KnowledgeTagRow.name)
)
out: dict[str, list[str]] = {nid: [] for nid in note_ids}
for note_id, name in (await session.execute(stmt)).all():
out.setdefault(note_id, []).append(name)
return out
async def _replace_note_tags(self, session: AsyncSession, note_id: str, names: list[str]) -> None:
await session.execute(delete(KnowledgeNoteTagRow).where(KnowledgeNoteTagRow.note_id == note_id))
tags = await self._get_or_create_tags(session, names)
for tag in tags:
session.add(KnowledgeNoteTagRow(note_id=note_id, tag_id=tag.id))
# -- notes ------------------------------------------------------------
async def create_note(
self,
*,
title: str,
summary: str | None,
content_md: str,
source_type: str,
source_id: str | None,
status: str,
confidence: float = 0.0,
vault_path: str | None = None,
tags: list[str] | None = None,
created_by: str | None = None,
) -> dict[str, Any]:
note_id = _new_id("kb")
now = datetime.now(UTC)
row = KnowledgeNoteRow(
id=note_id,
title=title[:512],
summary=summary,
content_md=content_md,
vault_path=vault_path,
source_type=source_type,
source_id=source_id,
status=status,
confidence=confidence,
created_by=created_by,
updated_by=created_by,
created_at=now,
updated_at=now,
)
async with self._sf() as session:
session.add(row)
await session.flush()
await self._replace_note_tags(session, note_id, tags or [])
await session.commit()
await session.refresh(row)
d = self._note_to_dict(row)
d["tags"] = await self._tags_for(session, note_id)
return d
async def set_vault_path(self, note_id: str, vault_path: str) -> None:
async with self._sf() as session:
row = await session.get(KnowledgeNoteRow, note_id)
if row is None:
return
row.vault_path = vault_path
await session.commit()
async def update_note(
self,
note_id: str,
*,
title: str | None = None,
summary: str | None = None,
content_md: str | None = None,
status: str | None = None,
tags: list[str] | None = None,
vault_path: str | None = None,
updated_by: str | None = None,
) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(KnowledgeNoteRow, note_id)
if row is None:
return None
if title is not None:
row.title = title[:512]
if summary is not None:
row.summary = summary
if content_md is not None:
row.content_md = content_md
if status is not None:
row.status = status
if vault_path is not None:
row.vault_path = vault_path
if updated_by is not None:
row.updated_by = updated_by
row.updated_at = datetime.now(UTC)
if tags is not None:
await self._replace_note_tags(session, note_id, tags)
await session.commit()
await session.refresh(row)
d = self._note_to_dict(row)
d["tags"] = await self._tags_for(session, note_id)
return d
async def set_status(self, note_id: str, status: str, *, updated_by: str | None = None) -> dict[str, Any] | None:
return await self.update_note(note_id, status=status, updated_by=updated_by)
async def get_note(self, note_id: str, *, with_sources: bool = True) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(KnowledgeNoteRow, note_id)
if row is None:
return None
d = self._note_to_dict(row)
d["tags"] = await self._tags_for(session, note_id)
if with_sources:
stmt = select(KnowledgeSourceRow).where(KnowledgeSourceRow.note_id == note_id).order_by(KnowledgeSourceRow.created_at)
d["sources"] = [self._source_to_dict(r) for r in (await session.execute(stmt)).scalars()]
return d
async def find_by_source(self, source_type: str, source_id: str) -> dict[str, Any] | None:
"""Return the first (non-archived) note for a (source_type, source_id)."""
async with self._sf() as session:
stmt = (
select(KnowledgeNoteRow)
.where(KnowledgeNoteRow.source_type == source_type, KnowledgeNoteRow.source_id == source_id)
.where(KnowledgeNoteRow.status != "archived")
.limit(1)
)
row = (await session.execute(stmt)).scalars().first()
return self._note_to_dict(row) if row else None
async def list_notes(
self,
*,
keyword: str | None = None,
tag: str | None = None,
source_type: str | None = None,
status: str | None = None,
include_archived: bool = False,
limit: int = 20,
offset: int = 0,
) -> tuple[list[dict[str, Any]], int]:
async with self._sf() as session:
stmt = select(KnowledgeNoteRow)
count_stmt = select(func.count()).select_from(KnowledgeNoteRow)
conditions = []
if status:
conditions.append(KnowledgeNoteRow.status == status)
elif not include_archived:
conditions.append(KnowledgeNoteRow.status != "archived")
if source_type:
conditions.append(KnowledgeNoteRow.source_type == source_type)
if keyword:
like = f"%{keyword}%"
conditions.append(or_(KnowledgeNoteRow.title.like(like), KnowledgeNoteRow.summary.like(like), KnowledgeNoteRow.content_md.like(like)))
if tag:
tag_subq = (
select(KnowledgeNoteTagRow.note_id)
.join(KnowledgeTagRow, KnowledgeNoteTagRow.tag_id == KnowledgeTagRow.id)
.where(func.lower(KnowledgeTagRow.name) == tag.lower())
)
conditions.append(KnowledgeNoteRow.id.in_(tag_subq))
for cond in conditions:
stmt = stmt.where(cond)
count_stmt = count_stmt.where(cond)
total = (await session.execute(count_stmt)).scalar_one()
stmt = stmt.order_by(KnowledgeNoteRow.updated_at.desc()).limit(limit).offset(offset)
rows = list((await session.execute(stmt)).scalars())
ids = [r.id for r in rows]
tags_map = await self._tags_for_many(session, ids)
items = []
for r in rows:
d = self._note_to_dict(r)
d["tags"] = tags_map.get(r.id, [])
items.append(d)
return items, int(total)
async def list_all(self, *, include_archived: bool = False) -> list[dict[str, Any]]:
"""Return every note (light fields + tags) for graph export."""
async with self._sf() as session:
stmt = select(KnowledgeNoteRow)
if not include_archived:
stmt = stmt.where(KnowledgeNoteRow.status != "archived")
rows = list((await session.execute(stmt)).scalars())
tags_map = await self._tags_for_many(session, [r.id for r in rows])
out = []
for r in rows:
d = self._note_to_dict(r)
d["tags"] = tags_map.get(r.id, [])
out.append(d)
return out
# -- sources ----------------------------------------------------------
async def add_sources(self, note_id: str, sources: list[SourceDraft], *, created_by: str | None = None) -> None:
if not sources:
return
import json
async with self._sf() as session:
for src in sources:
session.add(
KnowledgeSourceRow(
id=_new_id("kbsrc"),
note_id=note_id,
source_type=src.source_type,
thread_id=src.thread_id,
message_id=src.message_id,
tool_name=src.tool_name,
title=(src.title or "")[:512] or None,
url=src.url,
snippet=src.snippet,
raw_json=json.dumps(src.raw, ensure_ascii=False) if src.raw else None,
created_by=created_by,
created_at=datetime.now(UTC),
)
)
await session.commit()
@@ -0,0 +1,71 @@
"""Internal domain models for the knowledge module.
These are plain dataclasses passed between the extractor, the vault writer and
the repository. The Gateway router defines its own Pydantic request/response
models on top of the dicts the service returns.
"""
from __future__ import annotations
from dataclasses import dataclass, field
# Allowed note source types (see product-knowledge-base-dev.md §7.1).
SOURCE_TYPES = ("thread", "search", "tool", "file", "manual", "report")
# Allowed lifecycle states.
STATUSES = ("draft", "approved", "archived")
# obsidian-wiki wiki-capture categories.
CATEGORIES = ("synthesis", "concepts", "references", "decision", "journal")
@dataclass
class SourceDraft:
"""A normalized knowledge source (search result, tool call, thread, …)."""
source_type: str
thread_id: str | None = None
message_id: str | None = None
tool_name: str | None = None
title: str | None = None
url: str | None = None
snippet: str | None = None
raw: dict | None = None
@dataclass
class NoteDraft:
"""A knowledge note ready to be persisted + written to the vault.
Produced by the extractor (from a thread / search / tool result) or built
directly for a manual note. ``content_md`` here is the *body* — the vault
writer prepends obsidian-wiki frontmatter.
"""
title: str
summary: str
content_md: str
category: str = "synthesis"
source_type: str = "manual"
source_id: str | None = None
status: str = "approved"
confidence: float = 0.0
tags: list[str] = field(default_factory=list)
key_points: list[str] = field(default_factory=list)
actions: list[str] = field(default_factory=list)
related: list[str] = field(default_factory=list)
sources: list[SourceDraft] = field(default_factory=list)
# Stable identity hint used for manifest dedup (e.g. ``thread:thread_id``).
source_key: str | None = None
def normalize_status(status: str | None) -> str:
"""Coerce an arbitrary status string to a valid lifecycle state."""
if status in STATUSES:
return status
return "approved"
def normalize_source_type(source_type: str | None) -> str:
"""Coerce an arbitrary source-type string to a valid value."""
if source_type in SOURCE_TYPES:
return source_type
return "manual"
@@ -0,0 +1,90 @@
"""Keyword / frontmatter / Markdown-content search (first release).
Vector and hybrid modes are reserved for phase 2 (see the dev doc §9.9 / §13);
for now ``vector`` and ``hybrid`` transparently fall back to keyword search so
the API contract is stable. Scoring weights title > tags > summary > body.
"""
from __future__ import annotations
import re
from typing import Any
from deerflow.knowledge.repository import KnowledgeRepository
_TOKEN_RE = re.compile(r"[\w一-鿿]+", re.UNICODE)
def _tokens(text: str) -> list[str]:
return [t.lower() for t in _TOKEN_RE.findall(text or "")]
def _snippet(content: str, terms: list[str], *, width: int = 160) -> str:
body = re.sub(r"\s+", " ", content or "").strip()
low = body.lower()
for term in terms:
idx = low.find(term)
if idx >= 0:
start = max(0, idx - width // 3)
end = min(len(body), start + width)
prefix = "…" if start > 0 else ""
suffix = "…" if end < len(body) else ""
return f"{prefix}{body[start:end]}{suffix}"
return body[:width] + ("…" if len(body) > width else "")
async def search_notes(
repo: KnowledgeRepository,
query: str,
*,
mode: str = "keyword",
limit: int = 8,
candidate_cap: int = 500,
) -> list[dict[str, Any]]:
"""Score notes against ``query`` and return the top ``limit`` matches."""
terms = _tokens(query)
if not terms:
return []
# Pull a candidate window (non-archived). Phase 2 will replace this with
# an index; for the first release in-Python scoring is sufficient.
candidates, _total = await repo.list_notes(limit=candidate_cap, offset=0)
scored: list[tuple[float, dict[str, Any]]] = []
for note in candidates:
title = (note.get("title") or "").lower()
summary = (note.get("summary") or "").lower()
content = (note.get("content_md") or "").lower()
tags = " ".join(note.get("tags") or []).lower()
score = 0.0
for term in terms:
if term in title:
score += 3.0
if term in tags:
score += 2.0
if term in summary:
score += 2.0
score += min(content.count(term), 5) * 0.5
if score <= 0:
continue
scored.append((score, note))
scored.sort(key=lambda x: x[0], reverse=True)
max_score = scored[0][0] if scored else 1.0
results = []
for score, note in scored[:limit]:
results.append(
{
"note_id": note["id"],
"chunk_id": None,
"title": note.get("title"),
"summary": note.get("summary"),
"snippet": _snippet(note.get("content_md") or note.get("summary") or "", terms),
"score": round(score / max_score, 4),
"tags": note.get("tags") or [],
"source_type": note.get("source_type"),
"updated_at": note.get("updated_at"),
"sources": [],
}
)
return results
@@ -0,0 +1,282 @@
"""KnowledgeService — high-level orchestration for the knowledge base.
Ties together the repository (DB mirror), the :class:`ObsidianWikiAdapter`
(Markdown vault + index/log/hot/manifest + graph export) and the extractor
(thread/search → :class:`NoteDraft`). The Gateway router only talks to this
service; everything obsidian-wiki specific stays inside the ``knowledge``
module.
"""
from __future__ import annotations
import logging
from pathlib import Path
from typing import Any
from deerflow.config.knowledge_config import KnowledgeConfig
from deerflow.config.runtime_paths import runtime_home
from deerflow.knowledge import extractor
from deerflow.knowledge.markdown_vault import MarkdownVault
from deerflow.knowledge.obsidian_adapter import ObsidianWikiAdapter
from deerflow.knowledge.repository import KnowledgeRepository
from deerflow.knowledge.schemas import NoteDraft, normalize_source_type, normalize_status
from deerflow.knowledge.search import search_notes
from deerflow.knowledge.sources import from_reference_batches, from_search_results
logger = logging.getLogger(__name__)
class EmptyThreadError(ValueError):
"""Raised when a thread has no sediment-able content."""
def _resolve_vault_dir(vault_dir: str) -> Path:
path = Path(vault_dir)
if path.is_absolute():
return path.resolve()
return (runtime_home() / vault_dir).resolve()
class KnowledgeService:
"""Application-facing knowledge base API."""
def __init__(self, repo: KnowledgeRepository, config: KnowledgeConfig) -> None:
self.repo = repo
self.config = config
self.project_name = config.project_name
vault_root = _resolve_vault_dir(config.vault_dir)
self.adapter = ObsidianWikiAdapter(
vault_root,
project_name=config.project_name,
export_html=config.export.graph_html_enabled,
export_json=config.export.graph_json_enabled,
)
try:
self.adapter.ensure_ready()
except Exception: # pragma: no cover - vault init best-effort
logger.exception("Failed to initialize knowledge vault at %s", vault_root)
# -- internal helpers -------------------------------------------------
async def _persist_draft(self, draft: NoteDraft, *, created_by: str | None) -> dict[str, Any]:
"""Create a DB row, write the vault file, attach sources, return note."""
note = await self.repo.create_note(
title=draft.title,
summary=draft.summary,
content_md=draft.content_md,
source_type=draft.source_type,
source_id=draft.source_id,
status=draft.status,
confidence=draft.confidence,
tags=draft.tags,
created_by=created_by,
)
note_id = note["id"]
try:
rel = self.adapter.write_note(note_id, draft, created=note["created_at"], updated=note["updated_at"], status=draft.status)
await self.repo.set_vault_path(note_id, rel)
note["vault_path"] = rel
except Exception:
logger.exception("Failed to write knowledge note %s to vault", note_id)
if draft.sources:
await self.repo.add_sources(note_id, draft.sources, created_by=created_by)
return await self.repo.get_note(note_id) or note
# -- capture ----------------------------------------------------------
async def capture_thread(
self,
*,
thread_id: str,
messages: list[dict[str, Any]],
thread_title: str | None = None,
mode: str = "summary",
title: str | None = None,
status: str = "approved",
tags: list[str] | None = None,
include_sources: bool = True,
reference_batches: list[dict[str, Any]] | None = None,
created_by: str | None = None,
skip_existing: bool = False,
) -> dict[str, Any]:
"""Sediment a thread into a knowledge note.
Returns ``{"note": <note>, "created": bool}``. Raises
:class:`EmptyThreadError` when there is nothing worth keeping.
"""
if skip_existing:
existing = await self.repo.find_by_source("thread", thread_id)
if existing is None and self.adapter.manifest.has_source(f"thread:{thread_id}"):
existing = await self.repo.find_by_source("thread", thread_id)
if existing:
return {"note": existing, "created": False}
sources = []
if include_sources and reference_batches:
sources = from_reference_batches(reference_batches, thread_id=thread_id, max_items=self.config.max_source_items)
draft = extractor.extract_thread(
messages,
thread_id=thread_id,
thread_title=thread_title,
mode=mode,
title=title,
tags=tags,
project_name=self.project_name,
sources=sources,
max_messages=self.config.max_thread_messages,
)
if draft is None:
raise EmptyThreadError(f"Thread {thread_id} has no sediment-able content")
draft.status = normalize_status(status)
note = await self._persist_draft(draft, created_by=created_by)
return {"note": note, "created": True}
async def capture_search(
self,
*,
query: str,
results: list[dict[str, Any]] | None,
thread_id: str | None = None,
message_id: str | None = None,
title: str | None = None,
tags: list[str] | None = None,
status: str = "approved",
created_by: str | None = None,
) -> dict[str, Any]:
"""Sediment search/tool results into a reference note."""
sources = from_search_results(query, results, thread_id=thread_id, message_id=message_id, max_items=self.config.max_source_items)
draft = extractor.extract_search(query, sources, thread_id=thread_id, project_name=self.project_name, title=title, tags=tags)
draft.status = normalize_status(status)
return await self._persist_draft(draft, created_by=created_by)
async def create_manual_note(
self,
*,
title: str,
content_md: str,
tags: list[str] | None = None,
status: str = "approved",
summary: str | None = None,
created_by: str | None = None,
) -> dict[str, Any]:
"""Create a note directly from product-edited Markdown."""
# Strip any frontmatter the user may have pasted; vault writer adds its own.
_, body = MarkdownVault.parse(content_md)
body = body or content_md
base_tags = list(tags or [])
for default in (self.project_name, "knowledge-base"):
if default and default not in base_tags:
base_tags.append(default)
draft = NoteDraft(
title=title.strip() or "知识笔记",
summary=(summary or body.strip().splitlines()[0] if body.strip() else "")[:280],
content_md=body,
category="synthesis",
source_type="manual",
source_id=None,
status=normalize_status(status),
tags=base_tags,
confidence=0.5,
related=extractor.related_links(self.project_name),
)
return await self._persist_draft(draft, created_by=created_by)
# -- CRUD -------------------------------------------------------------
async def list_notes(
self,
*,
keyword: str | None = None,
tag: str | None = None,
source_type: str | None = None,
status: str | None = None,
limit: int = 20,
offset: int = 0,
) -> dict[str, Any]:
items, total = await self.repo.list_notes(
keyword=keyword,
tag=tag,
source_type=normalize_source_type(source_type) if source_type else None,
status=status,
limit=limit,
offset=offset,
)
return {"items": items, "total": total}
async def get_note(self, note_id: str) -> dict[str, Any] | None:
return await self.repo.get_note(note_id)
async def update_note(
self,
note_id: str,
*,
title: str | None = None,
content_md: str | None = None,
tags: list[str] | None = None,
status: str | None = None,
summary: str | None = None,
updated_by: str | None = None,
) -> dict[str, Any] | None:
current = await self.repo.get_note(note_id, with_sources=False)
if current is None:
return None
body = content_md
if body is not None:
_, parsed = MarkdownVault.parse(body)
body = parsed or body
note = await self.repo.update_note(
note_id,
title=title,
summary=summary,
content_md=body,
status=normalize_status(status) if status else None,
tags=tags,
updated_by=updated_by,
)
if note is None:
return None
try:
rel = self.adapter.update_note_file(
note_id,
current.get("vault_path"),
title=note["title"],
content_md=note["content_md"],
tags=note.get("tags") or [],
status=note["status"],
summary=note.get("summary") or "",
category="synthesis",
created=note["created_at"],
updated=note["updated_at"],
)
if rel and rel != current.get("vault_path"):
await self.repo.set_vault_path(note_id, rel)
except Exception:
logger.exception("Failed to update knowledge note %s in vault", note_id)
return await self.repo.get_note(note_id)
async def delete_note(self, note_id: str, *, updated_by: str | None = None, hard: bool = False) -> bool:
"""Soft delete (archive) by default; hard delete removes the vault file."""
current = await self.repo.get_note(note_id, with_sources=False)
if current is None:
return False
note = await self.repo.set_status(note_id, "archived", updated_by=updated_by)
try:
self.adapter.archive_note(note_id, current.get("vault_path"), updated=(note or current)["updated_at"])
if hard and current.get("vault_path"):
self.adapter.vault.delete_note(current["vault_path"])
self.adapter.manifest.remove_page(note_id)
except Exception:
logger.exception("Failed to archive knowledge note %s in vault", note_id)
return True
# -- search & graph ---------------------------------------------------
async def search(self, query: str, *, mode: str = "keyword", limit: int = 8) -> list[dict[str, Any]]:
return await search_notes(self.repo, query, mode=mode, limit=limit)
async def export_graph(self) -> dict[str, Any]:
from deerflow.utils.time import now_iso
notes = await self.repo.list_all()
result = self.adapter.export_graph(notes, updated=now_iso())
return {"files": result["files"], "stats": result["stats"]}
def export_file_path(self, filename: str) -> Path | None:
return self.adapter.export_file_path(filename)
@@ -0,0 +1,51 @@
"""Built-in obsidian-wiki skill templates.
Mirrors the key intent of ``ar9av/obsidian-wiki`` skills so the backend can
write vault content that stays compatible with the original tooling without
shelling out to its CLI:
- ``wiki-capture`` — how a conversation becomes a *declarative* knowledge note
(frontmatter schema + Context/Finding/Reasoning/Implications/Related body).
- ``wiki-query`` — layered retrieval (hot.md → index.md → frontmatter → body).
- ``wiki-export`` — graph artifacts (graph.json / graph.html / graphml / cypher).
These are intentionally lightweight: the structures here are what
:mod:`deerflow.knowledge.markdown_vault` and :mod:`deerflow.knowledge.extractor`
produce, and what the first-release LLM prompt (future enhancement) should
target.
"""
from __future__ import annotations
# Category → vault sub-directory (wiki-capture page taxonomy, §8 of the dev doc).
CATEGORY_DIRS = {
"synthesis": "synthesis",
"concepts": "concepts",
"concept": "concepts",
"references": "references",
"reference": "references",
"source": "references",
"decision": "synthesis",
"journal": "journal",
"session": "journal",
}
# Body section headings (wiki-capture). Declarative knowledge, never a chat log.
BODY_SECTIONS = ("Context", "Finding / Decision", "Reasoning", "Implications", "Related")
# Inline provenance markers reused from wiki-capture.
INFERRED_MARKER = "^[inferred]"
AMBIGUOUS_MARKER = "^[ambiguous]"
WIKI_CAPTURE_GUIDANCE = """\
Capture *declarative* knowledge, not a chat transcript. Classify the note as one
of: synthesis, concept, source, decision, session. Prefer at least two
[[wikilinks]]. Mark inferred content with ^[inferred] and uncertain/conflicting
content with ^[ambiguous]. Project-specific knowledge belongs under
projects/<project-name>/...
"""
def category_to_dir(category: str | None) -> str:
"""Map a wiki-capture category to its vault sub-directory."""
return CATEGORY_DIRS.get((category or "").strip().lower(), "synthesis")
@@ -0,0 +1,82 @@
"""Normalize knowledge sources from threads / searches / tool results.
Source items come from a few shapes: backend reference-batches (built by
``deerflow.runtime.references``), raw search-result payloads, and tool-call
messages. They are all normalized into :class:`SourceDraft` so the repository
and frontend see one consistent structure.
"""
from __future__ import annotations
from typing import Any
from deerflow.knowledge.schemas import SourceDraft
def _clip(value: Any, limit: int = 2000) -> str | None:
if value is None:
return None
text = str(value).strip()
if not text:
return None
return text[:limit]
def from_reference_batches(batches: list[dict[str, Any]] | None, *, thread_id: str | None = None, max_items: int = 20) -> list[SourceDraft]:
"""Convert backend reference-batches into :class:`SourceDraft`."""
out: list[SourceDraft] = []
seen: set[str] = set()
for batch in batches or []:
msg_id = batch.get("assistant_message_id") or batch.get("id")
for src in batch.get("sources", []) or []:
if not isinstance(src, dict):
continue
url = _clip(src.get("url"), 1024)
title = _clip(src.get("title"), 512) or _clip(src.get("source"), 512) or "来源"
dedup_key = url or f"{title}:{_clip(src.get('snippet'), 80)}"
if dedup_key in seen:
continue
seen.add(dedup_key)
out.append(
SourceDraft(
source_type="search" if url else "tool",
thread_id=thread_id,
message_id=_clip(msg_id, 128),
tool_name=_clip(src.get("skillName"), 128),
title=title,
url=url,
snippet=_clip(src.get("snippet")),
raw=None,
)
)
if len(out) >= max_items:
return out
return out
def from_search_results(
query: str,
results: list[dict[str, Any]] | None,
*,
thread_id: str | None = None,
message_id: str | None = None,
max_items: int = 20,
) -> list[SourceDraft]:
"""Convert raw search results (title/url/snippet) into :class:`SourceDraft`."""
out: list[SourceDraft] = []
for item in (results or [])[:max_items]:
if not isinstance(item, dict):
continue
out.append(
SourceDraft(
source_type="search",
thread_id=thread_id,
message_id=message_id,
tool_name=_clip(item.get("tool_name") or item.get("tool"), 128),
title=_clip(item.get("title"), 512) or query[:200],
url=_clip(item.get("url"), 1024),
snippet=_clip(item.get("snippet") or item.get("content")),
raw={k: item.get(k) for k in ("score",) if item.get(k) is not None} or None,
)
)
return out
@@ -24,6 +24,12 @@ from deerflow.persistence.agents.model import (
from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow
from deerflow.persistence.memory_config.model import UserMemoryConfigRow
from deerflow.persistence.feedback.model import FeedbackRow
from deerflow.knowledge.models import (
KnowledgeNoteRow,
KnowledgeNoteTagRow,
KnowledgeSourceRow,
KnowledgeTagRow,
)
from deerflow.persistence.llm_metrics.model import LlmCallMetricRow
from deerflow.persistence.models.run_event import RunEventRow
from deerflow.persistence.notifications.model import NotificationRow
@@ -60,6 +66,10 @@ __all__ = [
"AgentRow",
"AgentSkillRow",
"AdminLeaderboardDailyStatRow",
"KnowledgeNoteRow",
"KnowledgeNoteTagRow",
"KnowledgeSourceRow",
"KnowledgeTagRow",
"LlmCallMetricRow",
"NotificationRow",
"RecommendedQuestionRow",