flowchart TD
    subgraph ingest["接入与特征层"]
        producer["RiskSignalProducer\n模拟上游信号"]
        kafkaRaw["Kafka\nraw_risk_signals\n64 partitions"]
        flink["Flink PyJob\n窗口聚合 · CEP · 异常评分"]
        kafkaEnrich["Kafka\nenriched_risk_signals"]
    end

    subgraph core["决策核心层"]
        case["case-service :8001\nKafka消费 · Redis幂等\nPostgreSQL存储"]
        router["fast-path-router :8007\n快慢路径分流"]
        orch["orchestrator-service :8002\nLangGraph StateGraph\n6节点 · Claude LLM"]
    end

    subgraph agents["智能体层（均含 Redis 缓存）"]
        rules["rules-explainer :8003\nv2.0 · Claude\n规则解释+监管映射"]
        score["score-analyst :8004\nv2.0 · Claude\n因子归因+置信区间"]
        graph_a["graph-risk :8008\nv2.0 · Neo4j\n图关系风险"]
        behavior["behavior-sequence :8009\nv2.0\n行为序列风险"]
        evidence["evidence-retriever :8005\nv3.0 · ES BM25\n知识库RAG检索"]
        verifier["verifier :8006\nv2.1 · Claude\n质量交叉验证"]
        policy["policy-calibration :8010\nv2.0\n策略校准+最终建议"]
    end

    subgraph storage["存储层"]
        pg[("PostgreSQL\n:5432\n案例持久化")]
        redis[("Redis\n:6379\nmaxmem 2gb")]
        es[("Elasticsearch\n:9200\nrisk_knowledge")]
        neo4j[("Neo4j\n:7474\n图关系")]
    end

    producer -->|"JSON signal"| kafkaRaw
    kafkaRaw --> flink
    flink -->|"+computed_features"| kafkaEnrich
    kafkaEnrich --> case
    case -->|"幂等检查"| redis
    case -->|"写入"| pg
    case --> router
    router -->|"FAST路径"| case
    router -->|"SLOW路径"| orch

    orch --> rules & score & graph_a & behavior
    orch --> evidence
    orch --> verifier
    orch --> policy
    orch -->|"缓存读写"| redis

    rules & score & verifier -->|"结果缓存"| redis
    evidence -->|"BM25检索"| es
    evidence -->|"结果缓存"| redis
    graph_a --> neo4j

    policy -->|"decision_support"| case

flowchart LR
    S(["START"]) --> route

    route{"route_node\n判断路径"}
    route -->|"fast_path=true\nrisk<0.45 单规则 无异常"| build

    route -->|"fast_path=false\n高风险 / 复杂信号"| parallel

    subgraph slow["慢路径（Slow Path）"]
        parallel["parallel_analysis_node\nasyncio.gather\n① rules-explainer\n② score-analyst\n③ graph-risk\n④ behavior-sequence"]
        ev["evidence_node\n⑤ evidence-retriever\nES BM25 + Redis缓存"]
        vf["verify_node\n⑥ verifier\nClaude 质量验证\n→ quality_score"]
        pol["policy_node\n⑦ policy-calibration\n置信度汇总\n→ final_recommendation"]
        parallel --> ev --> vf --> pol --> build
    end

    build["build_result_node\n组装 decision_support\nJSON"]
    build --> E(["END"])

    style S fill:#555,color:#fff,stroke:none
    style E fill:#555,color:#fff,stroke:none
    style route fill:#d4860a,color:#fff
    style parallel fill:#5b4fc9,color:#fff
    style build fill:#1a7d4a,color:#fff

sequenceDiagram
    participant Entry as GraphEntry
    participant Route as route_node
    participant Parallel as parallel_analysis_node
    participant Evidence as evidence_node
    participant Verify as verify_node
    participant Policy as policy_node
    participant Build as build_result_node

    Entry->>Entry: init(case_id, signal, trace_id, start_ms)
    Route->>Route: write(fast_path)
    alt slow_path
        Parallel->>Parallel: write(rules_explanation, score_analysis, graph_analysis, behavior_analysis)
        Parallel->>Parallel: incr(cache_hits, llm_calls)
        Evidence->>Evidence: write(evidence)
        Verify->>Verify: write(verification.quality_score, verification.issues)
        Policy->>Policy: write(policy_result, confidence_breakdown)
    end
    Build->>Build: write(decision_support)
    Build->>Build: append(errors?) and finalize(audit fields)

sequenceDiagram
    autonumber
    participant P as RiskSignalProducer
    participant K1 as Kafka<br/>raw_risk_signals
    participant FL as Flink PyJob
    participant K2 as Kafka<br/>enriched_risk_signals
    participant CS as case-service
    participant RD as Redis
    participant DB as PostgreSQL
    participant RT as fast-path-router
    participant OC as orchestrator<br/>(LangGraph)
    participant AG as agents<br/>(并发4个)
    participant EV as evidence-retriever
    participant VF as verifier

    P->>K1: publish(signal)
    K1->>FL: consume(raw_signal)
    FL->>FL: 计算 computed_features
    FL->>K2: publish(enriched_signal)

    K2->>CS: consume(enriched_signal)
    CS->>RD: GET case:created:{id}（幂等检查）
    alt 已处理
        RD-->>CS: "1" → 跳过
    else 新信号
        CS->>DB: INSERT case
        CS->>RD: SETEX case:created:{id} 86400
        CS->>RT: POST /route

        alt FAST路径
            RT-->>CS: route=FAST + decision_support
            CS->>DB: UPDATE decision_support
        else SLOW路径
            RT-->>CS: route=SLOW
            CS->>OC: POST /analyze

            OC->>OC: route_node（判断fast/slow）
            par 并发分析
                OC->>AG: POST /explain（rules-explainer）
                OC->>AG: POST /analyze（score-analyst）
                OC->>AG: POST /analyze-graph（graph-risk）
                OC->>AG: POST /analyze-sequence（behavior-sequence）
            end
            OC->>EV: POST /retrieve（ES BM25 检索）
            OC->>VF: POST /verify（quality_score 评估）
            OC->>OC: policy_node + build_result_node

            OC-->>CS: decision_support JSON
            CS->>DB: UPDATE decision_support
        end
    end

flowchart LR
    source[KafkaSource raw_risk_signals] --> parse[Parse JSON]
    parse --> keyBy[KeyBy entity_id]
    keyBy --> tw[TumblingWindow 60s]
    keyBy --> sw[SlidingWindow 300s/60s]
    keyBy --> sess[SessionWindow gap=120s]
    keyBy --> cep[CEP Pattern 5min]

    tw --> f1[tx_count_1min]
    sw --> f2[velocity_delta]
    sess --> f3[burst_flag]
    cep --> f4[cep_pattern_flag]
    keyBy --> f5[anomaly_score]

    f1 --> merge[Merge computed_features]
    f2 --> merge
    f3 --> merge
    f4 --> merge
    f5 --> merge
    merge --> sink[KafkaSink enriched_risk_signals]

flowchart TD
    msg[Consume enriched_risk_signals] --> idem{Redis key exists? case:created:id}
    idem -- yes --> skip[Skip message]
    idem -- no --> dbcheck{Case exists in DB?}
    dbcheck -- yes --> repair[Repair Redis key and skip]
    dbcheck -- no --> create[Insert case row + trace_id]
    create --> mark[SETEX case:created:id 86400]
    mark --> route[POST fast-path-router /route]
    route --> path{path_type}
    path -- FAST --> fastsave[Store decision_support directly]
    path -- SLOW --> orch[POST orchestrator /analyze]
    orch --> slowsave[Store decision_support from orchestrator]
    fastsave --> done[Done]
    slowsave --> done

flowchart LR
    input[Input rules_hit + summary] --> cache{Redis evidence_cache hit?}
    cache -- yes --> cached[Return cached top-k]
    cache -- no --> bm25[ES BM25 multi_match]
    bm25 --> boost_rule[Rule boost]
    boost_rule --> boost_cep[CEP co-occurrence boost +0.20]
    boost_cep --> avail{ES available?}
    avail -- no --> local[Local lexical/vector fallback]
    avail -- yes --> esres[ES candidates]
    esres --> merge[Merge + dedupe + rank]
    local --> merge
    merge --> topk[Top-5 evidence]
    topk --> writecache[Write evidence_cache TTL 600s]
    writecache --> output[Return RetrieveResponse]

flowchart TD
    qs[quality_score] --> d1{score >= 0.9}
    d1 -- yes --> excellent[Excellent: auto-pass verification]
    d1 -- no --> d2{score >= 0.7}
    d2 -- yes --> good[Good: keep model recommendation]
    d2 -- no --> d3{score >= 0.5}
    d3 -- yes --> acceptable[Acceptable: REVIEW with caution]
    d3 -- no --> human[NEEDS_HUMAN_REVIEW]

flowchart LR
    r[risk_score_component] --> agg[Confidence Aggregator]
    g[graph_component] --> agg
    b[behavior_component] --> agg
    e[evidence_component] --> agg
    v[verification_component] --> agg

    agg --> conf[calibrated_confidence]
    agg --> rec[final_recommendation]
    agg --> refs[policy_refs]