wsk/deerflow-code
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'master' of http://192.168.0.200/knowledge-base-bj/deerflow-code into dhk-1

Browse Source 
# Conflicts:
#	frontend-web/src/components/workspace/workspace-nav-menu.tsx
#	offline-backend-20260512/backend/app/gateway/routers/multi_agent.py
This commit is contained in:
fengxiangdebaozi committed 2026-07-26 20:55:50 +08:00
2 parents f1e53ef5c0 a368037e81
commit 239b798c5d
42 files changed
+2416 -282
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

frontend-web/src/App.tsx
+4 -1
View File
@@ -68,7 +68,10 @@ function AppProviders({ children }: { children: React.ReactNode }) {
// 走 next-themes forcedTheme（不写 localStorage）→ 同浏览器另开的正常系统页不受影响。
const embedTheme = resolveForcedEmbedTheme(location.search);
if (embedTheme) return embedTheme;
const routeTheme = resolveForcedThemeForPath(location.pathname);
const routeTheme = resolveForcedThemeForPath(
location.pathname,
location.search,
);
if (routeTheme) return routeTheme;
if (location.pathname === "/embed/roundtable") {
const wp = readWujieProps();
frontend-web/src/components/wiki/ReportMarkdownView.tsx
+189
View File
@@ -0,0 +1,189 @@
"use client";
import { BookOpen, ListTree } from "lucide-react";
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
import { MarkdownContent } from "@/components/workspace/messages/markdown-content";
import { Badge } from "@/components/ui/badge";
import { reasoningPlugins } from "@/core/streamdown";
import { createMarkdownAnchorResolver } from "@/core/utils/markdown";
import { cn } from "@/lib/utils";
import "./wiki.css";
function stripMarkdownToText(markdown: string): string {
return markdown
.replace(/```[\s\S]*?```/g, " ")
.replace(/~~~[\s\S]*?~~~/g, " ")
.replace(/!\[[^\]]*\]\([^)]+\)/g, " ")
.replace(/\[([^\]]+)\]\([^)]+\)/g, "$1")
.replace(/^#{1,6}\s+/gm, "")
.replace(/^[>\-*+\d.)\s]+/gm, "")
.replace(/[`*_~|]/g, "")
.replace(/<[^>]+>/g, " ")
.replace(/\s+/g, " ")
.trim();
}
function getMarkdownExcerpt(markdown: string): string {
const text = stripMarkdownToText(markdown);
if (!text) return "";
return text.length > 150 ? `${text.slice(0, 150)}...` : text;
}
function countReadableWords(markdown: string): number {
const text = stripMarkdownToText(markdown);
if (!text) return 0;
const cjk = text.match(/[\u4e00-\u9fff]/g)?.length ?? 0;
const words = text.replace(/[\u4e00-\u9fff]/g, " ").match(/[A-Za-z0-9]+/g)?.length ?? 0;
return cjk + words;
}
function selectorForId(id: string): string {
const css = window.CSS;
if (css?.escape) return `#${css.escape(id)}`;
return `[id="${id.replace(/\\/g, "\\\\").replace(/"/g, '\\"')}"]`;
}
export function ReportMarkdownView({
content,
title,
subtitle,
badgeLabel = "报告阅读",
className,
compactHeader = false,
}: {
content: string;
title?: string;
subtitle?: string;
badgeLabel?: string;
className?: string;
compactHeader?: boolean;
}) {
const rootRef = useRef<HTMLDivElement>(null);
const anchorResolver = useMemo(() => createMarkdownAnchorResolver(content), [content]);
const headings = useMemo(
() => anchorResolver.headings.filter((item) => item.level >= 1 && item.level <= 3),
[anchorResolver],
);
const [activeHeadingId, setActiveHeadingId] = useState<string | null>(null);
const excerpt = useMemo(() => getMarkdownExcerpt(content), [content]);
const sectionCount = headings.length;
const wordCount = useMemo(() => countReadableWords(content), [content]);
useEffect(() => {
setActiveHeadingId(headings[0]?.id ?? null);
}, [headings]);
useEffect(() => {
const root = rootRef.current;
if (!root || headings.length === 0) return;
const targets = headings
.map((item) => root.querySelector<HTMLElement>(selectorForId(item.id)))
.filter((item): item is HTMLElement => !!item);
if (targets.length === 0) return;
const observer = new IntersectionObserver(
(entries) => {
const visible = entries
.filter((entry) => entry.isIntersecting)
.sort((a, b) => a.boundingClientRect.top - b.boundingClientRect.top)[0];
if (visible?.target.id) setActiveHeadingId(visible.target.id);
},
{
root: root.closest("[data-radix-scroll-area-viewport]") ?? null,
rootMargin: "-96px 0px -65% 0px",
threshold: [0, 1],
},
);
targets.forEach((target) => observer.observe(target));
return () => observer.disconnect();
}, [headings, content]);
const scrollToHeading = useCallback((id: string) => {
const root = rootRef.current;
const target =
root?.querySelector<HTMLElement>(selectorForId(id)) ?? document.getElementById(id);
setActiveHeadingId(id);
target?.scrollIntoView({ behavior: "smooth", block: "start" });
}, []);
return (
<div
ref={rootRef}
className={cn("report-markdown-shell grid min-h-0 gap-5 lg:grid-cols-[minmax(0,1fr)_300px]", className)}
>
<article className="report-markdown-paper min-w-0 overflow-hidden">
<header className={cn("report-markdown-header", compactHeader && "report-markdown-header-compact")}>
<div className="flex flex-wrap items-center gap-2">
<Badge variant="secondary" className="report-markdown-badge">
<BookOpen className="size-3.5" />
{badgeLabel}
</Badge>
{subtitle && (
<span className="max-w-full truncate text-xs text-slate-500 dark:text-slate-400">
{subtitle}
</span>
)}
</div>
{title && <h1 className="report-markdown-title">{title}</h1>}
{!compactHeader && excerpt && <p className="report-markdown-excerpt">{excerpt}</p>}
<div className="report-markdown-metrics">
<span>{sectionCount || 1} 个章节</span>
<span>约 {wordCount} 字</span>
</div>
</header>
<div className="wiki-markdown report-markdown-body">
<MarkdownContent
className="max-w-none"
content={content}
isLoading={false}
remarkPlugins={reasoningPlugins.remarkPlugins}
rehypePlugins={reasoningPlugins.rehypePlugins}
/>
</div>
</article>
<aside className="hidden min-w-0 lg:block">
<div className="report-markdown-toc">
<div className="mb-3 flex items-center justify-between gap-2">
<div className="flex items-center gap-2 text-sm font-semibold text-slate-900 dark:text-slate-100">
<ListTree className="size-4" />
内容目录
</div>
<span className="rounded-md bg-slate-100 px-1.5 py-0.5 text-xs text-slate-500 dark:bg-slate-800 dark:text-slate-400">
{sectionCount || 0}
</span>
</div>
{headings.length > 0 ? (
<nav className="report-markdown-toc-list">
{headings.map((item) => (
<button
key={item.id}
type="button"
className={cn(
"report-markdown-toc-item",
activeHeadingId === item.id && "report-markdown-toc-item-active",
item.level === 2 && "report-markdown-toc-level-2",
item.level === 3 && "report-markdown-toc-level-3",
)}
title={item.text}
onClick={() => scrollToHeading(item.id)}
>
{item.text}
</button>
))}
</nav>
) : (
<div className="rounded-md border border-dashed p-3 text-xs text-slate-500 dark:border-slate-800 dark:text-slate-400">
暂无标题目录
</div>
)}
</div>
</aside>
</div>
);
}
frontend-web/src/components/wiki/wiki.css
+334
View File
@@ -111,6 +111,340 @@
border-color: #eef1f5;
}
.public-report-markdown {
color: #334155;
font-feature-settings: "kern";
text-rendering: optimizeLegibility;
}
.public-report-markdown h1 {
font-size: 30px;
letter-spacing: 0;
margin-bottom: 24px;
}
.public-report-markdown h2 {
font-size: 22px;
margin-top: 42px;
margin-bottom: 16px;
padding-bottom: 10px;
}
.public-report-markdown h3 {
font-size: 18px;
margin-top: 30px;
margin-bottom: 12px;
}
.public-report-markdown p,
.public-report-markdown li {
font-size: 16px;
line-height: 1.9;
}
.public-report-markdown p {
margin-bottom: 16px;
}
.public-report-markdown ul,
.public-report-markdown ol {
margin: 14px 0 18px;
padding-left: 26px;
}
.public-report-markdown ul {
list-style-type: disc;
}
.public-report-markdown ol {
list-style-type: decimal;
}
.public-report-markdown li {
padding-left: 2px;
}
.public-report-markdown code {
border-radius: 6px;
background: #f1f5f9;
padding: 2px 6px;
color: #be123c;
font-size: 0.92em;
}
.public-report-markdown pre {
margin: 20px 0;
border-radius: 8px;
background: #0f172a;
padding: 16px;
overflow-x: auto;
}
.public-report-markdown pre code {
background: transparent;
padding: 0;
color: inherit;
}
.public-report-markdown table {
display: block;
overflow-x: auto;
border-radius: 8px;
border: 1px solid #e8ecf1;
}
.public-report-markdown th,
.public-report-markdown td {
min-width: 120px;
vertical-align: top;
}
.public-report-markdown blockquote {
margin: 20px 0;
}
.dark .public-report-markdown {
color: #cbd5e1;
}
.dark .public-report-markdown code {
background: #1e293b;
color: #f9a8d4;
}
.dark .public-report-markdown pre {
background: #020617;
}
.dark .public-report-markdown table {
border-color: #334155;
}
.report-markdown-shell {
align-items: start;
padding: 2px;
}
.report-markdown-paper {
border: 1px solid rgba(203, 213, 225, 0.85);
border-radius: 8px;
background:
linear-gradient(180deg, rgba(248, 250, 252, 0.92), rgba(255, 255, 255, 0) 220px),
#ffffff;
box-shadow:
0 1px 2px rgba(15, 23, 42, 0.04),
0 18px 44px rgba(15, 23, 42, 0.07);
}
.report-markdown-header {
border-bottom: 1px solid #e8edf4;
padding: 26px 34px 22px;
}
.report-markdown-header-compact {
padding-bottom: 18px;
}
.report-markdown-badge {
gap: 6px;
border: 1px solid #dbe3ef;
background: #ffffff;
color: #475569;
}
.report-markdown-title {
margin: 14px 0 0;
color: #0f172a;
font-size: 26px;
font-weight: 700;
line-height: 1.35;
letter-spacing: 0;
}
.report-markdown-excerpt {
margin: 14px 0 0;
max-width: 840px;
color: #64748b;
font-size: 14px;
line-height: 1.8;
}
.report-markdown-metrics {
display: flex;
flex-wrap: wrap;
gap: 8px;
margin-top: 16px;
color: #64748b;
font-size: 12px;
}
.report-markdown-metrics span {
border: 1px solid #e2e8f0;
border-radius: 6px;
background: rgba(255, 255, 255, 0.78);
padding: 4px 8px;
}
.report-markdown-body {
max-width: 920px;
margin: 0 auto;
padding: 32px 42px 44px;
}
.report-markdown-body h1 {
font-size: 28px;
}
.report-markdown-body h2 {
position: relative;
margin-top: 44px;
border-bottom-color: #e5edf7;
font-size: 22px;
}
.report-markdown-body h2::before {
content: "";
position: absolute;
left: -16px;
top: 0.34em;
width: 4px;
height: 1.05em;
border-radius: 999px;
background: #2563eb;
}
.report-markdown-body h3 {
color: #1e293b;
font-size: 18px;
}
.report-markdown-body p,
.report-markdown-body li {
font-size: 16px;
line-height: 1.92;
}
.report-markdown-body blockquote {
border-left-color: #2563eb;
background: #f8fbff;
}
.report-markdown-body table {
box-shadow: 0 1px 0 rgba(15, 23, 42, 0.04);
}
.report-markdown-toc {
position: sticky;
top: 16px;
max-height: calc(100vh - 32px);
overflow: hidden;
border: 1px solid rgba(203, 213, 225, 0.82);
border-radius: 8px;
background: rgba(255, 255, 255, 0.92);
padding: 14px;
box-shadow: 0 12px 34px rgba(15, 23, 42, 0.06);
backdrop-filter: blur(10px);
}
.report-markdown-toc-list {
max-height: calc(100vh - 128px);
overflow: auto;
padding-right: 4px;
}
.report-markdown-toc-item {
display: block;
width: 100%;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
border-left: 2px solid transparent;
border-radius: 6px;
padding: 7px 8px;
color: #64748b;
font-size: 12px;
line-height: 1.4;
text-align: left;
transition: background 160ms ease, border-color 160ms ease, color 160ms ease;
}
.report-markdown-toc-item:hover {
background: #f1f5f9;
color: #0f172a;
}
.report-markdown-toc-item-active {
border-left-color: #2563eb;
background: #eff6ff;
color: #1d4ed8;
font-weight: 600;
}
.report-markdown-toc-level-2 {
padding-left: 18px;
}
.report-markdown-toc-level-3 {
padding-left: 30px;
color: #94a3b8;
}
.dark .report-markdown-paper {
border-color: #1f2937;
background:
linear-gradient(180deg, rgba(15, 23, 42, 0.9), rgba(2, 6, 23, 0) 220px),
#020617;
box-shadow: 0 18px 44px rgba(0, 0, 0, 0.24);
}
.dark .report-markdown-header {
border-bottom-color: #1e293b;
}
.dark .report-markdown-badge,
.dark .report-markdown-metrics span {
border-color: #334155;
background: #0f172a;
color: #cbd5e1;
}
.dark .report-markdown-title {
color: #f8fafc;
}
.dark .report-markdown-excerpt,
.dark .report-markdown-metrics {
color: #94a3b8;
}
.dark .report-markdown-body h3 {
color: #e2e8f0;
}
.dark .report-markdown-body h2::before {
background: #60a5fa;
}
.dark .report-markdown-body blockquote {
background: #0f172a;
border-left-color: #60a5fa;
}
.dark .report-markdown-toc {
border-color: #1f2937;
background: rgba(2, 6, 23, 0.88);
}
.dark .report-markdown-toc-item:hover {
background: #0f172a;
color: #f8fafc;
}
.dark .report-markdown-toc-item-active {
background: rgba(37, 99, 235, 0.16);
color: #93c5fd;
}
/* Anchor target offset so headings clear the floating "目录" pill. */
.wiki-markdown h1,
.wiki-markdown h2,
frontend-web/src/components/workspace/agents/agent-selector.tsx
+43 -39
View File
@@ -93,43 +93,47 @@ export function AgentSelector({
return (
<>
<Dialog open={open} onOpenChange={setOpen}>
<DialogTrigger asChild>
{iconMode ? (
<Tooltip content={iconTooltipLabel}>
<Button
type="button"
variant="ghost"
size="icon"
disabled={disabled}
aria-label={selectedAgent ? `已选择智能体：${triggerLabel}` : "选择智能体"}
className={cn(
"relative size-9 rounded-full text-muted-foreground shadow-none hover:bg-muted/60 hover:text-foreground",
selectedAgent && "text-amber-600 hover:text-amber-700",
selectedAgent && iconActiveClassName,
iconButtonClassName,
)}
>
<BotIcon className="size-4" />
{selectedAgent && (
<span
role="button"
tabIndex={0}
aria-label="清除已选 agent"
className="absolute -right-1 -top-1 inline-flex size-4 items-center justify-center rounded-full border bg-background text-foreground shadow-sm hover:bg-muted"
onClick={handleClear}
onKeyDown={(e) => {
if (e.key === "Enter" || e.key === " ") {
e.preventDefault();
onAgentChange(null);
}
}}
>
<XIcon className="size-2.5" />
</span>
)}
</Button>
</Tooltip>
) : (
{iconMode ? (
<Tooltip content={iconTooltipLabel}>
<div>
<DialogTrigger asChild>
<Button
type="button"
variant="ghost"
size="icon"
disabled={disabled}
aria-label={selectedAgent ? `已选择智能体：${triggerLabel}` : "选择智能体"}
className={cn(
"relative size-9 rounded-full text-muted-foreground shadow-none hover:bg-muted/60 hover:text-foreground",
selectedAgent && "text-amber-600 hover:text-amber-700",
selectedAgent && iconActiveClassName,
iconButtonClassName,
)}
>
<BotIcon className="size-4" />
{selectedAgent && (
<span
role="button"
tabIndex={0}
aria-label="清除已选 agent"
className="absolute -right-1 -top-1 inline-flex size-4 items-center justify-center rounded-full border bg-background text-foreground shadow-sm hover:bg-muted"
onClick={handleClear}
onKeyDown={(e) => {
if (e.key === "Enter" || e.key === " ") {
e.preventDefault();
onAgentChange(null);
}
}}
>
<XIcon className="size-2.5" />
</span>
)}
</Button>
</DialogTrigger>
</div>
</Tooltip>
) : (
<DialogTrigger asChild>
<PromptInputButton
type="button"
disabled={disabled}
@@ -167,8 +171,8 @@ export function AgentSelector({
</span>
)}
</PromptInputButton>
)}
</DialogTrigger>
</DialogTrigger>
)}
<DialogContent className="overflow-hidden p-0 sm:max-w-md">
<DialogTitle className="sr-only">选择 Agent</DialogTitle>
<Command className="**:data-[slot=command-input-wrapper]:h-auto">
frontend-web/src/components/workspace/agents/skill-transfer-selector.tsx
+31 -10
View File
@@ -45,11 +45,32 @@ function skillLabel(skill: Skill) {
function matchesQuery(skill: Skill, q: string) {
if (!q) return true;
return (
skill.name.toLowerCase().includes(q) ||
(skill.name_zh ?? "").toLowerCase().includes(q) ||
(skill.description ?? "").toLowerCase().includes(q)
);
const haystack = [
skill.name,
skill.name_zh,
skill.description,
skill.detail,
skill.category,
skill.license,
skill.source,
skill.state,
skill.owner_username,
skill.created_at,
skill.last_activity_at,
skill.use_count != null ? `使用 ${skill.use_count}` : null,
skill.view_count != null ? `查看 ${skill.view_count}` : null,
skill.patch_count != null ? `变迁 ${skill.patch_count}` : null,
...(skill.tags ?? []).flatMap((tag) => [
tag.name,
tag.description,
tag.color,
tag.usage_count != null ? `标签使用 ${tag.usage_count}` : null,
]),
]
.filter((value): value is string => typeof value === "string" && value.length > 0)
.join(" ")
.toLowerCase();
return haystack.includes(q);
}
export function SkillTransferSelector({
@@ -159,7 +180,7 @@ export function SkillTransferSelector({
<div className="relative flex-1">
<SearchIcon className="text-muted-foreground absolute left-2 top-1/2 size-3.5 -translate-y-1/2" />
<Input
placeholder="搜索全部技能(中英文名称、描述)…"
placeholder="搜索技能名称、描述、分类、标签、变迁..."
value={search}
onChange={(e) => {
setSearch(e.currentTarget.value);
@@ -239,7 +260,7 @@ export function SkillTransferSelector({
{/* Two-column transfer panes */}
<div className="grid grid-cols-2 gap-3">
<SkillPane
title="可用技能"
title="未勾选技能"
count={availableSkills.length}
items={availableView}
variant="available"
@@ -248,10 +269,10 @@ export function SkillTransferSelector({
onPageChange={setAvailablePage}
onPick={add}
onViewDetail={onViewDetail}
emptyHint={query || category !== ALL_CATEGORIES ? "没有匹配的技能" : "已全部启用"}
emptyHint={query || category !== ALL_CATEGORIES ? "没有匹配的技能" : "已全部勾选"}
/>
<SkillPane
title="已启用技能"
title="已勾选技能"
count={selectedSkills.length}
items={selectedView}
variant="selected"
@@ -260,7 +281,7 @@ export function SkillTransferSelector({
onPageChange={setSelectedPage}
onPick={remove}
onViewDetail={onViewDetail}
emptyHint={query ? "没有匹配的已选技能" : "尚未启用任何技能"}
emptyHint={query ? "没有匹配的已选技能" : "尚未勾选任何技能"}
/>
</div>
</div>
frontend-web/src/components/workspace/settings/skill-settings-page.tsx
+15 -10
View File
@@ -470,11 +470,11 @@ function SkillSettingsList({ onClose }: { onClose?: () => void }) {
</Button>
</div>
</header>
<TagSearchBar
targetType="skill"
onSearchChange={setSearch}
tagIds={tagIds}
onTagIdsChange={setTagIds}
<TagSearchBar
targetType="skill"
onSearchChange={setSearch}
tagIds={tagIds}
onTagIdsChange={setTagIds}
actions={
<div className="flex items-center gap-2">
<ViewModeToggle value={viewMode} onChange={setViewMode} />
@@ -497,11 +497,16 @@ function SkillSettingsList({ onClose }: { onClose?: () => void }) {
<SelectItem value="name_desc">名称 Z-A</SelectItem>
</SelectContent>
</Select>
</div>
}
/>
{publisher && (
<div className="flex items-center gap-2">
</div>
}
/>
{!isLoading && !error && (
<div className="text-muted-foreground flex justify-end text-sm tabular-nums">
共 {orderedSkills.length} 个技能
</div>
)}
{publisher && (
<div className="flex items-center gap-2">
<span className="text-muted-foreground text-xs">筛选发布者：</span>
<button
type="button"
frontend-web/src/components/workspace/workspace-nav-menu.tsx
+2 -3
View File
@@ -11,7 +11,6 @@ import {
GaugeIcon,
HistoryIcon,
MessageCircleQuestionIcon,
PlugIcon,
ScrollTextIcon,
Settings2Icon,
SettingsIcon,
@@ -219,14 +218,14 @@ export function WorkspaceNavMenu() {
<TagsIcon />
{t.adminTags.title}
</DropdownMenuItem>
{isAdmin && (
{/* {isAdmin && (
<DropdownMenuItem
onClick={() => navigate("/page/workspace/admin/plugins")}
>
<PlugIcon />
插件管理
</DropdownMenuItem>
)}
)} */}
{isAdmin && (
<DropdownMenuItem
onClick={() => navigate("/page/workspace/admin/positions")}
frontend-web/src/core/page-layout/chats-iframe-route.test.mjs
+17
View File
@@ -31,3 +31,20 @@ test("forces light theme for landing and chats_iframe routes only", () => {
);
assert.equal(resolveForcedThemeForPath("/page/workspace/chats/new"), undefined);
});
test("honors the dark-blue theme query on chats_iframe routes", () => {
assert.equal(
resolveForcedThemeForPath(
"/page/workspace/chats_iframe/new",
"?from=iframe&theme=dark-blue",
),
"dark",
);
assert.equal(
resolveForcedThemeForPath(
"/page/workspace/chats_iframe/new",
"?from=iframe&theme=red",
),
"light",
);
});
frontend-web/src/core/page-layout/chats-iframe-route.ts
+11 -2
View File
@@ -1,3 +1,5 @@
import { resolveEmbedThemeMode } from "@/core/embed/embed-session";
export function isChatsIframeWorkspacePath(pathname: string): boolean {
return (
pathname === "/page/workspace/chats_iframe/new" ||
@@ -7,10 +9,17 @@ export function isChatsIframeWorkspacePath(pathname: string): boolean {
export function resolveForcedThemeForPath(
pathname: string,
): "light" | undefined {
if (pathname === "/landing" || isChatsIframeWorkspacePath(pathname)) {
search = "",
): "light" | "dark" | undefined {
if (pathname === "/landing") {
return "light";
}
if (isChatsIframeWorkspacePath(pathname)) {
const normalizedSearch = search.startsWith("?") ? search : `?${search}`;
const rawTheme = new URLSearchParams(normalizedSearch).get("theme")?.trim();
return rawTheme ? resolveEmbedThemeMode(rawTheme) : "light";
}
return undefined;
}
frontend-web/src/core/plugin-management/domain-agent.test.mjs
+1 -1
View File
@@ -187,4 +187,4 @@ test("unmapped incoming url resets auto-selected agent back to default when ther
forceDefaultAssistant: true,
},
);
});
});
frontend-web/src/core/system-settings/api.ts
+4
View File
@@ -504,12 +504,16 @@ export interface ModelConcurrencySettings {
enabled: boolean;
default_limit: number;
model_limits: Record<string, number>;
user_default_limit: number;
user_model_limits: Record<string, number>;
}
export interface ModelConcurrencyUpdate {
enabled?: boolean;
default_limit?: number;
model_limits?: Record<string, number>;
user_default_limit?: number;
user_model_limits?: Record<string, number>;
}
export async function loadSkillCompressionSettings(): Promise<SkillCompressionSettings> {
frontend-web/src/core/threads/hooks.ts
+43 -13
View File
@@ -322,11 +322,24 @@ function isModelConcurrencyLimitError(message: string): boolean {
return (
/\b429\b/.test(message) ||
/并发已达到上限/.test(message) ||
/并发调用已达到上限/.test(message) ||
/使用人数较多/.test(message) ||
/too many requests/i.test(message)
);
}
function friendlySendLimitMessage(message: string): string | null {
if (isActiveRunConflictError(message)) {
return "当前对话还在生成回答，请等待本轮回复结束，或先点“停止”再发送新的问题。";
}
if (isModelConcurrencyLimitError(message)) {
return /当前账号对该模型/.test(message)
? "当前账号对该模型的并发调用已达到上限，请等待已有回答结束后再试。"
: "当前使用人数较多，请稍等一下再试。";
}
return null;
}
function buildCurrentUserLabel(): string {
const auth = getStoredAuth();
if (!auth) {
@@ -684,14 +697,11 @@ export function useThreadStream({
}
return;
}
// 上一轮还没结束又发了新问题（后端 409）：给一句明确的中文提示，而不是
// 把原始的英文/HTTP 错误抛给用户。
if (isActiveRunConflictError(message)) {
toast.info("当前对话还在生成回答，请等待本轮回复结束，或先点「停止」再发送新的问题。");
return;
}
if (isModelConcurrencyLimitError(message)) {
toast.info("当前使用人数较多，请稍等一下再试。");
// 上一轮还没结束又发了新问题（409）或并发达到上限（429）：
// 给一句明确的中文提示，而不是把原始 HTTP 错误抛给用户。
const friendly = friendlySendLimitMessage(message);
if (friendly) {
toast.info(friendly);
return;
}
toast.error(message);
@@ -1197,11 +1207,31 @@ export function useThreadStream({
},
);
void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
} catch (error) {
setOptimisticMessages([]);
setIsUploading(false);
throw error;
} finally {
} catch (error) {
const message =
error instanceof Error ? error.message : String(error);
const friendly = friendlySendLimitMessage(message);
if (friendly) {
setOptimisticMessages((messages) => {
const human = messages.find((item) => item.type === "human");
const kept = human ? [human] : [];
return [
...kept,
{
type: "ai",
id: `opt-ai-limit-${Date.now()}`,
content: friendly,
additional_kwargs: { element: "task" },
} as Message,
];
});
setIsUploading(false);
return;
}
setOptimisticMessages([]);
setIsUploading(false);
throw error;
} finally {
sendInFlightRef.current = false;
}
},
frontend-web/src/pages/IframeChatPage.tsx
+1 -1
View File
@@ -707,4 +707,4 @@ export default function IframeChatPage() {
</ChatBox>
</ThreadContext.Provider>
);
}
}
frontend-web/src/pages/ModelConcurrencySettingsPage.tsx
+108 -49
View File
@@ -13,6 +13,21 @@ import {
useUpdateModelConcurrencySettings,
} from "@/core/system-settings/hooks";
function cleanLimitMap(values: Record<string, string>, label: string): Record<string, number> | null {
const cleaned: Record<string, number> = {};
for (const [name, rawValue] of Object.entries(values)) {
const trimmed = rawValue.trim();
if (!trimmed) continue;
const parsed = Math.round(Number(trimmed));
if (!Number.isFinite(parsed) || parsed < 1) {
toast.error(`${label}「${name}」必须是大于 0 的整数`);
return null;
}
cleaned[name] = Math.min(parsed, 10000);
}
return cleaned;
}
export default function ModelConcurrencySettingsPage({
embedded = false,
}: {
@@ -24,13 +39,16 @@ export default function ModelConcurrencySettingsPage({
const [enabled, setEnabled] = useState(false);
const [defaultLimit, setDefaultLimit] = useState(10);
const [userDefaultLimit, setUserDefaultLimit] = useState(0);
const [modelLimits, setModelLimits] = useState<Record<string, string>>({});
const [userModelLimits, setUserModelLimits] = useState<Record<string, string>>({});
const initializedRef = useRef(false);
useEffect(() => {
if (!data || initializedRef.current) return;
setEnabled(data.enabled);
setDefaultLimit(data.default_limit);
setUserDefaultLimit(data.user_default_limit ?? 0);
setModelLimits(
Object.fromEntries(
Object.entries(data.model_limits ?? {}).map(([name, limit]) => [
@@ -39,39 +57,45 @@ export default function ModelConcurrencySettingsPage({
]),
),
);
setUserModelLimits(
Object.fromEntries(
Object.entries(data.user_model_limits ?? {}).map(([name, limit]) => [
name,
String(limit),
]),
),
);
initializedRef.current = true;
}, [data]);
const modelNames = useMemo(() => models.map((model) => model.name), [models]);
const handleModelLimitChange = (modelName: string, value: string) => {
setModelLimits((prev) => ({ ...prev, [modelName]: value }));
};
const handleSave = async () => {
const nextDefault = Math.max(1, Math.min(Math.round(defaultLimit) || 10, 10000));
const cleaned: Record<string, number> = {};
for (const [name, rawValue] of Object.entries(modelLimits)) {
const trimmed = rawValue.trim();
if (!trimmed) continue;
const parsed = Math.round(Number(trimmed));
if (!Number.isFinite(parsed) || parsed < 1) {
toast.error(`模型 ${name} 的并发上限必须是大于 0 的整数`);
return;
}
cleaned[name] = Math.min(parsed, 10000);
}
const nextUserDefault = Math.max(0, Math.min(Math.round(userDefaultLimit) || 0, 10000));
const cleanedModelLimits = cleanLimitMap(modelLimits, "模型总上限");
if (!cleanedModelLimits) return;
const cleanedUserModelLimits = cleanLimitMap(userModelLimits, "单用户上限");
if (!cleanedUserModelLimits) return;
try {
await update.mutateAsync({
enabled,
default_limit: nextDefault,
model_limits: cleaned,
model_limits: cleanedModelLimits,
user_default_limit: nextUserDefault,
user_model_limits: cleanedUserModelLimits,
});
setDefaultLimit(nextDefault);
setUserDefaultLimit(nextUserDefault);
setModelLimits(
Object.fromEntries(
Object.entries(cleaned).map(([name, limit]) => [name, String(limit)]),
Object.entries(cleanedModelLimits).map(([name, limit]) => [name, String(limit)]),
),
);
setUserModelLimits(
Object.fromEntries(
Object.entries(cleanedUserModelLimits).map(([name, limit]) => [name, String(limit)]),
),
);
toast.success("已保存");
@@ -85,7 +109,7 @@ export default function ModelConcurrencySettingsPage({
className={
embedded
? "flex w-full flex-col gap-6"
: "mx-auto flex h-full min-h-0 w-full max-w-3xl flex-col gap-6 overflow-y-auto px-6 py-10"
: "mx-auto flex h-full min-h-0 w-full max-w-4xl flex-col gap-6 overflow-y-auto px-6 py-10"
}
>
<div>
@@ -93,7 +117,7 @@ export default function ModelConcurrencySettingsPage({
模型并发限制
</h1>
<p className="text-muted-foreground mt-1 text-sm">
按模型限制同时进行中的问答数。达到上限后，新请求会收到“当前使用人数较多，请稍等一下再试”的提示。
按模型控制系统总并发，也可以限制单个用户对某个模型同时发起的调用数。达到上限时会在发送问题后立即提示，问题会保留在对话中。
</p>
</div>
@@ -102,42 +126,59 @@ export default function ModelConcurrencySettingsPage({
<Loader2Icon className="text-muted-foreground size-6 animate-spin" />
</div>
) : error ? (
<div className="text-destructive rounded-xl border border-dashed py-16 text-center text-sm">
<div className="text-destructive rounded-lg border border-dashed py-16 text-center text-sm">
加载失败：{error.message}
</div>
) : (
<div className="flex flex-col gap-5 rounded-xl border bg-white px-5 py-5 dark:border-[#213D5E] dark:bg-[#02244C] dark:text-white">
<div className="flex flex-col gap-5 rounded-lg border bg-white px-5 py-5 dark:border-[#213D5E] dark:bg-[#02244C] dark:text-white">
<div className="flex items-center justify-between gap-4">
<div>
<div className="text-sm font-medium">开启限制</div>
<div className="text-muted-foreground mt-0.5 text-xs">
关闭后，不再按模型并发数拦截问答请求。
关闭后，不再按模型或用户并发数拦截问答请求。
</div>
</div>
<Switch checked={enabled} onCheckedChange={setEnabled} />
</div>
<div className="flex flex-col gap-1.5">
<p className="text-sm font-medium">默认上限</p>
<p className="text-muted-foreground text-xs">
未单独配置的模型，使用这个并发上限。
</p>
<Input
type="number"
min={1}
max={10000}
value={defaultLimit}
disabled={!enabled}
onChange={(e) => setDefaultLimit(Number(e.target.value))}
className="w-32"
/>
<div className="grid gap-4 md:grid-cols-2">
<div className="flex flex-col gap-1.5">
<p className="text-sm font-medium">模型默认总上限</p>
<p className="text-muted-foreground text-xs">
未单独配置的模型，使用这个系统总并发上限。
</p>
<Input
type="number"
min={1}
max={10000}
value={defaultLimit}
disabled={!enabled}
onChange={(e) => setDefaultLimit(Number(e.target.value))}
className="w-32"
/>
</div>
<div className="flex flex-col gap-1.5">
<p className="text-sm font-medium">单用户默认上限</p>
<p className="text-muted-foreground text-xs">
同一用户对同一模型的默认并发上限。填 0 表示不启用默认用户限制。
</p>
<Input
type="number"
min={0}
max={10000}
value={userDefaultLimit}
disabled={!enabled}
onChange={(e) => setUserDefaultLimit(Number(e.target.value))}
className="w-32"
/>
</div>
</div>
<div className="flex flex-col gap-3">
<div>
<p className="text-sm font-medium">按模型单独设置</p>
<p className="text-muted-foreground mt-0.5 text-xs">
留空表示使用默认上限。
留空表示使用对应默认值；单用户上限填空且默认值为 0 时，不限制该维度。
</p>
</div>
@@ -150,24 +191,42 @@ export default function ModelConcurrencySettingsPage({
modelNames.map((modelName) => (
<div
key={modelName}
className="flex items-center justify-between gap-4 rounded-lg border px-4 py-3"
className="grid gap-3 rounded-lg border px-4 py-3 md:grid-cols-[minmax(0,1fr)_8rem_8rem]"
>
<div className="min-w-0">
<div className="truncate text-sm font-medium">{modelName}</div>
<div className="text-muted-foreground text-xs">
留空时使用默认上限 {defaultLimit || 10}
默认总上限 {defaultLimit || 10}；默认单用户上限 {userDefaultLimit || "不限制"}
</div>
</div>
<Input
type="number"
min={1}
max={10000}
value={modelLimits[modelName] ?? ""}
disabled={!enabled}
onChange={(e) => handleModelLimitChange(modelName, e.target.value)}
className="w-32"
placeholder={String(defaultLimit || 10)}
/>
<label className="flex flex-col gap-1">
<span className="text-muted-foreground text-xs">模型总上限</span>
<Input
type="number"
min={1}
max={10000}
value={modelLimits[modelName] ?? ""}
disabled={!enabled}
onChange={(e) =>
setModelLimits((prev) => ({ ...prev, [modelName]: e.target.value }))
}
placeholder={String(defaultLimit || 10)}
/>
</label>
<label className="flex flex-col gap-1">
<span className="text-muted-foreground text-xs">单用户上限</span>
<Input
type="number"
min={1}
max={10000}
value={userModelLimits[modelName] ?? ""}
disabled={!enabled}
onChange={(e) =>
setUserModelLimits((prev) => ({ ...prev, [modelName]: e.target.value }))
}
placeholder={userDefaultLimit > 0 ? String(userDefaultLimit) : "不限制"}
/>
</label>
</div>
))
)}
frontend-web/src/pages/NewAgentPage.tsx
+11 -83
View File
@@ -3,9 +3,7 @@
import {
ArrowLeftIcon,
CheckCircleIcon,
InfoIcon,
SaveIcon,
SearchIcon,
} from "lucide-react";
import { useParams, useRouter } from "next/navigation";
import { useCallback, useEffect, useMemo, useState } from "react";
@@ -38,6 +36,7 @@ import {
type SetupAgentApprovalPayload,
} from "@/components/workspace/messages";
import { ThreadContext } from "@/components/workspace/messages/context";
import { SkillTransferSelector } from "@/components/workspace/agents/skill-transfer-selector";
import { SkillDetailDialog } from "@/components/workspace/skills/skill-detail-dialog";
import { getAPIClient } from "@/core/api";
import type { Agent } from "@/core/agents";
@@ -51,7 +50,7 @@ import { getStoredAuth } from "@/core/auth";
import { useLiveAccount } from "@/core/auth/use-live-account";
import { useI18n } from "@/core/i18n/hooks";
import { useModels } from "@/core/models/hooks";
import { isAgentSkillCandidate, type Skill } from "@/core/skills";
import { isAgentSkillCandidate } from "@/core/skills";
import { useSkills } from "@/core/skills/hooks";
import { useThreadStream } from "@/core/threads/hooks";
import { uuid } from "@/core/utils/uuid";
@@ -98,7 +97,6 @@ export default function NewAgentPage() {
const [selectedSkills, setSelectedSkills] = useState<string[]>([]);
const [isSkillSelectionTouched, setIsSkillSelectionTouched] = useState(false);
const [detailSkillName, setDetailSkillName] = useState<string | null>(null);
const [skillSearch, setSkillSearch] = useState("");
// 恢复已有 bootstrap 会话时该 thread 已存在，无需再 create（置为 true，跳过创建）。
const [threadPrepared, setThreadPrepared] = useState(!!resumeThreadId);
const { skills, isLoading: isSkillsLoading } = useSkills();
@@ -123,17 +121,6 @@ export default function NewAgentPage() {
[skills, storedAuth],
);
const filteredSkills = useMemo(() => {
const q = skillSearch.trim().toLowerCase();
if (!q) return enabledSkills;
return enabledSkills.filter(
(skill) =>
skill.name.toLowerCase().includes(q) ||
(skill.name_zh ?? "").toLowerCase().includes(q) ||
(skill.description ?? "").toLowerCase().includes(q),
);
}, [enabledSkills, skillSearch]);
useEffect(() => {
if (isSkillSelectionTouched) {
return;
@@ -380,14 +367,10 @@ export default function NewAgentPage() {
[handleChatSubmit],
);
const toggleSkill = useCallback((skill: Skill) => {
const handleSkillSelectionChange = useCallback((next: string[]) => {
setIsSkillSelectionTouched(true);
setFormError("");
setSelectedSkills((prev) =>
prev.includes(skill.name)
? prev.filter((name) => name !== skill.name)
: [...prev, skill.name],
);
setSelectedSkills(next);
}, []);
const busy = isCreatingAgent || isSavingDraft || thread.isLoading;
@@ -624,68 +607,13 @@ export default function NewAgentPage() {
已选 {selectedSkills.length} 个
</span>
</div>
{enabledSkills.length > 0 && (
<div className="relative">
<SearchIcon className="text-muted-foreground absolute left-2 top-1/2 size-3.5 -translate-y-1/2" />
<Input
placeholder="搜索技能名称或描述…"
value={skillSearch}
onChange={(e) => setSkillSearch(e.currentTarget.value)}
className="h-8 pl-7 text-xs"
/>
</div>
)}
<div className="max-h-[240px] overflow-y-auto rounded-md border p-2 dark:border-[#1B477E] dark:bg-[#022248]">
{isSkillsLoading ? (
<p className="text-muted-foreground text-sm">技能加载中...</p>
) : enabledSkills.length === 0 ? (
<p className="text-muted-foreground text-sm">
当前没有可用技能。
</p>
) : filteredSkills.length === 0 ? (
<p className="text-muted-foreground text-sm">
没有匹配的技能。
</p>
) : (
<div className="grid grid-cols-1 gap-2 sm:grid-cols-2">
{filteredSkills.map((skill) => {
const checked = selectedSkills.includes(skill.name);
return (
<div
key={skill.name}
className="flex w-full min-w-0 items-center"
>
<Button
type="button"
size="sm"
variant={checked ? "default" : "outline"}
onClick={() => toggleSkill(skill)}
title={skill.description}
className={cn(
"h-8 min-w-0 flex-1 rounded-r-none",
!checked && "text-muted-foreground",
)}
>
<span className="truncate">
{skill.name_zh?.trim() ? skill.name_zh : skill.name}
</span>
</Button>
<Button
type="button"
size="icon"
variant={checked ? "default" : "outline"}
onClick={() => setDetailSkillName(skill.name)}
title={t.settings.skills.viewDetailAction}
className="h-8 w-7 rounded-l-none border-l-0"
>
<InfoIcon className="size-3.5" />
</Button>
</div>
);
})}
</div>
)}
</div>
<SkillTransferSelector
skills={enabledSkills}
selected={selectedSkills}
onChange={handleSkillSelectionChange}
isLoading={isSkillsLoading}
onViewDetail={setDetailSkillName}
/>
</div>
<SkillDetailDialog
frontend-web/src/pages/PublicScheduledTaskPage.tsx
+226 -13
View File
@@ -2,6 +2,7 @@
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
import {
BookOpen,
Check,
ChevronLeft,
ChevronRight,
@@ -23,6 +24,11 @@ import { toast } from "sonner";
import { Badge } from "@/components/ui/badge";
import { Button } from "@/components/ui/button";
import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card";
import {
Collapsible,
CollapsibleContent,
CollapsibleTrigger,
} from "@/components/ui/collapsible";
import {
Dialog,
DialogContent,
@@ -42,10 +48,12 @@ import {
} from "@/components/ui/tooltip";
import { MarkdownContent } from "@/components/workspace/messages/markdown-content";
import { HtmlPreviewFrame } from "@/components/html-preview-frame";
import { ReportMarkdownView } from "@/components/wiki/ReportMarkdownView";
import { getBackendBaseURL } from "@/core/config";
import { usePublicBackground } from "@/core/embed/background";
import { downloadMarkdownAsDocx } from "@/core/scheduled-tasks/markdown-docx";
import { reasoningPlugins } from "@/core/streamdown";
import { createMarkdownAnchorResolver } from "@/core/utils/markdown";
import { cn } from "@/lib/utils";
type PublicTask = {
@@ -264,6 +272,190 @@ function stripTimestampSuffix(title: string): string {
return cleaned.trim() || title;
}
function stripMarkdownToText(markdown: string): string {
return markdown
.replace(/```[\s\S]*?```/g, " ")
.replace(/~~~[\s\S]*?~~~/g, " ")
.replace(/!\[[^\]]*\]\([^)]+\)/g, " ")
.replace(/\[([^\]]+)\]\([^)]+\)/g, "$1")
.replace(/^#{1,6}\s+/gm, "")
.replace(/^[>\-*+\d.)\s]+/gm, "")
.replace(/[`*_~|]/g, "")
.replace(/<[^>]+>/g, " ")
.replace(/\s+/g, " ")
.trim();
}
function getMarkdownExcerpt(markdown: string): string {
const text = stripMarkdownToText(markdown);
if (!text) return "";
return text.length > 180 ? `${text.slice(0, 180)}...` : text;
}
function countReadableWords(markdown: string): number {
const text = stripMarkdownToText(markdown);
if (!text) return 0;
const cjk = text.match(/[\u4e00-\u9fff]/g)?.length ?? 0;
const words = text.replace(/[\u4e00-\u9fff]/g, " ").match(/[A-Za-z0-9]+/g)?.length ?? 0;
return cjk + words;
}
function selectorForId(id: string): string {
const css = window.CSS;
if (css?.escape) return `#${css.escape(id)}`;
return `[id="${id.replace(/\\/g, "\\\\").replace(/"/g, '\\"')}"]`;
}
function KnowledgeMarkdownView({
content,
title,
subtitle,
className,
}: {
content: string;
title?: string;
subtitle?: string;
className?: string;
isLoading?: boolean;
remarkPlugins?: unknown;
rehypePlugins?: unknown;
}) {
const rootRef = useRef<HTMLDivElement>(null);
const anchorResolver = useMemo(() => createMarkdownAnchorResolver(content), [content]);
const headings = useMemo(
() => anchorResolver.headings.filter((item) => item.level >= 1 && item.level <= 3),
[anchorResolver],
);
const [activeHeadingId, setActiveHeadingId] = useState<string | null>(null);
const excerpt = useMemo(() => getMarkdownExcerpt(content), [content]);
const sectionCount = headings.length;
const wordCount = useMemo(() => countReadableWords(content), [content]);
useEffect(() => {
setActiveHeadingId(headings[0]?.id ?? null);
}, [headings]);
useEffect(() => {
const root = rootRef.current;
if (!root || headings.length === 0) return;
const targets = headings
.map((item) => root.querySelector<HTMLElement>(selectorForId(item.id)))
.filter((item): item is HTMLElement => !!item);
if (targets.length === 0) return;
const observer = new IntersectionObserver(
(entries) => {
const visible = entries
.filter((entry) => entry.isIntersecting)
.sort((a, b) => a.boundingClientRect.top - b.boundingClientRect.top)[0];
if (visible?.target.id) {
setActiveHeadingId(visible.target.id);
}
},
{
root: root.closest("[data-radix-scroll-area-viewport]") ?? null,
rootMargin: "-96px 0px -65% 0px",
threshold: [0, 1],
},
);
targets.forEach((target) => observer.observe(target));
return () => observer.disconnect();
}, [headings, content]);
const scrollToHeading = useCallback((id: string) => {
const root = rootRef.current;
const target =
root?.querySelector<HTMLElement>(selectorForId(id)) ?? document.getElementById(id);
setActiveHeadingId(id);
target?.scrollIntoView({ behavior: "smooth", block: "start" });
}, []);
return (
<div
ref={rootRef}
className={cn(
"grid min-h-0 gap-5 lg:grid-cols-[minmax(0,1fr)_300px]",
className,
)}
>
<article className="min-w-0 overflow-hidden rounded-lg border border-slate-200/80 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-950">
<div className="border-b border-slate-100 bg-slate-50/90 px-6 py-5 dark:border-slate-800 dark:bg-slate-900/70 md:px-8">
<div className="flex flex-wrap items-center gap-2 text-xs text-slate-500 dark:text-slate-400">
<span className="inline-flex items-center gap-1 rounded-md bg-white px-2 py-1 font-medium text-slate-600 ring-1 ring-slate-200 dark:bg-slate-950 dark:text-slate-300 dark:ring-slate-800">
<BookOpen className="size-3.5" />
知识梳理
</span>
{subtitle && <span className="truncate">{subtitle}</span>}
</div>
{title && (
<h1 className="mt-3 text-xl font-semibold leading-snug text-slate-950 dark:text-slate-50 md:text-2xl">
{title}
</h1>
)}
{excerpt && (
<p className="mt-3 max-w-4xl text-sm leading-7 text-slate-600 dark:text-slate-300">
{excerpt}
</p>
)}
<div className="mt-4 flex flex-wrap gap-2 text-xs text-slate-500 dark:text-slate-400">
<span className="rounded-md bg-white px-2 py-1 ring-1 ring-slate-200 dark:bg-slate-950 dark:ring-slate-800">
{sectionCount || 1} 个章节
</span>
<span className="rounded-md bg-white px-2 py-1 ring-1 ring-slate-200 dark:bg-slate-950 dark:ring-slate-800">
约 {wordCount} 字
</span>
</div>
</div>
<div className="wiki-markdown public-report-markdown mx-auto w-full max-w-[980px] px-6 py-7 md:px-10 md:py-9">
<MarkdownContent
className="max-w-none"
content={content}
isLoading={false}
remarkPlugins={reasoningPlugins.remarkPlugins}
rehypePlugins={reasoningPlugins.rehypePlugins}
/>
</div>
</article>
<aside className="hidden min-w-0 lg:block">
<div className="sticky top-4 rounded-lg border border-slate-200/80 bg-white p-4 shadow-sm dark:border-slate-800 dark:bg-slate-950">
<div className="mb-2 flex items-center gap-2 text-sm font-semibold text-slate-800 dark:text-slate-100">
<List className="size-4" />
内容目录
</div>
{headings.length > 0 ? (
<nav className="max-h-[calc(100vh-210px)] space-y-1 overflow-auto pr-1">
{headings.map((item) => (
<button
key={item.id}
type="button"
className={cn(
"block w-full truncate rounded-md border-l-2 border-transparent px-2 py-1.5 text-left text-xs leading-5 text-slate-600 transition-colors hover:bg-slate-100 hover:text-slate-950 dark:text-slate-300 dark:hover:bg-slate-900 dark:hover:text-slate-50",
activeHeadingId === item.id &&
"border-primary bg-primary/10 font-medium text-primary dark:text-primary",
item.level === 2 && "pl-4",
item.level === 3 && "pl-7 text-slate-500 dark:text-slate-400",
)}
title={item.text}
onClick={() => scrollToHeading(item.id)}
>
{item.text}
</button>
))}
</nav>
) : (
<div className="rounded-md border border-dashed p-3 text-xs text-slate-500 dark:border-slate-800 dark:text-slate-400">
暂无标题目录
</div>
)}
</div>
</aside>
</div>
);
}
const BODY_TAB_VALUE = "__body__";
const HTML_TAB_VALUE = "__html__";
const RUN_LIST_COLLAPSED_WIDTH = 56;
@@ -688,12 +880,10 @@ function PublicFullscreenPreviewContent({ preview }: { preview: PublicFullscreen
}
if (preview.mode === "markdown") {
return (
<MarkdownContent
<ReportMarkdownView
className="max-w-none"
content={preview.content || "无正文内容"}
isLoading={false}
remarkPlugins={reasoningPlugins.remarkPlugins}
rehypePlugins={reasoningPlugins.rehypePlugins}
title={preview.title}
/>
);
}
@@ -701,12 +891,10 @@ function PublicFullscreenPreviewContent({ preview }: { preview: PublicFullscreen
const content = preview.file.content || "无文件内容";
if (isMarkdownFile(preview.file)) {
return (
<MarkdownContent
<ReportMarkdownView
className="max-w-none"
content={content}
isLoading={false}
remarkPlugins={reasoningPlugins.remarkPlugins}
rehypePlugins={reasoningPlugins.rehypePlugins}
title={preview.title}
/>
);
}
@@ -731,6 +919,7 @@ function publicFullscreenHref(taskId: string, runId: string, tab: ContentTab): s
function RunDetailPane({ run, taskId }: { run: PublicRun; taskId: string }) {
const queryClient = useQueryClient();
const [fullscreenPreview, setFullscreenPreview] = useState<PublicFullscreenPreview | null>(null);
const [runMetaOpen, setRunMetaOpen] = useState(false);
// HTML page run: render a sandboxed iframe preview instead of the MD tabs.
const isHtmlRun =
@@ -970,7 +1159,29 @@ function RunDetailPane({ run, taskId }: { run: PublicRun; taskId: string }) {
</TooltipContent>
)}
</Tooltip>
<p className="text-muted-foreground mt-1 text-xs">
<Collapsible open={runMetaOpen} onOpenChange={setRunMetaOpen} className="mt-2">
<CollapsibleTrigger asChild>
<Button
type="button"
variant="ghost"
size="sm"
className="text-muted-foreground hover:text-foreground h-7 px-2 text-xs"
>
<ChevronRight
className={cn("size-3.5 transition-transform", runMetaOpen && "rotate-90")}
/>
运行信息
</Button>
</CollapsibleTrigger>
<CollapsibleContent>
<div className="mt-2 grid gap-1 rounded-md border bg-muted/30 px-3 py-2 text-xs text-muted-foreground sm:grid-cols-3">
<div className="min-w-0 break-all">计划：{formatDate(run.scheduled_for)}</div>
<div>开始：{formatDate(run.started_at)}</div>
<div>完成：{formatDate(run.finished_at)}</div>
</div>
</CollapsibleContent>
</Collapsible>
<p className="hidden">
开始：{formatDate(run.started_at)} · 完成：{formatDate(run.finished_at)}
</p>
</div>
@@ -1122,12 +1333,14 @@ function RunDetailPane({ run, taskId }: { run: PublicRun; taskId: string }) {
无内容。点击「编辑」开始添加。
</div>
) : (
<MarkdownContent
<ReportMarkdownView
className="max-w-none pr-2"
content={activeSource}
isLoading={false}
remarkPlugins={reasoningPlugins.remarkPlugins}
rehypePlugins={reasoningPlugins.rehypePlugins}
title={
activeTabMeta?.kind === "file"
? stripExtension(activeTabMeta.fileName || "")
: run.result?.title
}
/>
)}
</ScrollArea>
frontend-web/src/pages/ScheduledTaskRunDetailPage.tsx
+12 -9
View File
@@ -20,6 +20,7 @@ import {
} from "@/components/workspace/workspace-container";
import { MarkdownContent } from "@/components/workspace/messages/markdown-content";
import { HtmlPreviewFrame } from "@/components/html-preview-frame";
import { ReportMarkdownView } from "@/components/wiki/ReportMarkdownView";
import {
downloadScheduledTaskRun,
downloadScheduledTaskRunFile,
@@ -91,11 +92,12 @@ function RenderPreview({ file }: { file: ScheduledTaskResultFileContent }) {
}
if (isMarkdownFile(file)) {
return (
<MarkdownContent
<ReportMarkdownView
className="max-w-none"
content={file.content}
isLoading={false}
rehypePlugins={[]}
title={file.name.replace(/\.(md|markdown)$/i, "")}
badgeLabel="结果文件"
compactHeader
/>
);
}
@@ -114,11 +116,11 @@ function FullscreenPreviewContent({ preview }: { preview: FullscreenPreview }) {
}
if (preview.mode === "markdown") {
return (
<MarkdownContent
<ReportMarkdownView
className="max-w-none"
content={preview.content || "无正文内容"}
isLoading={false}
rehypePlugins={[]}
title={preview.title}
badgeLabel="正文预览"
/>
);
}
@@ -653,11 +655,12 @@ export default function ScheduledTaskRunDetailPage() {
</Button>
</div>
{result.format === "markdown" || result.format === "mixed" ? (
<MarkdownContent
<ReportMarkdownView
className="max-w-none"
content={result.content || "无正文内容"}
isLoading={false}
rehypePlugins={[]}
title={result.title || "正文预览"}
subtitle={`计划时间：${formatDate(run.scheduled_for)}`}
badgeLabel="任务报告"
/>
) : (
<pre className="whitespace-pre-wrap break-words text-sm">
offline-backend-20260512/.gitattributes
+3
View File
@@ -0,0 +1,3 @@
*.sh text eol=lf
offline-runtime-requirements.txt text eol=lf
wheelhouse/SHA256SUMS text eol=lf
offline-backend-20260512/README.md
+66
View File
@@ -0,0 +1,66 @@
# DeerFlow 离线后端部署补充
本目录提供与原始 `start_backend.sh` 配套的 wheelhouse。启动脚本保持原有结构，只新增一行 Redis Python 客户端的离线安装命令。
## 文件
- `offline-runtime-requirements.txt`：固定版本的完整依赖闭包。
- `prepare_wheelhouse.sh`：联网机器下载 Linux/Python 3.12 wheel，并验证传递依赖完整性。
- `start_backend.sh`：原始 Docker 启动脚本，加一行 Redis 缺包安装命令。
- `wheelhouse/`：22 个 wheel 和 `SHA256SUMS`，需要随发布目录一起复制。
## 联网机器准备
默认目标是后端镜像使用的 CPython 3.12、Linux x86_64：
```bash
cd offline-backend-20260512
bash prepare_wheelhouse.sh
```
直接依赖为：
- `croniter==6.2.2`
- `hindsight-client==0.6.2`
- `redis==5.2.1`
上述依赖在 Python 3.12/Linux x86_64 下的 19 个传递依赖也已锁定在 `offline-runtime-requirements.txt`。
国内镜像示例：
```bash
PIP_INDEX_URL=https://pypi.tuna.tsinghua.edu.cn/simple bash prepare_wheelhouse.sh
```
ARM64 镜像需要改目标平台：
```bash
TARGET_PLATFORM=manylinux2014_aarch64 bash prepare_wheelhouse.sh
```
## 离线机器启动
保持原脚本要求的发布目录结构，包括 Docker 镜像 tar、`.env`、`config.yaml`、`extensions_config.json`、`skill_evolution.json`、`skills/`、`wheelhouse/`、`backend/uv.lock` 和 `backend/`。
可先校验 wheel：
```bash
cd offline-backend-20260512/wheelhouse
sha256sum -c SHA256SUMS
cd ..
```
然后启动：
```bash
bash start_backend.sh
docker logs -f deerflow-backend
```
原脚本已有 `croniter` 和 `hindsight-client` 的离线补装逻辑；新增的一行命令只负责在缺少 `redis` 模块时，从 `/app/wheelhouse` 安装 `redis==5.2.1`。
## Redis 说明
`redis==5.2.1` 只是 Python 客户端，不包含 Redis 服务端。`config.yaml` 中 `concurrency.redis.enabled: false` 时无需部署 Redis；启用后必须另行准备 Redis 服务或离线 Redis Docker 镜像。
如果 Redis 运行在另一个容器，`redis://127.0.0.1:6379/0` 指向后端容器自身，不能直接使用。应将两个容器加入同一个 Docker network，并将 URL 改为类似 `redis://redis:6379/0`，同时为不同环境配置不同的 `namespace`。
offline-backend-20260512/backend/CLAUDE.md
+1
View File
@@ -1085,6 +1085,7 @@ Both the `view_image` tool and `ViewImageMiddleware` are registered whenever `vi
## Documentation
See `docs/` directory for detailed documentation:
- [../README.md](../README.md) - Offline Docker wheelhouse preparation and startup workflow. The pinned versions in `../offline-runtime-requirements.txt` and `../start_backend.sh` must stay synchronized. `start_backend.sh` stays aligned with the deployment script and adds only the Redis wheel fallback line. The download target defaults to CPython 3.12 on Linux x86_64 and verifies dependency closure using wheelhouse only.
- [CONFIGURATION.md](docs/CONFIGURATION.md) - Configuration options
- [ARCHITECTURE.md](docs/ARCHITECTURE.md) - Architecture details
- [API.md](docs/API.md) - API reference
offline-backend-20260512/backend/README.md
+1
View File
@@ -446,6 +446,7 @@ remain the system default and automatically fall back to `light`.
- [Configuration Guide](docs/CONFIGURATION.md)
- [Architecture Details](docs/ARCHITECTURE.md)
- [API Reference](docs/API.md)
- [Offline Docker wheelhouse deployment (ZH)](../README.md)
- [File Upload](docs/FILE_UPLOAD.md)
- [Path Examples](docs/PATH_EXAMPLES.md)
- [Context Summarization](docs/summarization.md)
offline-backend-20260512/backend/app/gateway/app.py
+13 -7
View File
@@ -857,8 +857,9 @@ async def _concurrency_sampler_loop(app: FastAPI) -> None:
logger.info("Concurrency sampler disabled by CONCURRENCY_SAMPLE_ENABLED")
return
store = getattr(app.state, "concurrency_sample_store", None)
run_mgr = getattr(app.state, "run_manager", None)
store = getattr(app.state, "concurrency_sample_store", None)
run_mgr = getattr(app.state, "run_manager", None)
concurrency_gate = getattr(app.state, "concurrency_gate", None)
if store is None or run_mgr is None:
logger.info("Concurrency sampler not started (no store / run manager — memory backend?)")
return
@@ -879,11 +880,16 @@ async def _concurrency_sampler_loop(app: FastAPI) -> None:
try:
# Runtime toggle (admin can turn sampling off without a restart).
# mtime-cached read, so this is cheap every tick.
from deerflow.config.system_settings import get_concurrency_monitor_settings
if get_concurrency_monitor_settings().enabled:
active = await run_mgr.list_active()
await store.record(len(active))
from deerflow.config.system_settings import get_concurrency_monitor_settings
if get_concurrency_monitor_settings().enabled:
active_count = None
if concurrency_gate is not None:
active_count = await concurrency_gate.count_active()
if active_count is None:
active = await run_mgr.list_active()
active_count = len(active)
await store.record(active_count)
ticks += 1
if ticks % purge_every_ticks == 0:
from datetime import UTC, datetime, timedelta
offline-backend-20260512/backend/app/gateway/deps.py
+15 -4
View File
@@ -381,8 +381,11 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
run_events_config = getattr(config, "run_events", None)
app.state.run_event_store = make_run_event_store(run_events_config)
# RunManager with store backing for persistence
app.state.run_manager = RunManager(store=app.state.run_store)
# RunManager with store backing for persistence
app.state.run_manager = RunManager(store=app.state.run_store)
from deerflow.runtime import make_redis_concurrency_gate
app.state.concurrency_gate = make_redis_concurrency_gate(config)
try:
yield
@@ -396,7 +399,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
await queue.stop()
set_auto_ingest_queue(None)
await close_engine()
await close_engine()
gate = getattr(app.state, "concurrency_gate", None)
if gate is not None:
await gate.aclose()
# ---------------------------------------------------------------------------
@@ -422,7 +428,12 @@ get_run_manager: Callable[[Request], RunManager] = _require("run_manager", "Run
get_checkpointer: Callable[[Request], Checkpointer] = _require("checkpointer", "Checkpointer")
get_run_event_store: Callable[[Request], RunEventStore] = _require("run_event_store", "Run event store")
get_feedback_repo: Callable[[Request], FeedbackRepository] = _require("feedback_repo", "Feedback")
get_run_store: Callable[[Request], RunStore] = _require("run_store", "Run store")
get_run_store: Callable[[Request], RunStore] = _require("run_store", "Run store")
def get_concurrency_gate(request: Request):
"""Return the optional cross-worker concurrency gate."""
return getattr(request.app.state, "concurrency_gate", None)
def get_store(request: Request):
offline-backend-20260512/backend/app/gateway/roundtable_inprocess_gateway.py
+8 -1
View File
@@ -32,6 +32,7 @@ from app.gateway.roundtable_model_fallback import (
)
from app.gateway.roundtable_run_policy import roundtable_run_policy
from app.gateway.roundtable_seat_skills import append_seat_skill_directive
from app.gateway.services import RUN_SCOPE_BACKGROUND_CHILD
from app.gateway.routers._roundtable_seed import (
COORDINATOR_AGENT_ID,
DASHBOARD_AGENT_ID,
@@ -346,6 +347,7 @@ async def _run_agent_turn(
body = RunCreateRequest(
assistant_id="lead_agent",
input={"messages": [{"type": "human", "content": [{"type": "text", "text": message}], "additional_kwargs": {}}]},
metadata={"run_scope": RUN_SCOPE_BACKGROUND_CHILD},
config={"recursion_limit": 1000},
context={
"agent_name": agent_name,
@@ -374,7 +376,12 @@ async def _run_agent_turn(
# 落 checkpoint）。额度满时本轮在此 await 排队，等同用户某个在跑轮次释放额度后再开。
async with _user_background_semaphore(user_id):
try:
record = await start_run(body, thread_id, request) # type: ignore[arg-type]
record = await start_run( # type: ignore[arg-type]
body,
thread_id,
request,
run_scope=RUN_SCOPE_BACKGROUND_CHILD,
)
except Exception as exc: # noqa: BLE001 — start_run 报错（409 抢占 / DB 抖动等）也要落诊断
if diag is not None:
await diag.record(
offline-backend-20260512/backend/app/gateway/routers/admin_active_runs.py
+44 -3
View File
@@ -22,7 +22,7 @@ from fastapi import APIRouter, Query, Request
from pydantic import BaseModel
from sqlalchemy import select
from app.gateway.deps import get_concurrency_sample_store, get_run_manager
from app.gateway.deps import get_concurrency_gate, get_concurrency_sample_store, get_run_manager
from app.gateway.routers.admin_users import _require_admin
from deerflow.config.system_settings import load_system_settings, save_system_settings
from deerflow.persistence.agents.model import AgentRow
@@ -149,6 +149,15 @@ def _elapsed_seconds(created_at: str, now: datetime) -> int:
return max(0, int(delta))
def _elapsed_seconds_from_epoch(created_at: float, now: datetime) -> int:
try:
if created_at <= 0:
return 0
return max(0, int(now.timestamp() - float(created_at)))
except Exception:
return 0
def _kind_for(metadata: dict | None) -> str:
"""Coarse human label for the run's origin, from its thread metadata."""
meta = metadata or {}
@@ -218,14 +227,46 @@ async def list_active_runs(request: Request) -> ActiveRunsResponse:
"""List every in-flight run with who/what is driving it (admin only)."""
await _require_admin(request)
run_mgr = get_run_manager(request)
records = await run_mgr.list_active()
gate = get_concurrency_gate(request)
leases = await gate.list_active() if gate is not None else None
records = [] if leases is not None else await run_mgr.list_active()
now = datetime.now(UTC)
maps = await _enrich(records)
enrich_source = leases if leases is not None else records
maps = await _enrich(enrich_source)
thread_meta = maps["thread_meta"]
user_emails = maps["user_emails"]
agent_names = maps["agent_names"]
items: list[ActiveRunItem] = []
if leases is not None:
for r in leases:
meta_row = thread_meta.get(r.thread_id)
user_id = r.user_id or (meta_row.user_id if meta_row else "") or ""
metadata = (meta_row.metadata_json if meta_row else None) or {}
assistant_id = r.assistant_id or ""
if assistant_id in _DEFAULT_ASSISTANT_IDS:
agent_name = "榛樿鍔╂墜"
else:
agent_name = agent_names.get(assistant_id) or assistant_id
items.append(
ActiveRunItem(
run_id=r.run_id,
thread_id=r.thread_id,
status="running",
created_at=datetime.fromtimestamp(r.created_at, tz=UTC).isoformat() if r.created_at else "",
elapsed_seconds=_elapsed_seconds_from_epoch(r.created_at, now),
user_id=user_id,
user_name=user_emails.get(user_id) or user_id or "鏈煡鐢ㄦ埛",
agent_id=assistant_id,
agent_name=agent_name,
model_name=r.model_name,
thread_title=(meta_row.display_name if meta_row else "") or "",
kind=_kind_for(metadata),
multitask_strategy="reject",
)
)
return ActiveRunsResponse(count=len(items), server_time=now.isoformat(), runs=items)
for r in records:
meta_row = thread_meta.get(r.thread_id)
user_id = (meta_row.user_id if meta_row else "") or ""
offline-backend-20260512/backend/app/gateway/routers/multi_agent.py
+7 -1
View File
@@ -66,6 +66,7 @@ from langgraph.checkpoint.base import empty_checkpoint
from app.gateway.roundtable_diag import record_foreground_diag
from app.gateway.roundtable_model_fallback import fallback_model_chain, is_llm_error_message, reason_text
from app.gateway.roundtable_run_policy import merge_skill_stop_names, roundtable_run_policy
from app.gateway.services import RUN_SCOPE_ORCHESTRATION_CHILD, internal_run_scope_headers
from app.gateway.roundtable_seat_skills import append_seat_skill_directive
from app.gateway.routers._roundtable_seed import (
ACTION_PLAN_AGENT_ID,
@@ -328,7 +329,11 @@ def _auth_headers(request: Request) -> dict[str, str]:
token = request.headers.get("authorization")
if not token:
raise HTTPException(status_code=401, detail="Missing Authorization header")
return {"Authorization": token, "Content-Type": "application/json"}
return {
"Authorization": token,
"Content-Type": "application/json",
**internal_run_scope_headers(),
}
async def _create_thread(
@@ -697,6 +702,7 @@ def _run_payload(
}
]
},
"metadata": {"run_scope": RUN_SCOPE_ORCHESTRATION_CHILD},
"config": {"recursion_limit": 1000},
"context": {
"agent_name": agent_name,
offline-backend-20260512/backend/app/gateway/routers/system_settings.py
+34
View File
@@ -221,12 +221,16 @@ class ModelConcurrencyResponse(BaseModel):
enabled: bool = Field(..., description="Whether per-model concurrency limiting is active")
default_limit: int = Field(..., description="Fallback max concurrent runs for models without an explicit override")
model_limits: dict[str, int] = Field(default_factory=dict, description="Per-model overrides: model_name -> max concurrent runs")
user_default_limit: int = Field(..., description="Fallback max concurrent runs per user and model. 0 disables the per-user fallback")
user_model_limits: dict[str, int] = Field(default_factory=dict, description="Per-user per-model overrides: model_name -> max concurrent runs for one user")
class ModelConcurrencyUpdateRequest(BaseModel):
enabled: bool | None = None
default_limit: int | None = Field(default=None, ge=1, le=10_000)
model_limits: dict[str, int] | None = None
user_default_limit: int | None = Field(default=None, ge=0, le=10_000)
user_model_limits: dict[str, int] | None = None
def _skill_compression_to_response(settings: SkillCompressionSettings) -> SkillCompressionResponse:
@@ -250,10 +254,24 @@ def _model_concurrency_to_response(settings: ModelConcurrencySettings) -> ModelC
if normalized_limit < 1:
continue
cleaned[model_name] = min(normalized_limit, 10_000)
cleaned_user: dict[str, int] = {}
for name, limit in (settings.user_model_limits or {}).items():
model_name = str(name).strip()
if not model_name:
continue
try:
normalized_limit = int(limit)
except (TypeError, ValueError):
continue
if normalized_limit < 1:
continue
cleaned_user[model_name] = min(normalized_limit, 10_000)
return ModelConcurrencyResponse(
enabled=settings.enabled,
default_limit=max(1, min(int(settings.default_limit), 10_000)),
model_limits=cleaned,
user_default_limit=max(0, min(int(getattr(settings, "user_default_limit", 0) or 0), 10_000)),
user_model_limits=cleaned_user,
)
@@ -331,6 +349,22 @@ async def update_model_concurrency(
continue
cleaned[model_name] = min(normalized_limit, 10_000)
update["model_limits"] = cleaned
if "user_default_limit" in fields and fields["user_default_limit"] is not None:
update["user_default_limit"] = max(0, min(int(fields["user_default_limit"]), 10_000))
if "user_model_limits" in fields and fields["user_model_limits"] is not None:
cleaned_user: dict[str, int] = {}
for name, limit in (body.user_model_limits or {}).items():
model_name = str(name).strip()
if not model_name:
continue
try:
normalized_limit = int(limit)
except (TypeError, ValueError):
continue
if normalized_limit < 1:
continue
cleaned_user[model_name] = min(normalized_limit, 10_000)
update["user_model_limits"] = cleaned_user
settings = current.model_concurrency.model_copy(update=update)
saved = save_system_settings(current.model_copy(update={"model_concurrency": settings}))
return _model_concurrency_to_response(saved.model_concurrency)
offline-backend-20260512/backend/app/gateway/services.py
+265 -22
View File
@@ -10,15 +10,17 @@ from __future__ import annotations
import asyncio
import json
import logging
import os
import re
import time
import uuid
from collections.abc import Mapping
from typing import Any
from fastapi import HTTPException, Request
from langchain_core.messages import HumanMessage
from app.gateway.deps import get_run_context, get_run_manager, get_stream_bridge
from app.gateway.deps import get_concurrency_gate, get_run_context, get_run_manager, get_stream_bridge
from app.gateway.utils import sanitize_log_param
from deerflow.config import get_app_config
from deerflow.config.agents_config import load_agent_config, validate_agent_id
@@ -36,9 +38,105 @@ from deerflow.runtime import (
UnsupportedStrategyError,
run_agent,
)
from deerflow.runtime.user_context import get_effective_user_id
logger = logging.getLogger(__name__)
RUN_SCOPE_EXTERNAL_ROOT = "external_root"
RUN_SCOPE_ORCHESTRATION_CHILD = "orchestration_child"
RUN_SCOPE_BACKGROUND_CHILD = "background_child"
_INTERNAL_RUN_SCOPES = {RUN_SCOPE_ORCHESTRATION_CHILD, RUN_SCOPE_BACKGROUND_CHILD}
_ALL_RUN_SCOPES = {RUN_SCOPE_EXTERNAL_ROOT, *_INTERNAL_RUN_SCOPES}
_INTERNAL_RUN_SCOPE_HEADER = "x-deerflow-internal-run-scope-token"
_INTERNAL_RUN_SCOPE_TOKEN = (
os.getenv("DEERFLOW_INTERNAL_RUN_SCOPE_TOKEN", "").strip()
or uuid.uuid4().hex
)
def internal_run_scope_headers() -> dict[str, str]:
"""Headers used by same-process loopback calls to mark internal child runs."""
return {_INTERNAL_RUN_SCOPE_HEADER: _INTERNAL_RUN_SCOPE_TOKEN}
def _metadata_run_scope(body: Any) -> str | None:
metadata = getattr(body, "metadata", None) or {}
if not isinstance(metadata, dict):
return None
raw = str(metadata.get("run_scope") or "").strip()
return raw or None
def _request_is_loopback(request: Request) -> bool:
client = getattr(request, "client", None)
host = str(getattr(client, "host", "") or "").strip()
return host in {"127.0.0.1", "::1", "localhost"}
def _has_internal_run_scope_header(request: Request) -> bool:
headers = getattr(request, "headers", None)
token = headers.get(_INTERNAL_RUN_SCOPE_HEADER) if headers is not None else None
return str(token or "") == _INTERNAL_RUN_SCOPE_TOKEN
def _resolve_run_scope(body: Any, request: Request, run_scope: str | None) -> str:
"""Return the trusted run scope used for concurrency accounting."""
if run_scope:
scope = str(run_scope).strip()
return scope if scope in _ALL_RUN_SCOPES else RUN_SCOPE_EXTERNAL_ROOT
scope = _metadata_run_scope(body)
if (
scope in _INTERNAL_RUN_SCOPES
and _request_is_loopback(request)
and _has_internal_run_scope_header(request)
):
return scope
if scope == RUN_SCOPE_EXTERNAL_ROOT:
return RUN_SCOPE_EXTERNAL_ROOT
return RUN_SCOPE_EXTERNAL_ROOT
async def _run_agent_with_concurrency_lease(
*,
concurrency_gate: Any | None,
lease_acquired: bool,
lease_metadata: dict[str, Any],
heartbeat_interval_seconds: int,
run_agent_kwargs: dict[str, Any],
) -> None:
"""Run the agent while refreshing and finally releasing an optional lease."""
heartbeat_task: asyncio.Task | None = None
async def _heartbeat_loop() -> None:
if concurrency_gate is None or not lease_acquired:
return
interval = max(5, int(heartbeat_interval_seconds or 20))
while True:
await asyncio.sleep(interval)
await concurrency_gate.heartbeat(**lease_metadata)
if concurrency_gate is not None and lease_acquired:
heartbeat_task = asyncio.create_task(
_heartbeat_loop(),
name=f"concurrency-heartbeat:{lease_metadata.get('run_id')}",
)
try:
await run_agent(**run_agent_kwargs)
finally:
if heartbeat_task is not None:
heartbeat_task.cancel()
try:
await heartbeat_task
except asyncio.CancelledError:
pass
if concurrency_gate is not None and lease_acquired:
await concurrency_gate.release(**lease_metadata)
# ---------------------------------------------------------------------------
# SSE formatting
@@ -506,6 +604,49 @@ def _resolve_effective_run_model_name(body: Any) -> str:
return default_model_name
def _resolve_effective_model_name(body: Any, app_config: Any | None = None) -> str:
"""Compatibility wrapper for resolving the effective run model."""
if app_config is None:
return _resolve_effective_run_model_name(body)
default_model_name = app_config.models[0].name if app_config.models else None
if not default_model_name:
raise ValueError("No chat models are configured. Please configure at least one model in config.yaml.")
requested_model_name = _extract_requested_model_name(body)
if requested_model_name and app_config.get_model_config(requested_model_name):
return requested_model_name
agent_id = _extract_effective_agent_id(body)
if agent_id:
try:
agent_config = load_agent_config(agent_id)
except Exception:
agent_config = None
agent_model_name = str(getattr(agent_config, "model", "") or "").strip()
if agent_model_name and app_config.get_model_config(agent_model_name):
return agent_model_name
return default_model_name
def _resolve_model_concurrency_limit(model_name: str, app_config: Any) -> int | None:
"""Read legacy per-model max-concurrency values from model config extras."""
model_config = app_config.get_model_config(model_name)
extras = getattr(model_config, "model_extra", None) or {}
for key in ("max_concurrency", "model_concurrency_limit"):
raw = extras.get(key)
if raw is None:
continue
try:
limit = int(raw)
except (TypeError, ValueError):
continue
if limit > 0:
return limit
return None
def _get_model_concurrency_limit(model_name: str) -> int | None:
"""Return the configured limit for a model, or ``None`` when limiting is off."""
settings = load_system_settings().model_concurrency
@@ -520,6 +661,25 @@ def _get_model_concurrency_limit(model_name: str) -> int | None:
return max(1, limit)
def _get_user_model_concurrency_limit(model_name: str) -> int | None:
"""Return the configured per-user limit for a model, or ``None`` when off."""
settings = load_system_settings().model_concurrency
if not settings.enabled:
return None
raw_limit = settings.user_model_limits.get(
model_name,
getattr(settings, "user_default_limit", 0) or 0,
)
try:
limit = int(raw_limit)
except (TypeError, ValueError):
limit = int(getattr(settings, "user_default_limit", 0) or 0)
if limit < 1:
return None
return max(1, limit)
async def _sync_legacy_agent_records(store) -> None:
"""Import existing folder-backed agents as built-in records.
@@ -655,6 +815,7 @@ async def start_run(
request: Request,
*,
enforce_access: bool = True,
run_scope: str | None = None,
) -> RunRecord:
"""Create a RunRecord and launch the background agent task.
@@ -682,6 +843,7 @@ async def start_run(
bridge = get_stream_bridge(request)
run_mgr = get_run_manager(request)
run_ctx = get_run_context(request)
concurrency_gate = get_concurrency_gate(request)
disconnect = DisconnectMode.cancel if body.on_disconnect == "cancel" else DisconnectMode.continue_
@@ -696,10 +858,56 @@ async def start_run(
t_post_enforce = time.perf_counter()
effective_model_name = _resolve_effective_run_model_name(body)
effective_user_id = get_effective_user_id()
effective_run_scope = _resolve_run_scope(body, request, run_scope)
model_concurrency_limit = _get_model_concurrency_limit(effective_model_name)
user_model_concurrency_limit = (
_get_user_model_concurrency_limit(effective_model_name)
if effective_run_scope == RUN_SCOPE_EXTERNAL_ROOT
else None
)
run_id = str(uuid.uuid4())
run_metadata = dict(body.metadata or {})
run_metadata["model_name"] = effective_model_name
run_metadata["user_id"] = effective_user_id
run_metadata["run_scope"] = effective_run_scope
redis_lease_acquired = False
redis_gate_degraded = True
user_model_accounted = user_model_concurrency_limit is not None
lease_metadata = {
"run_id": run_id,
"thread_id": thread_id,
"model_name": effective_model_name,
"user_id": effective_user_id,
"user_model_accounted": user_model_accounted,
}
if concurrency_gate is not None:
acquire = await concurrency_gate.try_acquire(
run_id=run_id,
thread_id=thread_id,
model_name=effective_model_name,
user_id=effective_user_id,
assistant_id=body.assistant_id or "",
model_limit=model_concurrency_limit,
user_model_limit=user_model_concurrency_limit,
reject_thread=body.multitask_strategy == "reject",
)
redis_lease_acquired = acquire.acquired
redis_gate_degraded = acquire.degraded
if acquire.rejected:
if acquire.reason == "thread":
raise HTTPException(
status_code=409,
detail="Current thread already has an active run. Please wait or stop it before sending another message.",
)
if acquire.reason == "user_model":
raise HTTPException(
status_code=429,
detail="This account has reached the concurrency limit for the selected model. Please try again later.",
)
raise HTTPException(status_code=429, detail="The selected model is busy. Please try again later.")
try:
record = await run_mgr.create_or_reject(
@@ -709,19 +917,35 @@ async def start_run(
metadata=run_metadata,
kwargs={"input": body.input, "config": body.config},
multitask_strategy=body.multitask_strategy,
model_concurrency_limit=(effective_model_name, model_concurrency_limit) if model_concurrency_limit is not None else None,
model_concurrency_limit=(effective_model_name, model_concurrency_limit) if redis_gate_degraded and model_concurrency_limit is not None else None,
user_model_concurrency_limit=(effective_model_name, effective_user_id, user_model_concurrency_limit) if redis_gate_degraded and user_model_concurrency_limit is not None else None,
run_id=run_id,
)
except ModelConcurrencyLimitError as exc:
raise HTTPException(status_code=429, detail="当前使用人数较多，请稍等一下再试。") from exc
if concurrency_gate is not None and redis_lease_acquired:
await concurrency_gate.release(**lease_metadata)
if getattr(exc, "scope", "") == "user_model":
detail = "This account has reached the concurrency limit for the selected model. Please try again later."
else:
detail = "The selected model is busy. Please try again later."
raise HTTPException(status_code=429, detail=detail) from exc
except ConflictError as exc:
if concurrency_gate is not None and redis_lease_acquired:
await concurrency_gate.release(**lease_metadata)
# 同一会话上一轮回答还没结束，又提交了新问题 → 拒绝。给用户中文提示，
# 而不是把内部英文「Thread ... already has an active run」直接抛到前端。
raise HTTPException(
status_code=409,
detail="当前对话还在生成回答，请等待本轮回复结束，或先点「停止」再发送新的问题。",
detail="Current thread already has an active run. Please wait or stop it before sending another message.",
) from exc
except UnsupportedStrategyError as exc:
if concurrency_gate is not None and redis_lease_acquired:
await concurrency_gate.release(**lease_metadata)
raise HTTPException(status_code=501, detail=str(exc)) from exc
except Exception:
if concurrency_gate is not None and redis_lease_acquired:
await concurrency_gate.release(**lease_metadata)
raise
t_post_create = time.perf_counter()
# Upsert thread metadata so the thread appears in /threads/search,
@@ -759,7 +983,12 @@ async def start_run(
logger.warning("Failed to upsert thread_meta for %s (non-fatal)", sanitize_log_param(thread_id))
t_post_thread_upsert = time.perf_counter()
agent_factory = resolve_agent_factory(body.assistant_id)
try:
agent_factory = resolve_agent_factory(body.assistant_id)
except Exception:
if concurrency_gate is not None and redis_lease_acquired:
await concurrency_gate.release(**lease_metadata)
raise
graph_input = normalize_input(body.input)
# Admin-configurable prompt prefix: when the caller opted in via
@@ -797,24 +1026,38 @@ async def start_run(
stream_modes = normalize_stream_modes(body.stream_mode)
t_pre_task = time.perf_counter()
task = asyncio.create_task(
run_agent(
bridge,
run_mgr,
record,
ctx=run_ctx,
agent_factory=agent_factory,
graph_input=graph_input,
config=config,
stream_modes=stream_modes,
stream_subgraphs=body.stream_subgraphs,
interrupt_before=body.interrupt_before,
interrupt_after=body.interrupt_after,
# 透传 SDK ``Command``：resume / update / goto。Pregel 会据此从最近
# checkpoint 处恢复，而不是把 graph_input 当新一轮输入跑。
command=getattr(body, "command", None),
heartbeat_interval_seconds = 20
if concurrency_gate is not None:
heartbeat_interval_seconds = int(
getattr(getattr(concurrency_gate, "config", None), "heartbeat_interval_seconds", 20) or 20
)
)
try:
task = asyncio.create_task(
_run_agent_with_concurrency_lease(
concurrency_gate=concurrency_gate,
lease_acquired=redis_lease_acquired,
lease_metadata=lease_metadata,
heartbeat_interval_seconds=heartbeat_interval_seconds,
run_agent_kwargs={
"bridge": bridge,
"run_manager": run_mgr,
"record": record,
"ctx": run_ctx,
"agent_factory": agent_factory,
"graph_input": graph_input,
"config": config,
"stream_modes": stream_modes,
"stream_subgraphs": body.stream_subgraphs,
"interrupt_before": body.interrupt_before,
"interrupt_after": body.interrupt_after,
"command": getattr(body, "command", None),
},
)
)
except Exception:
if concurrency_gate is not None and redis_lease_acquired:
await concurrency_gate.release(**lease_metadata)
raise
record.task = task
# Title sync is handled by worker.py's finally block which reads the
offline-backend-20260512/backend/config.yaml
+23
View File
@@ -9,6 +9,29 @@ log_level: info
gateway:
port: 8001
# Optional Redis-backed global concurrency coordination.
# Leave disabled to keep the existing in-process behavior.
# When enabled, Redis operations are short-timeout and fail-open by default:
# if Redis is slow or unavailable, user Q&A continues with local-only limits.
concurrency:
redis:
enabled: false
url: redis://127.0.0.1:6379/0
password: null
# Prefix with the deployment/system id because Redis may be shared by
# multiple projects. Keep this unique per system to avoid cross-project
# leases affecting each other.
namespace: cmzs_deerflow:concurrency
socket_connect_timeout_ms: 80
socket_timeout_ms: 80
acquire_timeout_ms: 120
operation_timeout_ms: 80
circuit_breaker_failures: 3
circuit_breaker_cooldown_seconds: 30
lease_ttl_seconds: 90
heartbeat_interval_seconds: 20
fail_open: true
models:
- name: deepseek-chat
display_name: DeepSeek Chat
offline-backend-20260512/backend/packages/harness/deerflow/config/app_config.py
+6 -4
View File
@@ -11,8 +11,9 @@ from pydantic import BaseModel, ConfigDict, Field
from deerflow.config.acp_config import ACPAgentConfig, load_acp_config_from_dict
from deerflow.config.agents_api_config import AgentsApiConfig, load_agents_api_config_from_dict
from deerflow.config.browser_context_config import BrowserContextConfig
from deerflow.config.checkpointer_config import CheckpointerConfig, load_checkpointer_config_from_dict
from deerflow.config.database_config import DatabaseConfig
from deerflow.config.checkpointer_config import CheckpointerConfig, load_checkpointer_config_from_dict
from deerflow.config.concurrency_config import ConcurrencyConfig
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
@@ -137,8 +138,9 @@ class AppConfig(BaseModel):
gateway: GatewayConfig = Field(default_factory=GatewayConfig, description="Gateway FastAPI listening port (used by intra-process loopback)")
run_events: RunEventsConfig = Field(default_factory=RunEventsConfig, description="Run event storage configuration")
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")
stream_bridge: StreamBridgeConfig | None = Field(default=None, description="Stream bridge configuration")
concurrency: ConcurrencyConfig = Field(default_factory=ConcurrencyConfig, description="Optional cross-worker concurrency coordination")
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")
knowledge_ingest: KnowledgeIngestConfig = Field(default_factory=KnowledgeIngestConfig, description="External knowledge-ingest endpoint configuration")
task_deeplink: TaskDeeplinkConfig = Field(default_factory=TaskDeeplinkConfig, description="任务深链(rwfx)目的树闸门与跳转地址配置(前端经 /api/public/config/task-deeplink 读取)")
offline-backend-20260512/backend/packages/harness/deerflow/config/concurrency_config.py
+38
View File
@@ -0,0 +1,38 @@
"""Runtime concurrency coordination configuration."""
from __future__ import annotations
from pydantic import BaseModel, Field
class RedisConcurrencyConfig(BaseModel):
"""Optional Redis-backed cross-worker concurrency coordination."""
enabled: bool = Field(default=False, description="Enable Redis-backed global run concurrency coordination.")
url: str | None = Field(default=None, description="Redis URL, for example redis://127.0.0.1:6379/0.")
password: str | None = Field(
default=None,
description="Optional Redis password. You may also include credentials directly in url.",
)
namespace: str = Field(
default="cmzs_deerflow:concurrency",
description="Redis key prefix for concurrency leases. Use a deployment/system prefix when Redis is shared.",
)
socket_connect_timeout_ms: int = Field(default=80, ge=1, le=10_000)
socket_timeout_ms: int = Field(default=80, ge=1, le=10_000)
acquire_timeout_ms: int = Field(default=120, ge=1, le=10_000)
operation_timeout_ms: int = Field(default=80, ge=1, le=10_000)
circuit_breaker_failures: int = Field(default=3, ge=1, le=100)
circuit_breaker_cooldown_seconds: int = Field(default=30, ge=1, le=3600)
lease_ttl_seconds: int = Field(default=90, ge=10, le=3600)
heartbeat_interval_seconds: int = Field(default=20, ge=5, le=600)
fail_open: bool = Field(
default=True,
description="Allow runs to proceed with local-only limits when Redis is unavailable or slow.",
)
class ConcurrencyConfig(BaseModel):
"""Top-level concurrency coordination config."""
redis: RedisConcurrencyConfig = Field(default_factory=RedisConcurrencyConfig)
offline-backend-20260512/backend/packages/harness/deerflow/config/system_settings.py
+10
View File
@@ -461,6 +461,16 @@ class ModelConcurrencySettings(BaseModel):
default_factory=dict,
description="Optional per-model overrides: model_name -> max concurrent runs.",
)
user_default_limit: int = Field(
default=0,
ge=0,
le=10_000,
description="Fallback max concurrent runs per user and model. 0 disables the per-user fallback.",
)
user_model_limits: dict[str, int] = Field(
default_factory=dict,
description="Optional per-user per-model overrides: model_name -> max concurrent runs for one user.",
)
model_config = ConfigDict(extra="ignore")
offline-backend-20260512/backend/packages/harness/deerflow/runtime/__init__.py
+10
View File
@@ -7,14 +7,19 @@ directly from ``deerflow.runtime``.
from .checkpointer import checkpointer_context, get_checkpointer, make_checkpointer, reset_checkpointer
from .runs import (
AcquireDecision,
AcquireResult,
ActiveLease,
ConflictError,
DisconnectMode,
ModelConcurrencyLimitError,
RedisConcurrencyGate,
RunContext,
RunManager,
RunRecord,
RunStatus,
UnsupportedStrategyError,
make_redis_concurrency_gate,
run_agent,
)
from .serialization import serialize, serialize_channel_values, serialize_lc_object, serialize_messages_tuple
@@ -28,14 +33,19 @@ __all__ = [
"make_checkpointer",
"reset_checkpointer",
# runs
"AcquireDecision",
"AcquireResult",
"ActiveLease",
"ConflictError",
"DisconnectMode",
"ModelConcurrencyLimitError",
"RedisConcurrencyGate",
"RunContext",
"RunManager",
"RunRecord",
"RunStatus",
"UnsupportedStrategyError",
"make_redis_concurrency_gate",
"run_agent",
# serialization
"serialize",
offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/__init__.py
+6
View File
@@ -2,16 +2,22 @@
from .manager import ConflictError, ModelConcurrencyLimitError, RunManager, RunRecord, UnsupportedStrategyError
from .schemas import DisconnectMode, RunStatus
from .concurrency_gate import AcquireDecision, AcquireResult, ActiveLease, RedisConcurrencyGate, make_redis_concurrency_gate
from .worker import RunContext, run_agent
__all__ = [
"AcquireDecision",
"AcquireResult",
"ActiveLease",
"ConflictError",
"DisconnectMode",
"ModelConcurrencyLimitError",
"RedisConcurrencyGate",
"RunContext",
"RunManager",
"RunRecord",
"RunStatus",
"UnsupportedStrategyError",
"make_redis_concurrency_gate",
"run_agent",
]
offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/concurrency_gate.py
+379
View File
@@ -0,0 +1,379 @@
"""Optional cross-worker run concurrency gates.
The Redis gate is deliberately fail-open by default: Redis coordinates global
limits when it is fast and healthy, but it must not block the chat path when it
is slow or unavailable.
"""
from __future__ import annotations
import asyncio
import logging
import time
from dataclasses import dataclass
from enum import Enum
from typing import Any
from deerflow.config.concurrency_config import RedisConcurrencyConfig
logger = logging.getLogger(__name__)
class AcquireDecision(str, Enum):
ACQUIRED = "acquired"
REJECTED = "rejected"
DEGRADED = "degraded"
@dataclass(frozen=True)
class AcquireResult:
decision: AcquireDecision
reason: str = ""
source: str = "redis"
@property
def acquired(self) -> bool:
return self.decision == AcquireDecision.ACQUIRED
@property
def rejected(self) -> bool:
return self.decision == AcquireDecision.REJECTED
@property
def degraded(self) -> bool:
return self.decision == AcquireDecision.DEGRADED
@dataclass(frozen=True)
class ActiveLease:
run_id: str
thread_id: str = ""
model_name: str = ""
user_id: str = ""
assistant_id: str = ""
created_at: float = 0.0
expires_at: float = 0.0
class CircuitBreaker:
"""Tiny in-process circuit breaker for Redis failures."""
def __init__(self, *, failure_threshold: int, cooldown_seconds: int) -> None:
self._failure_threshold = max(1, int(failure_threshold))
self._cooldown_seconds = max(1, int(cooldown_seconds))
self._failures = 0
self._opened_until = 0.0
def allow(self) -> bool:
return time.monotonic() >= self._opened_until
def record_success(self) -> None:
self._failures = 0
self._opened_until = 0.0
def record_failure(self) -> None:
self._failures += 1
if self._failures >= self._failure_threshold:
self._opened_until = time.monotonic() + self._cooldown_seconds
@property
def open(self) -> bool:
return not self.allow()
class RedisConcurrencyGate:
"""Redis lease-based global concurrency limiter."""
_ACQUIRE_SCRIPT = """
local now = tonumber(ARGV[1])
local expires_at = tonumber(ARGV[2])
local run_id = ARGV[3]
local model_limit = tonumber(ARGV[4])
local user_model_limit = tonumber(ARGV[5])
local reject_thread = tonumber(ARGV[6])
local payload = ARGV[7]
local ttl_ms = tonumber(ARGV[8])
for i = 1, 4 do
redis.call('ZREMRANGEBYSCORE', KEYS[i], '-inf', now)
end
if reject_thread == 1 and redis.call('ZCARD', KEYS[4]) > 0 then
return {'rejected', 'thread'}
end
if model_limit > 0 and redis.call('ZCARD', KEYS[2]) >= model_limit then
return {'rejected', 'model'}
end
if user_model_limit > 0 and redis.call('ZCARD', KEYS[3]) >= user_model_limit then
return {'rejected', 'user_model'}
end
redis.call('ZADD', KEYS[1], expires_at, run_id)
redis.call('ZADD', KEYS[2], expires_at, run_id)
if user_model_limit > 0 then
redis.call('ZADD', KEYS[3], expires_at, run_id)
end
redis.call('ZADD', KEYS[4], expires_at, run_id)
redis.call('SET', KEYS[5], payload, 'PX', ttl_ms)
return {'acquired', ''}
"""
def __init__(self, config: RedisConcurrencyConfig) -> None:
self.config = config
self._client: Any | None = None
self._breaker = CircuitBreaker(
failure_threshold=config.circuit_breaker_failures,
cooldown_seconds=config.circuit_breaker_cooldown_seconds,
)
@property
def enabled(self) -> bool:
return bool(self.config.enabled and self.config.url)
def health(self) -> dict[str, Any]:
return {
"enabled": self.enabled,
"source": "redis",
"circuit_open": self._breaker.open,
"fail_open": self.config.fail_open,
}
async def aclose(self) -> None:
client = self._client
self._client = None
if client is not None:
try:
await client.aclose()
except Exception:
logger.debug("Failed to close Redis concurrency client", exc_info=True)
def _ns(self) -> str:
return self.config.namespace.strip().rstrip(":") or "cmzs_deerflow:concurrency"
def _keys(self, *, run_id: str, model_name: str, user_id: str, thread_id: str) -> list[str]:
ns = self._ns()
return [
f"{ns}:all",
f"{ns}:model:{model_name}",
f"{ns}:user_model:{model_name}:{user_id}",
f"{ns}:thread:{thread_id}",
f"{ns}:run:{run_id}",
]
async def _redis(self) -> Any:
if self._client is not None:
return self._client
try:
from redis.asyncio import Redis
except Exception as exc: # pragma: no cover - exercised when dependency is absent
raise RuntimeError("redis package is not installed") from exc
timeout = max(0.001, self.config.socket_timeout_ms / 1000)
connect_timeout = max(0.001, self.config.socket_connect_timeout_ms / 1000)
self._client = Redis.from_url(
self.config.url,
password=self.config.password,
socket_timeout=timeout,
socket_connect_timeout=connect_timeout,
decode_responses=True,
)
return self._client
async def _with_timeout(self, coro, timeout_ms: int) -> Any:
return await asyncio.wait_for(coro, timeout=max(0.001, timeout_ms / 1000))
def _degraded(self, reason: str) -> AcquireResult:
return AcquireResult(AcquireDecision.DEGRADED, reason=reason)
def _on_failure(self, action: str, exc: BaseException) -> None:
self._breaker.record_failure()
logger.warning(
"Redis concurrency %s failed; fail_open=%s circuit_open=%s: %s",
action,
self.config.fail_open,
self._breaker.open,
exc,
)
async def try_acquire(
self,
*,
run_id: str,
thread_id: str,
model_name: str,
user_id: str,
assistant_id: str = "",
model_limit: int | None = None,
user_model_limit: int | None = None,
reject_thread: bool = True,
) -> AcquireResult:
if not self.enabled:
return self._degraded("disabled")
if not self._breaker.allow():
return self._degraded("circuit_open")
now = time.time()
expires_at = now + self.config.lease_ttl_seconds
payload = {
"run_id": run_id,
"thread_id": thread_id,
"model_name": model_name,
"user_id": user_id,
"assistant_id": assistant_id,
"created_at": now,
"expires_at": expires_at,
}
import json
keys = self._keys(run_id=run_id, model_name=model_name, user_id=user_id, thread_id=thread_id)
ttl_ms = int(self.config.lease_ttl_seconds * 1000)
try:
client = await self._redis()
raw = await self._with_timeout(
client.eval(
self._ACQUIRE_SCRIPT,
len(keys),
*keys,
str(now),
str(expires_at),
run_id,
str(max(0, int(model_limit or 0))),
str(max(0, int(user_model_limit or 0))),
"1" if reject_thread else "0",
json.dumps(payload, ensure_ascii=False),
str(ttl_ms),
),
self.config.acquire_timeout_ms,
)
self._breaker.record_success()
except Exception as exc:
self._on_failure("acquire", exc)
if self.config.fail_open:
return self._degraded(type(exc).__name__)
raise
status = raw[0] if isinstance(raw, list) and raw else ""
reason = raw[1] if isinstance(raw, list) and len(raw) > 1 else ""
if status == "acquired":
return AcquireResult(AcquireDecision.ACQUIRED)
return AcquireResult(AcquireDecision.REJECTED, reason=str(reason or "limit"))
async def heartbeat(
self,
*,
run_id: str,
thread_id: str,
model_name: str,
user_id: str,
user_model_accounted: bool = True,
) -> None:
if not self.enabled or not self._breaker.allow():
return
now = time.time()
expires_at = now + self.config.lease_ttl_seconds
keys = self._keys(run_id=run_id, model_name=model_name, user_id=user_id, thread_id=thread_id)
try:
client = await self._redis()
pipe = client.pipeline(transaction=False)
refresh_keys = [keys[0], keys[1], keys[3]]
if user_model_accounted:
refresh_keys.append(keys[2])
for key in refresh_keys:
pipe.zadd(key, {run_id: expires_at})
pipe.expire(keys[4], self.config.lease_ttl_seconds)
await self._with_timeout(pipe.execute(), self.config.operation_timeout_ms)
self._breaker.record_success()
except Exception as exc:
self._on_failure("heartbeat", exc)
async def release(
self,
*,
run_id: str,
thread_id: str,
model_name: str,
user_id: str,
user_model_accounted: bool = True,
) -> None:
if not self.enabled or not self._breaker.allow():
return
keys = self._keys(run_id=run_id, model_name=model_name, user_id=user_id, thread_id=thread_id)
try:
client = await self._redis()
pipe = client.pipeline(transaction=False)
release_keys = [keys[0], keys[1], keys[3]]
if user_model_accounted:
release_keys.append(keys[2])
for key in release_keys:
pipe.zrem(key, run_id)
pipe.delete(keys[4])
await self._with_timeout(pipe.execute(), self.config.operation_timeout_ms)
self._breaker.record_success()
except Exception as exc:
self._on_failure("release", exc)
async def count_active(self) -> int | None:
if not self.enabled or not self._breaker.allow():
return None
key = f"{self._ns()}:all"
now = time.time()
try:
client = await self._redis()
pipe = client.pipeline(transaction=False)
pipe.zremrangebyscore(key, "-inf", now)
pipe.zcard(key)
result = await self._with_timeout(pipe.execute(), self.config.operation_timeout_ms)
self._breaker.record_success()
return int(result[-1] or 0)
except Exception as exc:
self._on_failure("count", exc)
return None
async def list_active(self, *, limit: int = 1000) -> list[ActiveLease] | None:
if not self.enabled or not self._breaker.allow():
return None
import json
all_key = f"{self._ns()}:all"
now = time.time()
try:
client = await self._redis()
await self._with_timeout(client.zremrangebyscore(all_key, "-inf", now), self.config.operation_timeout_ms)
run_ids = await self._with_timeout(client.zrange(all_key, 0, max(0, limit - 1)), self.config.operation_timeout_ms)
if not run_ids:
self._breaker.record_success()
return []
keys = [f"{self._ns()}:run:{run_id}" for run_id in run_ids]
payloads = await self._with_timeout(client.mget(keys), self.config.operation_timeout_ms)
leases: list[ActiveLease] = []
for run_id, payload in zip(run_ids, payloads, strict=False):
if not payload:
continue
try:
data = json.loads(payload)
except Exception:
continue
leases.append(
ActiveLease(
run_id=str(data.get("run_id") or run_id),
thread_id=str(data.get("thread_id") or ""),
model_name=str(data.get("model_name") or ""),
user_id=str(data.get("user_id") or ""),
assistant_id=str(data.get("assistant_id") or ""),
created_at=float(data.get("created_at") or 0),
expires_at=float(data.get("expires_at") or 0),
)
)
self._breaker.record_success()
return leases
except Exception as exc:
self._on_failure("list", exc)
return None
def make_redis_concurrency_gate(config: Any) -> RedisConcurrencyGate | None:
concurrency = getattr(config, "concurrency", None)
redis_config = getattr(concurrency, "redis", None)
if redis_config is None or not getattr(redis_config, "enabled", False) or not getattr(redis_config, "url", None):
return None
return RedisConcurrencyGate(redis_config)
offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py
+20 -3
View File
@@ -218,6 +218,8 @@ class RunManager:
kwargs: dict | None = None,
multitask_strategy: str = "reject",
model_concurrency_limit: tuple[str, int] | None = None,
user_model_concurrency_limit: tuple[str, str, int] | None = None,
run_id: str | None = None,
) -> RunRecord:
"""Atomically check for inflight runs and create a new one.
@@ -228,7 +230,7 @@ class RunManager:
This method holds the lock across both the check and the insert,
eliminating the TOCTOU race in separate ``has_inflight`` + ``create``.
"""
run_id = str(uuid.uuid4())
run_id = run_id or str(uuid.uuid4())
now = _now_iso()
_supported_strategies = ("reject", "interrupt", "rollback")
@@ -261,7 +263,20 @@ class RunManager:
model_name, limit = model_concurrency_limit
active_for_model = sum(1 for r in self._runs.values() if r.status in (RunStatus.pending, RunStatus.running) and str((r.metadata or {}).get("model_name") or "").strip() == model_name)
if active_for_model >= max(1, int(limit)):
raise ModelConcurrencyLimitError(model_name, limit)
raise ModelConcurrencyLimitError(model_name, limit, scope="model")
if user_model_concurrency_limit is not None:
model_name, user_id, limit = user_model_concurrency_limit
active_for_user_model = sum(
1
for r in self._runs.values()
if r.status in (RunStatus.pending, RunStatus.running)
and str((r.metadata or {}).get("model_name") or "").strip() == model_name
and str((r.metadata or {}).get("user_id") or "").strip() == user_id
and str((r.metadata or {}).get("run_scope") or "external_root").strip() == "external_root"
)
if active_for_user_model >= max(1, int(limit)):
raise ModelConcurrencyLimitError(model_name, limit, scope="user_model", user_id=user_id)
record = RunRecord(
run_id=run_id,
@@ -316,9 +331,11 @@ class ConflictError(Exception):
class ModelConcurrencyLimitError(Exception):
"""Raised when a model-specific in-flight run limit has been reached."""
def __init__(self, model_name: str, limit: int) -> None:
def __init__(self, model_name: str, limit: int, *, scope: str = "model", user_id: str | None = None) -> None:
self.model_name = model_name
self.limit = limit
self.scope = scope
self.user_id = user_id
super().__init__(f"Model {model_name} has reached its concurrency limit")
offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/model_concurrency.py
+39
View File
@@ -0,0 +1,39 @@
"""Small in-memory gate used by legacy model-concurrency tests."""
from __future__ import annotations
import asyncio
class ModelConcurrencyGate:
"""Track per-model active counts with atomic try-acquire semantics."""
def __init__(self) -> None:
self._lock = asyncio.Lock()
self._counts: dict[str, int] = {}
async def try_acquire(self, model_name: str, limit: int) -> bool:
key = str(model_name).strip()
if not key:
return True
async with self._lock:
current = self._counts.get(key, 0)
if current >= max(1, int(limit)):
return False
self._counts[key] = current + 1
return True
async def release(self, model_name: str) -> None:
key = str(model_name).strip()
if not key:
return
async with self._lock:
current = self._counts.get(key, 0)
if current <= 1:
self._counts.pop(key, None)
else:
self._counts[key] = current - 1
async def snapshot(self) -> dict[str, int]:
async with self._lock:
return dict(self._counts)
offline-backend-20260512/backend/packages/harness/pyproject.toml
+4 -3
View File
@@ -23,9 +23,10 @@ dependencies = [
"markdownify>=1.2.2",
"markitdown[all,xlsx]>=0.0.1a2",
"pydantic>=2.12.5",
"pyyaml>=6.0.3",
"readabilipy>=0.3.0",
"tavily-python>=0.7.17",
"pyyaml>=6.0.3",
"readabilipy>=0.3.0",
"redis>=5.0.0",
"tavily-python>=0.7.17",
"firecrawl-py>=1.15.0",
"tiktoken>=0.8.0",
"ddgs>=9.10.0",
offline-backend-20260512/backend/tests/test_model_concurrency.py
+106
View File
@@ -10,10 +10,17 @@ import pytest
from app.gateway.routers import system_settings as ss_router
from app.gateway.services import (
RUN_SCOPE_EXTERNAL_ROOT,
RUN_SCOPE_ORCHESTRATION_CHILD,
_get_model_concurrency_limit,
_get_user_model_concurrency_limit,
_resolve_run_scope,
_resolve_effective_run_model_name,
internal_run_scope_headers,
)
from deerflow.config import system_settings as ss
from deerflow.config.concurrency_config import RedisConcurrencyConfig
from deerflow.runtime.runs.concurrency_gate import RedisConcurrencyGate
from deerflow.runtime.runs.manager import ModelConcurrencyLimitError, RunManager
@@ -52,23 +59,31 @@ async def test_model_concurrency_settings_round_trip(
enabled=True,
default_limit=8,
model_limits={"glm5": 10, "deepseek": 4},
user_default_limit=2,
user_model_limits={"glm5": 1},
),
req,
)
assert saved.enabled is True
assert saved.default_limit == 8
assert saved.model_limits == {"glm5": 10, "deepseek": 4}
assert saved.user_default_limit == 2
assert saved.user_model_limits == {"glm5": 1}
file_path = tmp_runtime_home / "system_settings.json"
payload = json.loads(file_path.read_text(encoding="utf-8"))
assert payload["model_concurrency"]["enabled"] is True
assert payload["model_concurrency"]["default_limit"] == 8
assert payload["model_concurrency"]["model_limits"]["glm5"] == 10
assert payload["model_concurrency"]["user_default_limit"] == 2
assert payload["model_concurrency"]["user_model_limits"]["glm5"] == 1
loaded = await ss_router.get_model_concurrency(req)
assert loaded.enabled is True
assert loaded.default_limit == 8
assert loaded.model_limits["deepseek"] == 4
assert loaded.user_default_limit == 2
assert loaded.user_model_limits["glm5"] == 1
def test_resolve_effective_run_model_name_prefers_requested_model(
@@ -183,6 +198,53 @@ async def test_run_manager_rejects_model_limit_atomically(
)
@pytest.mark.asyncio
async def test_run_manager_rejects_user_model_limit_atomically(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(
"app.gateway.services.load_system_settings",
lambda: SimpleNamespace(
model_concurrency=SimpleNamespace(
enabled=True,
default_limit=10,
model_limits={},
user_default_limit=3,
user_model_limits={"glm5": 1},
)
),
)
assert _get_user_model_concurrency_limit("glm5") == 1
assert _get_user_model_concurrency_limit("other") == 3
run_mgr = RunManager()
await run_mgr.create_or_reject(
"t-child",
metadata={
"model_name": "glm5",
"user_id": "u1",
"run_scope": RUN_SCOPE_ORCHESTRATION_CHILD,
},
user_model_concurrency_limit=("glm5", "u1", 1),
)
await run_mgr.create_or_reject("t1", metadata={"model_name": "glm5", "user_id": "u1"})
await run_mgr.create_or_reject(
"t2",
metadata={"model_name": "glm5", "user_id": "u2"},
user_model_concurrency_limit=("glm5", "u2", 1),
)
with pytest.raises(ModelConcurrencyLimitError) as exc_info:
await run_mgr.create_or_reject(
"t3",
metadata={"model_name": "glm5", "user_id": "u1"},
user_model_concurrency_limit=("glm5", "u1", 1),
)
assert exc_info.value.scope == "user_model"
assert exc_info.value.user_id == "u1"
def test_get_model_concurrency_limit_skips_when_disabled(
monkeypatch: pytest.MonkeyPatch,
) -> None:
@@ -193,7 +255,51 @@ def test_get_model_concurrency_limit_skips_when_disabled(
enabled=False,
default_limit=1,
model_limits={"glm5": 1},
user_default_limit=1,
user_model_limits={"glm5": 1},
)
),
)
assert _get_model_concurrency_limit("glm5") is None
assert _get_user_model_concurrency_limit("glm5") is None
def test_internal_run_scope_is_only_trusted_for_loopback_or_server_argument() -> None:
body = SimpleNamespace(metadata={"run_scope": RUN_SCOPE_ORCHESTRATION_CHILD})
trusted_loopback_request = SimpleNamespace(
client=SimpleNamespace(host="127.0.0.1"),
headers=internal_run_scope_headers(),
)
untrusted_loopback_request = SimpleNamespace(
client=SimpleNamespace(host="127.0.0.1"),
headers={},
)
public_request = SimpleNamespace(
client=SimpleNamespace(host="10.0.0.8"),
headers=internal_run_scope_headers(),
)
assert (
_resolve_run_scope(body, trusted_loopback_request, None)
== RUN_SCOPE_ORCHESTRATION_CHILD
)
assert (
_resolve_run_scope(body, untrusted_loopback_request, None)
== RUN_SCOPE_EXTERNAL_ROOT
)
assert _resolve_run_scope(body, public_request, None) == RUN_SCOPE_EXTERNAL_ROOT
assert (
_resolve_run_scope(
SimpleNamespace(metadata={}),
public_request,
RUN_SCOPE_ORCHESTRATION_CHILD,
)
== RUN_SCOPE_ORCHESTRATION_CHILD
)
def test_redis_concurrency_namespace_has_system_prefix() -> None:
assert RedisConcurrencyConfig().namespace == "cmzs_deerflow:concurrency"
gate = RedisConcurrencyGate(RedisConcurrencyConfig(namespace=""))
assert gate._ns() == "cmzs_deerflow:concurrency"
offline-backend-20260512/offline-runtime-requirements.txt
+27
View File
@@ -0,0 +1,27 @@
# Direct runtime packages layered into the prebuilt offline backend image.
# Keep these three pins synchronized with start_backend.sh.
croniter==6.2.2
hindsight-client==0.6.2
redis==5.2.1
# Fully pinned dependency closure for CPython 3.12 on Linux. Keeping the
# transitive packages here makes a newly generated wheelhouse reproducible.
aiohappyeyeballs==2.7.1
aiohttp==3.14.2
aiohttp-retry==2.9.1
aiosignal==1.4.0
annotated-types==0.7.0
async-timeout==5.0.1
attrs==26.1.0
frozenlist==1.8.0
idna==3.18
multidict==6.7.1
propcache==0.5.2
pydantic==2.13.4
pydantic-core==2.46.4
python-dateutil==2.9.0.post0
six==1.17.0
typing-extensions==4.16.0
typing-inspection==0.4.2
urllib3==2.7.0
yarl==1.24.5
offline-backend-20260512/prepare_wheelhouse.sh
Executable
+99
View File
@@ -0,0 +1,99 @@
#!/usr/bin/env bash
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
PYTHON_BIN="${PYTHON_BIN:-python3}"
REQUIREMENTS_FILE="${REQUIREMENTS_FILE:-${SCRIPT_DIR}/offline-runtime-requirements.txt}"
WHEELHOUSE_DIR="${WHEELHOUSE_DIR:-${SCRIPT_DIR}/wheelhouse}"
PIP_INDEX_URL="${PIP_INDEX_URL:-https://pypi.org/simple}"
# The supplied backend image is CPython 3.12 on Debian x86_64. These values can
# be overridden when building an offline bundle for a different image/CPU.
TARGET_PLATFORM="${TARGET_PLATFORM:-manylinux2014_x86_64}"
TARGET_PYTHON_VERSION="${TARGET_PYTHON_VERSION:-312}"
TARGET_IMPLEMENTATION="${TARGET_IMPLEMENTATION:-cp}"
TARGET_ABI="${TARGET_ABI:-cp312}"
die() {
echo "ERROR: $*" >&2
exit 1
}
require_cmd() {
command -v "$1" >/dev/null 2>&1 || die "command not found: $1"
}
require_wheel() {
local pattern="$1"
compgen -G "${WHEELHOUSE_DIR}/${pattern}" >/dev/null || die "wheel not downloaded: ${pattern}"
}
require_cmd "${PYTHON_BIN}"
require_cmd sha256sum
[ -f "${REQUIREMENTS_FILE}" ] || die "requirements file not found: ${REQUIREMENTS_FILE}"
"${PYTHON_BIN}" -m pip --version >/dev/null 2>&1 || die "pip is not available for ${PYTHON_BIN}"
mkdir -p "${WHEELHOUSE_DIR}"
download_args=(
download
--dest "${WHEELHOUSE_DIR}"
--only-binary=:all:
--platform "${TARGET_PLATFORM}"
--python-version "${TARGET_PYTHON_VERSION}"
--implementation "${TARGET_IMPLEMENTATION}"
--abi "${TARGET_ABI}"
)
echo "Downloading Linux/Python runtime wheels"
echo " requirements: ${REQUIREMENTS_FILE}"
echo " destination: ${WHEELHOUSE_DIR}"
echo " platform: ${TARGET_PLATFORM}"
echo " Python/ABI: ${TARGET_PYTHON_VERSION}/${TARGET_ABI}"
"${PYTHON_BIN}" -m pip "${download_args[@]}" \
--index-url "${PIP_INDEX_URL}" \
--requirement "${REQUIREMENTS_FILE}"
require_wheel 'croniter-6.2.2-*.whl'
require_wheel 'hindsight_client-0.6.2-*.whl'
require_wheel 'redis-5.2.1-*.whl'
# Resolve the same requirements a second time using only wheelhouse. This
# catches a missing transitive dependency before the bundle reaches offline.
verify_dir="$(mktemp -d)"
cleanup() {
if [ -n "${verify_dir:-}" ] && [ -d "${verify_dir}" ]; then
rm -rf -- "${verify_dir}"
fi
}
trap cleanup EXIT
verify_args=(
download
--dest "${verify_dir}"
--no-index
--find-links "${WHEELHOUSE_DIR}"
--only-binary=:all:
--platform "${TARGET_PLATFORM}"
--python-version "${TARGET_PYTHON_VERSION}"
--implementation "${TARGET_IMPLEMENTATION}"
--abi "${TARGET_ABI}"
--requirement "${REQUIREMENTS_FILE}"
)
"${PYTHON_BIN}" -m pip "${verify_args[@]}" >/dev/null
shopt -s nullglob
wheels=("${WHEELHOUSE_DIR}"/*.whl)
[ "${#wheels[@]}" -gt 0 ] || die "wheelhouse is empty: ${WHEELHOUSE_DIR}"
(
cd "${WHEELHOUSE_DIR}"
sha256sum -- *.whl | LC_ALL=C sort > SHA256SUMS
)
echo
echo "Wheelhouse is complete for the pinned runtime requirements."
echo " wheel count: ${#wheels[@]}"
echo " checksums: ${WHEELHOUSE_DIR}/SHA256SUMS"
echo "Copy the wheelhouse directory with the deployment bundle."
offline-backend-20260512/start_backend.sh
Executable
+140
View File
@@ -0,0 +1,140 @@
#!/usr/bin/env bash
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
IMAGE_TAG="${IMAGE_TAG:-deerflow-backend-offline:20260505}"
CONTAINER_NAME="${CONTAINER_NAME:-deerflow-backend}"
HOST_PORT="${HOST_PORT:-8001}"
IMAGE_TAR="${IMAGE_TAR:-${SCRIPT_DIR}/deerflow-backend-offline-image-20260505.tar}"
# Default timezone for the container. The offline image is built from a base
# layer that ships with UTC, which made scheduled tasks fire at the wrong wall
# clock and made log timestamps confusing for ops. ``Asia/Shanghai`` keeps the
# scheduler aligned with the operator's expectation; override with TZ=... when
# deploying to a different region.
CONTAINER_TZ="${CONTAINER_TZ:-${TZ:-Asia/Shanghai}}"
die() {
echo "ERROR: $*" >&2
exit 1
}
require_cmd() {
command -v "$1" >/dev/null 2>&1 || die "command not found: $1"
}
require_file() {
[ -f "$1" ] || die "required file not found: $1"
}
require_dir() {
[ -d "$1" ] || die "required directory not found: $1"
}
require_env_key() {
local key="$1"
grep -Eq "^[[:space:]]*${key}=" "${SCRIPT_DIR}/.env" || die "required .env key not found: ${key}"
}
require_cmd docker
if [ ! -f "${SCRIPT_DIR}/.env" ]; then
if [ -f "${SCRIPT_DIR}/.env.example" ]; then
cp "${SCRIPT_DIR}/.env.example" "${SCRIPT_DIR}/.env"
echo "Generated .env from .env.example. Please edit .env and run this script again."
exit 1
fi
die "missing .env"
fi
if [ ! -f "${SCRIPT_DIR}/config.yaml" ]; then
if [ -f "${SCRIPT_DIR}/config.yaml.example" ]; then
cp "${SCRIPT_DIR}/config.yaml.example" "${SCRIPT_DIR}/config.yaml"
echo "Generated config.yaml from config.yaml.example. Please edit model/API settings and run this script again."
exit 1
fi
die "missing config.yaml"
fi
require_env_key "AUTH_JWT_SECRET"
require_env_key "SCHEDULED_TASKS_ENABLED"
require_env_key "ELEMENT_ENABLED"
require_file "${SCRIPT_DIR}/config.yaml"
require_file "${SCRIPT_DIR}/extensions_config.json"
require_file "${SCRIPT_DIR}/skill_evolution.json"
require_file "${SCRIPT_DIR}/backend/pyproject.toml"
require_file "${SCRIPT_DIR}/backend/app/gateway/app.py"
require_file "${SCRIPT_DIR}/backend/packages/harness/deerflow/runtime/scheduler/element.py"
require_dir "${SCRIPT_DIR}/skills"
require_dir "${SCRIPT_DIR}/wheelhouse"
# Knowledge base directory, mounted into the container at /mnt/knowledge_base.
require_dir "${SCRIPT_DIR}/backend/knowledge/knowledge_base"
mkdir -p "${SCRIPT_DIR}/data"
if ! docker image inspect "${IMAGE_TAG}" >/dev/null 2>&1; then
require_file "${IMAGE_TAR}"
echo "Loading Docker image from ${IMAGE_TAR}"
docker load -i "${IMAGE_TAR}"
fi
if docker ps -a --format '{{.Names}}' | awk -v name="${CONTAINER_NAME}" '$0 == name { found=1 } END { exit(found ? 0 : 1) }'; then
echo "Removing existing container ${CONTAINER_NAME}"
docker rm -f "${CONTAINER_NAME}" >/dev/null
fi
echo "Starting backend container ${CONTAINER_NAME}"
docker run -d \
--name "${CONTAINER_NAME}" \
--restart unless-stopped \
-p "${HOST_PORT}:8001" \
--env-file "${SCRIPT_DIR}/.env" \
-e TZ="${CONTAINER_TZ}" \
-e DEER_FLOW_PROJECT_ROOT=/app \
-e DEER_FLOW_HOME=/app/backend/.deer-flow \
-e DEER_FLOW_CONFIG_PATH=/app/config.yaml \
-e DEER_FLOW_EXTENSIONS_CONFIG_PATH=/app/extensions_config.json \
-e DEER_FLOW_SKILL_EVOLUTION_CONFIG_PATH=/app/skill_evolution.json \
-e DEER_FLOW_SKILLS_PATH=/app/skills \
-e UV_LINK_MODE=copy \
-v "${SCRIPT_DIR}/backend/app:/app/backend/app" \
-v "${SCRIPT_DIR}/backend/packages:/app/backend/packages" \
-v "${SCRIPT_DIR}/backend/scripts:/app/backend/scripts" \
-v "${SCRIPT_DIR}/backend/pyproject.toml:/app/backend/pyproject.toml:ro" \
-v "${SCRIPT_DIR}/backend/uv.lock:/app/backend/uv.lock:ro" \
-v "${SCRIPT_DIR}/backend/langgraph.json:/app/backend/langgraph.json:ro" \
-v "${SCRIPT_DIR}/config.yaml:/app/config.yaml:ro" \
-v "${SCRIPT_DIR}/extensions_config.json:/app/extensions_config.json:ro" \
-v "${SCRIPT_DIR}/skill_evolution.json:/app/skill_evolution.json:ro" \
-v "${SCRIPT_DIR}/skills:/app/skills:ro" \
-v "${SCRIPT_DIR}/wheelhouse:/app/wheelhouse:ro" \
-v "${SCRIPT_DIR}/backend/knowledge/knowledge_base:/mnt/knowledge_base:ro" \
-v "${SCRIPT_DIR}/data:/app/backend/.deer-flow" \
"${IMAGE_TAG}" \
sh -c 'cd backend
# Apply timezone inside the container. ``TZ`` alone is enough for Python /
# uvicorn log timestamps, but linking /etc/localtime keeps C-level tools
# (e.g. anything that reads /etc/localtime directly) on the same clock.
if [ -n "${TZ:-}" ] && [ -f "/usr/share/zoneinfo/${TZ}" ]; then
ln -sf "/usr/share/zoneinfo/${TZ}" /etc/localtime 2>/dev/null || true
echo "${TZ}" > /etc/timezone 2>/dev/null || true
fi
if [ "${SCHEDULER_INSTALL_DEPS:-1}" != "0" ]; then
uv run --no-sync python -c "import croniter" >/dev/null 2>&1 || {
PYTHON_BIN="$(uv run --no-sync python -c "import sys; print(sys.executable)")"
uv pip install --python "${PYTHON_BIN}" --no-index --find-links /app/wheelhouse ${SCHEDULER_PIP_PACKAGES:-croniter==6.2.2}
}
fi
if [ "${HINDSIGHT_INSTALL_DEPS:-1}" != "0" ]; then
uv run --no-sync python -c "import hindsight_client" >/dev/null 2>&1 || {
PYTHON_BIN="$(uv run --no-sync python -c "import sys; print(sys.executable)")"
uv pip install --python "${PYTHON_BIN}" --no-index --find-links /app/wheelhouse ${HINDSIGHT_PIP_PACKAGES:-hindsight-client==0.6.2}
}
fi
uv run --no-sync python -c "import redis" >/dev/null 2>&1 || uv pip install --python "$(uv run --no-sync python -c "import sys; print(sys.executable)")" --no-index --find-links /app/wheelhouse ${REDIS_PIP_PACKAGES:-redis==5.2.1}
PYTHONPATH=. uv run --no-sync python -m uvicorn app.gateway.app:app --host 0.0.0.0 --port 8001 --workers ${GATEWAY_WORKERS:-2}'
echo
echo "Backend started."
echo " URL: http://<server-ip>:${HOST_PORT}"
echo " Logs: docker logs -f ${CONTAINER_NAME}"
echo " Stop: ./stop_backend.sh"