kg-agent

申勤物业知识图谱 Agent 问答系统

基于知识图谱的物业业务智能问答系统：把市场部、财务部、人事部、运营部、采购部的业务数据 构建为以「项目」为核心的 Neo4j 知识图谱，通过 LangGraph Agent + DeepSeek 提供自然语言问答， 答案可溯源到图谱中的节点与关系。

功能特性

• 知识图谱构建：15 类数据模板（Excel）→ 字段校验 → 两阶段构建（先节点后关系）→ Neo4j；
• Agent 问答：问题理解（分类+槽位）→ 实体/概念/值 三层接地 → 能力拓展（派生属性推理） → 查询规划（含返回字段）→ 执行（按需返回字段）→ 规则+LLM 双层检查 → 带溯源回答；
• 多轮对话：同一会话内保留结构化上下文（问题/实体主键/回答），支持拆解式提问与指代； 多轮答案复用（reuse_check）由 LLM 结合 当前问题+历史问题+历史回答 判断是否可直接推导；
• API 服务：FastAPI + 全异步 astream，thread_id 会话管理，SSE 流式处理进度，多请求并发；
• 人工确认：槽位确认、计划确认、子图范围收缩，均可自动化或走 human-in-the-loop；
• 只读保障：所有问答 LLM 提示词注入只读约束，问答阶段禁止修改图谱数据。

技术栈

层	选型
环境	uv（Python ≥3.11）
图存储	Neo4j 5+（neo4j://127.0.0.1:7687）
LLM	DeepSeek（deepseek-v4-flash，OpenAI 兼容）
语义嵌入	Qwen3-Embedding-0.6B（本地模型）
Agent 编排	LangGraph
API	FastAPI + Uvicorn（全异步 astream）

快速开始

1. 环境准备

uv sync
cp .env.example .env

编辑 .env：

NEO4J_URI=neo4j://127.0.0.1:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=你的密码
DEEPSEEK_API_KEY=sk-你的key
DEEPSEEK_BASE_URL=https://api.deepseek.com
DEEPSEEK_MODEL=deepseek-v4-flash

2. 启动 Neo4j

在 Neo4j Desktop（或服务）中启动数据库，确认 Bolt 端口（默认 7687）可连接。

3. 准备嵌入模型

将 Qwen3-Embedding-0.6B 放到 models/Qwen3-Embedding-0.6B（API 服务启动时会预加载；CLI 首次问答会加载，之后进程内缓存）。

数据与图谱构建

生成模板与测试数据

uv run python scripts/generate_templates.py      # 生成 15 类数据模板（data/templates）
uv run python scripts/generate_test_data.py      # 生成与模板对应的测试数据（data/test_data）

构建知识图谱

uv run python scripts/build_graph.py data/config_example.json --clear

构建分两阶段（先创建全部节点，再创建全部关系），包含字段名写死校验、必填校验与引用存在性校验。

真实数据接入：按最新 DMS 范围的字段映射见 docs/DMS字段映射_最新范围.md；需要人工补填的字段清单见 data/manual_fill/DMS补数字段清单.csv（生成脚本 scripts/generate_dms_supplement.py）。

Agent 问答（CLI）

单次提问

uv run python scripts/ask.py "青浦区图书馆3月份有加班的人是谁"

常用参数：

--auto    自动确认（跳过人工确认，批处理/评测用）
--debug   显示每步实际执行的 实参/丢弃参数/Cypher/首行/检查报告
--user    提问人身份（工号或用户名，默认 admin01）

交互多轮对话

uv run python scripts/ask.py            # 或加 --auto 自动确认

问题 > 青浦区图书馆3月份有加班的人是谁
（回答）
问题 > 那他们4月份呢
（自动引用上一轮实体继续）
问题 > 清空
（重置上下文，开始新话题）

API 服务

uv run python -m knowledge_agent.api                 # 本机访问
uv run python -m knowledge_agent.api --host 0.0.0.0  # 局域网访问（需放行防火墙 8000 端口）

默认监听 127.0.0.1:8000（--host 0.0.0.0 时局域网内通过 http://<你的IP>:8000 访问）， 交互文档见 http://127.0.0.1:8000/docs。

⚠️ 当前未接鉴权，0.0.0.0 对外开放前请先加 API Key / 权限校验。

接口

接口	说明
POST /ask	body {thread_id, query, auto_confirm, reuse_check}；块返回完整结果；遇人工确认按 auto_confirm 处理
POST /ask/stream	SSE 流式：event: progress（节点级进度）、event: confirm、event: done
POST /threads/{id}/resume	恢复被中断会话（{reply: "确认"} 或修改意见）
GET /threads/{id}/history	会话历史（rounds + messages）

请求字段说明：

• thread_id：会话标识，同一 id 共享上下文（多轮）；
• query：用户问题；
• auto_confirm（默认 true）：自动确认槽位/计划/范围；false 时返回 need_confirm，前端确认后调 /resume；
• reuse_check（默认 true）：有历史回答时由 LLM 判断能否直接复用推导答案；false 强制每次完整检索。

SSE 进度事件说明（event: progress 的 data 含 node 与 label，label 为中文阶段说明，可直接展示）：

node	label（处理阶段）
understand	正在理解问题（分类 + 槽位抽取）
chat	正在生成回答
reuse_check	正在判断是否可复用历史回答
answer_reuse	正在基于历史回答推导答案
ground	正在把问题实体对齐到知识图谱
capability	正在推理派生属性（如 年龄 ← 出生日期）
cap_check / cap_llm_check	正在检查/审查派生规则
confirm	等待确认查询条件
plan	正在制定查询计划
confirm_plan	等待确认查询计划
run	正在检索知识图谱
run_check / run_llm_check	正在检查/审查检索结果
reflect	正在判断结果能否回答该问题
scope	正在缩小查询范围
answer	正在生成回答

回答生成时还会推送 event: answer_chunk（data: {"text": "..."}）——最终回答的逐块增量， 前端可边收边追加显示（打字机效果）；所有块拼接后与最终 answer 字段一致。 闲聊回复（chat 节点）同样走该事件。

前端接入示例（SSE 流式解析）

// 用 fetch 流式读取 SSE（支持 POST /ask/stream）
const res = await fetch("http://127.0.0.1:8000/ask/stream", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({
    thread_id: "user-001",
    query: "青浦区图书馆3月份有加班的人是谁",
    auto_confirm: true,
    reuse_check: true,
  }),
});

const reader = res.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
let answerText = "";

function onEvent(event, data) {
  if (event === "progress") {
    // data: {node, label}；node=answer/chat 时还带完整 answer
    if (data.node !== "answer" && data.node !== "chat") setStatus(data.label); // 更新“正在…”状态
  } else if (event === "answer_chunk") {
    // data: {text}——最终回答增量，直接追加（打字机效果）
    answerText += data.text;
    appendAnswer(data.text);
  } else if (event === "confirm") {
    // data: {message, options}——需要人工确认，展示后调 /threads/{id}/resume
    showConfirm(data);
  } else if (event === "done") {
    // data: 完整结果，answer 为最终回答（兜底，防丢块）
    setAnswer(data.answer);
  }
}

// 解析 SSE：按空行分隔事件，每事件含 "event: xxx" 与 "data: {...}"
while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  buffer += decoder.decode(value, { stream: true });
  let idx;
  while ((idx = buffer.indexOf("\n\n")) >= 0) {
    const raw = buffer.slice(0, idx);
    buffer = buffer.slice(idx + 2);
    const lines = raw.split("\n");
    const event = lines.find((l) => l.startsWith("event: "))?.slice(7);
    const dataLine = lines.find((l) => l.startsWith("data: "));
    if (event && dataLine) onEvent(event, JSON.parse(dataLine.slice(6)));
  }
}

说明：

• progress：更新处理状态（label 已是中文阶段说明）；
• answer_chunk：把 text 追加到回答区，实现逐字/逐块显示；拼接结果与 done.answer 一致；
• confirm（auto_confirm=false 时）：展示确认信息，用户确认后调 POST /threads/{id}/resume（{reply: "确认"} 或修改意见）；
• done：携带完整结果（含 answer、plan、subgraph 等），作为最终兜底；
• 原生 EventSource 只支持 GET，若要用它，需要把接口改为 GET（当前为 POST，用上面的 fetch 流式方式即可）。

curl 示例

Windows PowerShell 下中文 body 建议写 UTF-8 文件避免编码问题：

# body.json（UTF-8 保存）
# {"thread_id":"user-001","query":"青浦区图书馆3月份有加班的人是谁","auto_confirm":true,"reuse_check":true}

curl.exe --request POST --url http://127.0.0.1:8000/ask `
  --header "Content-Type: application/json" `
  --data "@body.json"

或使用 PowerShell 原生：

$body = @{ thread_id = "user-001"; query = "青浦区图书馆3月份有加班的人是谁"; auto_confirm = $true; reuse_check = $true } | ConvertTo-Json
Invoke-RestMethod -Uri "http://127.0.0.1:8000/ask" -Method Post `
  -ContentType "application/json; charset=utf-8" -Body ([System.Text.Encoding]::UTF8.GetBytes($body))

架构概览

graph LR
  Q[用户问题] --> U[问题理解 一次LLM]
  U --> R[复用检查 LLM判断]
  R -->|可复用| A2[直接给出历史推导答案]
  R -->|需检索| G[实体/概念/值 接地]
  G --> C[能力拓展 派生推理]
  C --> P[查询规划 含返回字段]
  P --> E[执行 Neo4j 按需返回]
  E --> A[回答 带溯源]

详细设计（节点职责、时间建模、检索/审查流程、安全与权限、DMS 接入）见 docs/技术方案.md。

目录结构

knowledge_agent/
├── pyproject.toml / uv.lock / .env
├── README.md
├── docs/                    # 技术方案、DMS 字段映射、架构图
├── data/
│   ├── templates/           # 15 类数据模板（+填写说明）
│   ├── test_data/           # 测试数据（与模板一一对应）
│   ├── manual_fill/         # DMS 补数字段清单 / 人工补填清单
│   └── config_example.json  # 构建配置
├── models/Qwen3-Embedding-0.6B
├── scripts/                 # 模板/测试数据/构建/问答 CLI/DMS 映射等
└── src/knowledge_agent/
    ├── config.py            # .env 配置
    ├── db.py                # Neo4j driver 单例
    ├── meta/                # 元知识图谱（单一事实源）
    ├── graph/               # 图谱构建（schemas/reader/builder）
    ├── retrieval/           # 实体索引/接地/查询工具集
    ├── agent/               # LangGraph 节点/派生引擎/概念接地/值接地/API
    └── api.py               # FastAPI 服务

安全与权限

• 所有问答 LLM 提示词注入只读约束（SYSTEM_SAFETY），禁止修改图谱数据；
• 权限模型（data/permissions.json：角色 × 意图 + 敏感字段）已就绪，permission_check 节点待接线；
• 完整安全方案见 docs/技术方案.md 第 11 章。

常见问题

现象	处理
Unable to retrieve routing information / 连接 7687 被拒	Neo4j 未启动；在 Neo4j Desktop 中 Start 数据库
端口无监听但服务已开	确认 .env 的 NEO4J_URI 与数据库实际端口一致
局域网其他人访问不到	启动加 --host 0.0.0.0，并放行 Windows 防火墙 8000 端口
回答数量对不上 / 结果为空	用 --debug 看执行详情（实参/丢弃参数/Cypher/行数）与检查报告
首次运行慢	API 服务启动时会预加载嵌入模型；首次业务请求仍可能有实体索引构建和 DeepSeek 调用，之后同进程会快
DMS token 过期	从 DMS 控制台重新复制 vuejs_token（见 scripts/generate_dms_supplement.py）