DevFlow/docs/03-模块文档/df-ai-AI集成模块.md

df-ai — AI 集成模块

创建: 2026-06-10 | 最后更新: 2026-06-13

---

概述

df-ai 是 DevFlow 的 AI 核心层，提供 LLM Provider 抽象、双协议实现（OpenAI 兼容 + Anthropic）、流式 SSE 解析、工具调用、上下文窗口管理和 embedding 支持。

---

当前状态

能力	状态
LlmProvider trait	✅
OpenAI 兼容 Provider（流式/非流式/embed）	✅
Anthropic Provider（流式）	✅ Sprint 8
ContextManager（分组滑窗）	✅ Sprint 11
embed() 向量生成	✅ Sprint 15
AiToolRegistry（基础设施）	✅ Sprint 5（12 工具注册在 commands/ai.rs）
coordinator	⬜ 空壳（B 路线待填）

---

文件结构

crates/df-ai/src/
├── lib.rs              — 公共导出
├── provider.rs         — LlmProvider trait（含 embed 默认实现）
├── openai_compat.rs    — OpenAI 兼容实现（chat + embed）
├── anthropic_compat.rs — Anthropic Messages API 实现（Sprint 8）
├── context.rs          — ContextManager 分组滑动窗口（Sprint 11）
├── ai_tools.rs         — AiToolRegistry + 12 工具定义
├── coordinator.rs      — AgentCoordinator 空壳（B 路线）
├── router.rs           — ModelRouter 模型路由空壳（route() 按 TaskType 选模型，当前全返回 default_model）
└── stream.rs           — StreamCollector 流式辅助（累积 chunk.delta 文本 + 跟踪 finished 标志）

---

LlmProvider Trait

pub type StreamResult = Pin<Box<dyn Stream<Item = anyhow::Result<StreamChunk>> + Send>>;

#[async_trait]
pub trait LlmProvider: Send + Sync {
    // 非流式完整响应（用于标题生成/知识提炼）
    async fn complete(&self, request: CompletionRequest)
        -> anyhow::Result<CompletionResponse>;

    // 流式 SSE（主对话）
    async fn stream(&self, request: CompletionRequest)
        -> anyhow::Result<StreamResult>;

    // 向量生成（默认 bail，openai_compat 覆盖实现）
    async fn embed(&self, _model: &str, _texts: Vec<String>)
        -> anyhow::Result<Vec<Vec<f32>>> {
        anyhow::bail!("该 Provider 不支持 embedding({})", self.name())
    }

    // Provider 名称（必填）
    fn name(&self) -> &str;

    // 支持的特性（必填：streaming / function_calling / vision）
    fn supported_features(&self) -> ProviderFeatures;
}

---

OpenAI 兼容 Provider（openai_compat.rs）

支持 OpenAI / DeepSeek / GLM / 本地 Ollama 等任意 OpenAI 兼容端点。

特性	实现
base_url 智能拼接	chat_url() 三分支：已含 /chat/completions 直用；以 /v<数字> 结尾（如 /api/paas/v4，由 ends_with_version 判定）补 /chat/completions；仅域名（如 api.openai.com）补 /v1/chat/completions
流式	stream: true + SSE 逐 chunk 解析（apply_openai_sse 纯函数）
工具调用	tool_calls 按 index 排序（消 HashMap 迭代乱序）
usage 解析	stream_options: {include_usage: true}，末 chunk 读累计值
finish 判定	逐 chunk 判 finish_reason：stop/tool_calls/length 三者同等视为正常 finished（length = max_tokens 截断，属正常终止而非断连）
embed	POST /v1/embeddings，响应按 index 排序返回 Vec<Vec<f32>>
connect timeout	connect_timeout(30s)，不设总 timeout（避免误砍流式长任务）

注：idle timeout（120s）与断连丢弃（finished_received）属上层 stream_llm（src-tauri/commands/ai.rs）的职责，不在本 Provider 层。

SSE 解析抽成 pub(crate) fn apply_openai_sse(data: &str, usage_accum: &mut Option<TokenUsage>) -> StreamChunk 纯函数，与 HTTP/eventsource 解耦，便于单测（喂构造 data 字符串验证 [DONE]/usage 覆盖/tool_calls 等分支）。

---

Anthropic Provider（anthropic_compat.rs，Sprint 8）

特性	实现
认证	x-api-key + anthropic-version: 2023-06-01
消息格式	顶层 system 字段，max_tokens 必填（请求缺省时兜底 DEFAULT_MAX_TOKENS = 4096）
流式 SSE	event 类型解析（全集见下）
工具调用	content_block type=tool_use
usage	message_start 初始化累加器(input_tokens)、message_delta 覆盖 completion(output_tokens 为累计值，非增量)、message_stop 经 take() 带出累积 usage
embed	不支持（Anthropic 无 embed API），继承 trait 默认 Err

SSE 事件全集（按 type 字段分发）:

event type	处理
message_start	用 message.usage.input_tokens 初始化累加器（output 置 0）
message_delta	usage.output_tokens 是累计值（非增量），直接覆盖 completion 并重算 total
content_block_delta(text_delta)	文本增量
content_block_delta(input_json_delta)	工具入参增量（带 index）
content_block_start(tool_use)	工具块开始，带 id + name
message_stop	返回 finished=true 终态 chunk，usage 经 take() 带出
error	返回 finished=true 终态空 chunk（不清空累加器）
其它（content_block_stop / ping）	空 chunk

SSE 解析抽成 pub(crate) fn apply_anthropic_event(data: &str, usage_accum: &mut Option<TokenUsage>) -> StreamChunk 纯函数，与 HTTP/eventsource 解耦，便于单测（喂构造 data 字符串验证事件分支与 usage 累积）。

支持 GLM 订阅端点（https://open.bigmodel.cn/api/anthropic）。

---

ContextManager 分组滑动窗口（context.rs，Sprint 11）

解决长对话无限增长导致 context_length_exceeded 死锁。

核心设计

• TokenEstimator：字符粗估，零依赖，保守 ±15%。单条公式 = content(chars×0.35 ceil) + per_message 4 + 每个 tool_call(per_tool_call 30 + name/arguments 各按 chars×0.35 ceil) + 有 tool_call_id 时 +3
• ContextConfig：max_tokens 128k / output_reserve 8192 / safety 0.85；预算公式 (max_tokens − output_reserve) × safety_ratio，由 ContextConfig::budget_limit() 提供（ContextManager::budget_limit() 仅转发它）
• 分组淘汰单元：工具调用三元组（ToolCallHead + ToolResultTail* + 紧随文本 Assistant）原子性同进同出，防上下文语义断裂
• 保护区：PROTECT_COUNT = 6（编译期常量，≈ 最近 2 个完整用户轮次），最后 6 条消息永不裁
• 裁剪范围：仅影响发送视图（build_for_request），全量历史（all_messages_clone）用于持久化

关键方法

方法	用途
push(message: ChatMessage)	追加消息并计 token、更新 history_tokens 缓存（push 不裁剪，裁剪统一在 build_for_request）
clear()	清空消息历史并重置 token 计数
len() -> usize	历史消息条数
is_empty() -> bool	历史是否为空
history_tokens() -> u32	当前历史累计 token（不含 system prompt）
budget_limit() -> u32	上下文预算上限，转发 config.budget_limit() = (max_tokens − output_reserve) × safety_ratio
iter() -> impl Iterator<Item = &ChatMessage>	只读迭代历史消息（如标题生成 filter）
build_for_request(sys_tokens) -> (Vec<ChatMessage>, bool)	返回裁剪后的消息列表 + 是否发生裁剪（用于 LLM 调用）
all_messages_clone()	返回全量消息（用于 save_conversation / 标题生成）
restore_from_messages(msgs)	切换对话时重建 ContextManager 缓存
replace_tool_result_content(tool_call_id, new_content) -> bool	原子更新工具结果内容（审批通过/拒绝时回填），返回是否找到并替换

---

AI 工具注册

归属：ai_tools.rs 仅提供基础设施（RiskLevel / AiTool / AiToolRegistry）；12 个工具的具体定义与注册在 src-tauri/src/commands/ai.rs::build_ai_tool_registry，handler 即唯一执行路径（schema+risk+实现同源）。

12 个内置工具，按风险分级：

风险	工具
Low（自动执行）	list_projects / list_tasks / list_ideas / read_file / list_directory
Medium（需审批）	update_project / create_project / create_task / create_idea / write_file
High（需审批）	delete_project / run_workflow

工具执行结果写 ai_tool_executions 表（审计日志）。

---

知识库集成（Sprint 15，逻辑在 src-tauri/commands/ai.rs）

知识注入

build_knowledge_context(state, query, config) -> String：

• config.auto_inject == false → 立即返回空（零开销）
• hybrid_search() top-3（LIKE 或 LIKE+向量混合）
• 每条命中调 increment_reuse_count（fire-and-forget）
• 格式化为 markdown，注入 system prompt 头部

知识提炼

extract_knowledge_from_conversation(db, conv_id, provider_cfg)：

• 后台 spawn（tauri::async_runtime::spawn），提炼失败仅 warn 不阻断
• 取最后 6 条 user/assistant 消息 → LLM JSON 输出（强制 JSON schema）
• parse 失败整批丢弃；成功则逐条写 candidate

maybe_spawn_extraction(...)：agentic loop 两处正常退出路径统一调用。

向量 embedding（Phase 5.5）

generate_embedding(state, text, config) -> Option<Vec<f32>>：

• 截断 8000 字 → 找 embedding_provider_id 对应 provider → embed()
• 失败返回 None（降级 LIKE）

hybrid_search(state, query, limit, config)：三层降级链：

1. vector_enabled == false → 纯 LIKE
2. embed 调用失败 → 纯 LIKE
3. 正常 → 双信号合并排序（同时 LIKE+向量 cos≥0.3 > 仅 LIKE > 仅向量 cos≥0.3）

---

LLM 并发控制（state.rs，Sprint 11 Part C）

LlmConcurrency：双层 Semaphore，运行时可调。

Semaphore	默认 permits	限流对象
global	3	全部 LLM 调用（stream_llm / 标题 / 提炼）
per_conv	2	单对话并发

本地工具执行（tools.execute）不限流，无外部成本。

---

决策能力缺口（B 路线，待立项）

能力	现状	需补
Planning（任务规划）	缺失	意图 → LLM 规划子任务 DAG
Coordinator（多 agent）	coordinator.rs::run() 全 TODO	实现多 agent 协作拆 DAG
Conditions（条件分支）	缺失（df-ai 无实现，条件求值属 df-workflow crate）	JSON Path / 比较 / and-or-not
Reflection（自纠）	缺失	执行后自检 / 重试

详见 Phase 2 计划 - 决策能力升级。

---

相关文档

• df-storage 存储层
• df-workflow 工作流引擎
• df-nodes 节点集合