报告解析迁移到 LangGraph 实现计划

面向 AI 代理的工作者： 必需子技能：使用 superpowers:subagent-driven-development（推荐）或 superpowers:executing-plans 逐任务实现此计划。步骤使用复选框（- [ ]）语法来跟踪进度。

目标： 将 PDF 报告解析从 Java（PdfParseService）迁移到 LangGraph（Python 算法 + LLM 兜底），Java 侧通过 HTTP 调用 LangGraph 接口。

架构： Java 保存文件 → 调 LangGraph /api/v1/report/parse → LangGraph 先用 Python 算法解析（复用 extract_full_report_v5.py 核心逻辑），失败时 LLM 兜底 → 返回结构化 JSON。Java 保留本地解析作为 fallback。

技术栈： Python 3.11, PyPDF2, FastAPI, LangChain, httpx, Java Spring Boot

文件结构

创建的文件

文件	职责
`cfc-langgraph/app/agents/report_parse_agent.py`	报告解析 Agent：算法解析 + LLM 兜底
`cfc-langgraph/app/parsers/pdf_parser.py`	Python PDF 解析核心逻辑（从 `extract_full_report_v5.py` 提取）
`cfc-langgraph/app/parsers/__init__.py`	包初始化
`cfc-langgraph/app/api/report_parse.py`	`POST /api/v1/report/parse` 接口
`cfc-langgraph/tests/test_report_parse.py`	解析测试

修改的文件

文件	修改
`cfc-langgraph/app/main.py`	注册新的 report_parse router
`cfc-backend/.../controller/HealthReportController.java`	`parsePreview()` 先调 LangGraph，失败回退 Java

任务 1：提取 Python PDF 解析核心模块

文件：

创建：cfc-langgraph/app/parsers/__init__.py
创建：cfc-langgraph/app/parsers/pdf_parser.py
参考：docs/参考资料/extract_full_report_v5.py

[ ] 步骤 1：创建 parsers 包

mkdir -p /app/cfc/cfc-langgraph/app/parsers
touch /app/cfc/cfc-langgraph/app/parsers/__init__.py

[ ] 步骤 2：编写 pdf_parser.py 核心函数

从 extract_full_report_v5.py 提取核心逻辑，封装为 parse_report_pdf(file_path: str) -> dict：

"""
菌群报告 PDF 解析器
从 extract_full_report_v5.py 提取核心逻辑，封装为可调用函数
"""
import re, json
from PyPDF2 import PdfReader

# === 常量 ===
RADICAL_MAP = {
    '\u2f52': '\u6c0f', '\u2f51': '\u6bcd', '\u2f59': '\u6b6f',
    '\u2f04': '\u4e59', '\u2f20': '\u4e00', '\u2f21': '\u4e28',
    '\u2f22': '\u4e3f', '\u2f23': '\u4e39', '\u2f2b': '\u5c38',
    '\u2f2d': '\u5c71', '\u2f44': '\u4e59', '\u2f53': '\u6c14',
    '\u2f55': '\u6c34', '\u2f5c': '\u725b', '\u2f5f': '\u7389',
    '\u2f7a': '\u7f8a', '\u2f81': '\u8089', '\u2f83': '\u81ea',
    '\u2f8a': '\u8272', '\u2f8e': '\u8840', '\u2f95': '\u79be',
    '\u2f96': '\u8c46', '\u2faf': '\u9762', '\u2fb9': '\u9999',
    '\u2fca': '\u9ed1', '\u2ec9': '\u8d1d', '\u2edd': '\u98df',
    '\u2ee2': '\u9a6c', '\u2ee9': '\u9ec4',
}

KNOWN_MACRO = ['碳水化合物', '蛋白质', '脂肪', '纤维素', '乳制品']
KNOWN_AMINO = ['苏氨酸', '异亮氨酸', '亮氨酸', '赖氨酸', '蛋氨酸', '胱氨酸',
    '苯丙氨酸', '酪氨酸', '缬氨酸', '组氨酸', '丙氨酸', '丝氨酸', '甘氨酸',
    '脯氨酸', '谷氨酸', '天门冬氨酸', '天冬氨酸', '天冬酰胺', '谷氨酰胺',
    '精氨酸', '色氨酸']
KNOWN_VITAMINS = ['维生素A', '维生素B1', '维生素B2', '维生素B5', '维生素B6',
    '叶酸', '维生素B12', '维生素C', '维生素D', '维生素K2', '维生素E']
KNOWN_TRACE = ['铁', '锌']
KNOWN_DISEASE_RISKS = ['炎症性肠炎', '肠易激综合征', '感染性腹泻', '自闭症',
    '抑郁症', '甲状腺疾病', '肺部感染或疾病', '自体免疫病', '结直肠癌',
    '肥胖', '便秘', '过敏', '失眠', '肝病', '肾病', '胃病', '胆病',
    '心脑血管疾病', 'II型糖尿病']
KNOWN_BARRIER = ['肠道炎症水平', '肠道产气', '肠道屏障', '脂多糖LPS',
    '次级胆汁酸', '对甲酚（p-Cresol）', '吲哚', '苯酚', '腐胺', '硫化氢', '尸胺']
KNOWN_SCFA = ['丁酸盐（Butyrate）', '丙酸盐（Propionate）', '乙酸盐（Acetate）', '异戊酸盐（Isovaleric）']
KNOWN_NEURO = ['血清素（5-HT）', 'γ-氨基丁酸（GABA）', '谷氨酸（Glutamate）',
    '色氨酸（Tryptophan）', 'DOPAC', '多巴胺', '组胺（Histamine）', '一氧化氮',
    '喹啉（Quinolinic）', '维生素K2', '肌醇（Inositol）', '肾上腺素',
    '去甲肾上腺素', '乙酰胆碱', '皮质醇']
KNOWN_ANTIBIOTICS = ['β-内酰胺酶类', '氨基糖苷类', '大环内酯类', '呋喃类',
    '喹诺酮类', '磺胺类', '甲氧苄啶类', '氯霉素类', '四环素类']
KNOWN_PATHOGENS = ['幽门螺杆菌', '艰难梭菌', '沙门氏菌', '志贺氏菌', '弯曲杆菌']


def norm(s):
    return ''.join(RADICAL_MAP.get(c, c) for c in s)


def detect_format(lines):
    """检测 triplet / inline 格式"""
    text = '\n'.join(lines)
    has_triplet = '指标范围' in text and '疾病风险评估' in text
    for line in lines:
        if len(line) > 15 and re.search(r'[\u4e00-\u9fff]+[\d.]+[\u4e00-\u9fff/]+', line):
            for known in KNOWN_DISEASE_RISKS:
                if known in line:
                    return 'inline'
    return 'triplet' if has_triplet else 'triplet'


def extract_text(file_path):
    """读取 PDF 并提取文本"""
    reader = PdfReader(file_path)
    lines = []
    for page in reader.pages:
        text = norm(page.extract_text() or '')
        for line in text.split('\n'):
            ls = line.strip()
            if ls:
                lines.append(ls)
    return lines


def parse_overview(lines):
    """提取报告概述"""
    text = '\n'.join(lines)
    r = {}
    m = re.search(r'编号[：:\s]*(\d+)', text)
    if m: r['report_number'] = m.group(1)
    m = re.search(r'姓名[：:\s]*([\u4e00-\u9fff]{2,10})', text)
    if m: r['person_name'] = re.sub(r'(编号|年龄|性别|备注|肠道).*', '', m.group(1))[:4]
    m = re.search(r'年龄[：:\s]*(\d+)', text)
    if m: r['age'] = int(m.group(1))
    m = re.search(r'性别[：:\s]*([\u4e00-\u9fff])', text)
    if m: r['gender'] = 'male' if m.group(1) == '男' else 'female'
    for kw in ['健康总分', '菌群健康', '慢病控制', '营养均衡', '肠道菌群平衡',
               '菌群多样性', '有益菌', '有害菌', '核心菌属']:
        m = re.search(rf'{kw}\s*(\d+)', text)
        if m: r[kw] = int(m.group(1))
    m = re.search(r'肠道预测年龄[：:\s]*([\d.]+)', text)
    if m: r['gut_age'] = m.group(1)
    m = re.search(r'肠型[：:\s]*(\S+)', text)
    if m: r['gut_type'] = m.group(1)
    return r


def parse_triplet_until(lines, stop_markers):
    """三元组解析：3行一组 名称/数值/状态"""
    results = []
    i = 0
    while i < len(lines):
        if any(lines[i] == sm or lines[i].startswith(sm) for sm in stop_markers):
            break
        if lines[i] in ('指标范围', '名称', '丰度', '评估'):
            i += 1
            continue
        name = lines[i]
        if i + 2 >= len(lines): break
        val = lines[i + 1]
        status = lines[i + 2]
        if re.match(r'^-?\d+\.?\d*$', val):
            results.append({'name': name, 'value': val, 'status': status})
            i += 3
        else:
            i += 1
    return results


def parse_report_pdf(file_path: str) -> dict:
    """主函数：解析 PDF 返回结构化数据"""
    lines = extract_text(file_path)
    fmt = detect_format(lines)
    result = {'format': fmt, 'overview': parse_overview(lines)}

    # 疾病风险评估
    for i, line in enumerate(lines):
        if '疾病风险评估' in line and '注' not in line:
            risks, _ = parse_triplet_until(lines[i+1:],
                ['主要营养评估', '氨基酸评估', '维生素评估', '微量元素评估', '抗生素风险评估'])
            result['disease_risks'] = [r for r in risks if '注' not in r['name']]
            break

    # 主要营养评估
    for i, line in enumerate(lines):
        if '主要营养评估' in line:
            nutrients, _ = parse_triplet_until(lines[i+1:], ['氨基酸评估'])
            result['nutrition'] = nutrients[:5]
            break

    # 氨基酸评估
    for i, line in enumerate(lines):
        if '氨基酸评估' in line:
            aminos, _ = parse_triplet_until(lines[i+1:], ['维生素评估', '微量元素评估'])
            result['amino_acids'] = aminos
            break

    # 维生素评估
    for i, line in enumerate(lines):
        if '维生素评估' in line:
            vits, _ = parse_triplet_until(lines[i+1:], ['微量元素评估', '抗生素风险评估'])
            result['vitamins'] = [r for r in vits if '维生素' in r['name']]
            result['trace_elements'] = [r for r in vits if '维生素' not in r['name']]
            break

    return result


def parse_report_pdf_with_fallback(file_path: str) -> dict:
    """算法解析 + 简单校验，返回结构化数据"""
    result = parse_report_pdf(file_path)
    # 简单校验：如果关键字段缺失，标记为解析不完整
    if not result.get('overview', {}).get('overallScore') and \
       not result.get('overview', {}).get('健康总分'):
        result['_parse_incomplete'] = True
    return result

[ ] 步骤 3：运行测试验证导入

cd /app/cfc/cfc-langgraph && python -c "from app.parsers.pdf_parser import parse_report_pdf; print('OK')"

预期输出：OK

[ ] 步骤 4：Commit

git add cfc-langgraph/app/parsers/
git commit -m "feat: 提取 PDF 解析核心模块到 LangGraph parsers"

任务 2：创建报告解析 Agent

文件：

创建：cfc-langgraph/app/agents/report_parse_agent.py

[ ] 步骤 1：编写 ReportParseAgent

"""
报告解析 Agent：算法解析 + LLM 兜底
"""
import json
import logging
from typing import Optional
from app.config import settings
from app.parsers.pdf_parser import parse_report_pdf_with_fallback

logger = logging.getLogger(__name__)


class ReportParseAgent:
"""报告解析 Agent"""

def __init__(self):
    self.llm_api_key = getattr(settings, 'openai_api_key', '') or \
                       getattr(settings, 'dify_api_key', '')

async def parse(self, file_path: str) -> dict:
    """解析 PDF 报告，算法解析 + LLM 兜底"""
    # 1. 算法解析
    result = parse_report_pdf_with_fallback(file_path)
    logger.info("算法解析完成: format=%s, overview_keys=%d",
                 result.get('format'), len(result.get('overview', {})))

    # 2. 如果解析不完整，LLM 兜底
    if result.get('_parse_incomplete') or not result.get('disease_risks'):
        logger.info("算法解析不完整，尝试 LLM 兜底")
        llm_result = await self._parse_with_llm(file_path)
        if llm_result:
            # 合并 LLM 结果到算法结果上（LLM 覆盖缺失字段）
            for key in ['disease_risks', 'nutrition', 'amino_acids',
                        'vitamins', 'trace_elements', 'indicators']:
                if key in llm_result and not result.get(key):
                    result[key] = llm_result[key]
            if llm_result.get('overview'):
                for k, v in llm_result['overview'].items():
                    if k not in result.get('overview', {}):
                        result.setdefault('overview', {})[k] = v

    # 清理内部标记
    result.pop('_parse_incomplete', None)
    return result

async def _parse_with_llm(self, file_path: str) -> Optional[dict]:
    """LLM 兜底解析"""
    try:
        from PyPDF2 import PdfReader
        reader = PdfReader(file_path)
        text = '\n'.join(page.extract_text() or '' for page in reader.pages)

        # 构造 prompt
        prompt = f"""你是一个肠道菌群检测报告解析专家。请从以下PDF文本中提取结构化数据，返回JSON格式。

文本内容：
{text[:8000]}

请按以下JSON Schema返回：
{{
"overview": {{ "person_name": "", "report_number": "", "age": 0, "gender": "male/female",
    "overallScore": 0, "gutHealthScore": 0, "chronicDiseaseScore": 0, "nutritionScore": 0,
    "gutAge": "", "gutType": "" }},
"disease_risks": [{{"name": "", "value": "", "status": ""}}],
"nutrition": [{{"name": "", "value": "", "status": ""}}],
"amino_acids": [{{"name": "", "value": "", "status": ""}}],
"vitamins": [{{"name": "", "value": "", "status": ""}}],
"trace_elements": [{{"name": "", "value": "", "status": ""}}]
}}

只返回JSON，不要其他文字。"""

        if self.llm_api_key and 'openai' in settings.model_type.lower():
            # 调 OpenAI 兼容 API
            import httpx
            async with httpx.AsyncClient(timeout=60) as client:
                resp = await client.post(
                    f"{settings.llm_base_url}/chat/completions",
                    json={
                        "model": settings.llm_model or "gpt-4o",
                        "messages": [{"role": "user", "content": prompt}],
                        "temperature": 0.1,
                    },
                    headers={"Authorization": f"Bearer {self.llm_api_key}"},
                )
                data = resp.json()
                content = data['choices'][0]['message']['content']
                content = content.replace('```json', '').replace('```', '').strip()
                return json.loads(content)
        else:
            logger.warning("LLM 未配置，跳过 LLM 兜底")
            return None
    except Exception as e:
        logger.warning("LLM 解析失败: %s", e)
        return None

[ ] 步骤 2：Commit

git add cfc-langgraph/app/agents/report_parse_agent.py
git commit -m "feat: 创建 ReportParseAgent（算法解析 + LLM 兜底）"

任务 3：创建 LangGraph 解析 API

文件：

创建：cfc-langgraph/app/api/report_parse.py
修改：cfc-langgraph/app/main.py

[ ] 步骤 1：创建 report_parse.py API

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import Optional
from app.agents.report_parse_agent import ReportParseAgent
import logging
import os

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/v1", tags=["report_parse"])
_agent = None


def get_agent():
global _agent
if _agent is None:
    _agent = ReportParseAgent()
return _agent


class ParseRequest(BaseModel):
file_path: str
family_id: Optional[int] = None
user_id: Optional[int] = None


class ParseResponse(BaseModel):
code: int = 200
message: str = "ok"
data: dict = {}


@router.post("/report/parse", response_model=ParseResponse)
async def parse_report(req: ParseRequest):
if not os.path.exists(req.file_path):
    raise HTTPException(status_code=400, detail=f"文件不存在: {req.file_path}")

agent = get_agent()
try:
    result = await agent.parse(req.file_path)
    return ParseResponse(data=result)
except Exception as e:
    logger.error("报告解析失败: %s", e, exc_info=True)
    return ParseResponse(code=500, message=f"解析失败: {str(e)}", data={})

[ ] 步骤 2：注册到 main.py

# 在 app/main.py 的 import 后添加
from app.api import report_parse

# 在 include_router 中添加
app.include_router(report_parse.router)

[ ] 步骤 3：验证启动

cd /app/cfc/cfc-langgraph && python -c "
from app.agents.report_parse_agent import ReportParseAgent
agent = ReportParseAgent()
print('ReportParseAgent OK')
from app.api.report_parse import router
print('ReportParse router OK')
"

预期输出：ReportParseAgent OK ReportParse router OK

[ ] 步骤 4：Commit

git add cfc-langgraph/app/api/report_parse.py cfc-langgraph/app/main.py
git commit -m "feat: 创建 POST /api/v1/report/parse 接口"

任务 4：修改 Java 控制器调用 LangGraph

文件：

修改：cfc-backend/.../controller/HealthReportController.java
[ ] 步骤 1：在 controller 中添加 LangGraph 客户端调用

在 HealthReportController.java 中修改 parsePreview() 方法，在文件保存后先调 LangGraph：

// 在 parsePreview() 方法的 try 块中，替换 PDF 解析逻辑
// 保存文件到磁盘
String filePath = saveUploadFileToDisk(file, userId);

// 调 LangGraph 解析
try {
    // 构造 LangGraph 请求
    Map<String, Object> lgRequest = new HashMap<>();
    lgRequest.put("file_path", filePath);
    if (familyId != null) lgRequest.put("family_id", familyId);
    lgRequest.put("user_id", userId);

    // 发送 HTTP POST 到 LangGraph
    String langGraphUrl = "http://localhost:8000/api/v1/report/parse";
    // 使用 RestTemplate 或 WebClient 调用
    // 这里假设有 RestTemplate bean
    ResponseEntity<Map> lgResponse = restTemplate.postForEntity(
            langGraphUrl, lgRequest, Map.class);
    Map<String, Object> lgBody = lgResponse.getBody();
    if (lgBody != null && Integer.valueOf(200).equals(lgBody.get("code"))) {
        Map<String, Object> lgData = (Map<String, Object>) lgBody.get("data");
        // 将 LangGraph 返回的 data 转换为 ParsedReportPayload.Payload
        // 并构建返回结果
        // ... 转换逻辑略 ...
        return Result.success(buildResultFromLgData(lgData));
    }
} catch (Exception e) {
    log.warn("LangGraph 解析失败，回退到本地 Java 解析: {}", e.getMessage());
}

// Fallback: 本地 Java 解析（现有逻辑不变）
ParsedReportResult parsed = pdfParseService.parse(file.getInputStream());

步骤 2：添加 RestTemplate Bean（如不存在）

在 cfc-backend/.../config/ 中检查是否有 RestTemplate Bean，如无则添加：

@Bean
public RestTemplate restTemplate() {
    return new RestTemplate();
}

[ ] 步骤 3：编译验证

cd /app/cfc/cfc-backend && mvn clean compile

[ ] 步骤 4：Commit

git add cfc-backend/src/main/java/com/etotem/cfc/controller/HealthReportController.java
git commit -m "feat: parsePreview 先调 LangGraph 解析，失败回退 Java"

任务 5：端到端测试

[ ] 步骤 1：上传测试 PDF 验证解析

# 启动 LangGraph
cd /app/cfc/cfc-langgraph && uvicorn app.main:app --port 8000 &

# 启动 Java 后端
cd /app/cfc/cfc-backend && mvn spring-boot:run &

# 用 curl 测试上传
curl -X POST http://localhost:9082/api/health/report/parse-preview \
-F "file=@/app/cfc/docs/参考资料/501999942-某人.pdf" \
-F "familyId=1"

预期：返回结构化 JSON，包含 overview、disease_risks、nutrition 等字段

[ ] 步骤 2：验证三份参考 PDF

for pdf in /app/cfc/docs/参考资料/*.pdf; do
echo "=== Testing: $pdf ==="
curl -s -X POST http://localhost:9082/api/health/report/parse-preview \
-F "file=@$pdf" -F "familyId=1" | python3 -c "import sys,json; d=json.load(sys.stdin); print('code:', d.get('code')); print('overview:', list(d.get('data',{}).get('overview',{}).keys())[:5])"
done

[ ] 步骤 3：性能测试

# 测试 10 次并发
for i in $(seq 1 10); do
curl -s -o /dev/null -w "%{http_code} %{time_total}s\n" \
-X POST http://localhost:9082/api/health/report/parse-preview \
-F "file=@/app/cfc/docs/参考资料/501999942-某人.pdf" -F "familyId=1" &
done
wait

预期：全部返回 200，单次 < 5s

2026-08-04-report-parsing-migration.md 19 KB History Raw

报告解析迁移到 LangGraph 实现计划

文件结构

创建的文件

修改的文件

任务 1：提取 Python PDF 解析核心模块

任务 2：创建报告解析 Agent

任务 3：创建 LangGraph 解析 API

任务 4：修改 Java 控制器调用 LangGraph

任务 5：端到端测试

2026-08-04-report-parsing-migration.md 19 KB

History Raw