group_fqcd_jr/开发文档/D7.4-开发引导.md

智能财富管家系统 — 开发实施引导

体系编号：D7.4　·　域：七、早期系统文档　·　编号体系见 D1.1 §4.0

⚠️ 文档状态标注（2026-09-17 文档整理 · CS-DOC-2026-017）

状态：已被现行开发计划覆盖 —— 开工请勿依据本文档。

• 覆盖关系：本文档（07-17，技术实施参考）的职责已由 客服agent/D2.3-客服Agent开发计划.html v1.0（批次划分、会签门、门禁、交付物）与 客服agent/D2.1-客服Agent执行Todolist.md v5.2（52 项逐项 DoD）承接。
• 口径差异：本文 Phase 划分基于早期系统模型（含投顾类 Agent、以 XX科技 为品牌），与本项目现行口径「南方基金（南方基金管理股份有限公司）· 热线 400-889-8899 · 官网 nffund.com」及「投顾模块已整体清除（CS-PURGE-2026-012/013）」不一致。
• 处置方式：按《文档规整方案》（CS-DOC-2026-014）D-2 —— 保留原文 + 加本标注，不做「只改品牌」（那会造出「品牌已对、业务仍旧」的更危险状态）。
• 权威入口：开发文档/D1.1-文档索引与权威声明.md 的「开工只读 5 份」。

本文档配套于：《XX科技·智能财富管家系统 — 项目开发需求文档》v1.0 文档性质：技术实施参考，非需求规格。提供代码示例、工具推荐、实现思路等开发辅助信息 适用对象：参与本项目开发的工程师（学生团队）

---

目录

• Phase 1：基础设施 + 智能客服Agent
• Phase 2：客户画像 + 数据分析Agent
• Phase 3：知识图谱 + 投顾/业务操作Agent
• Phase 4：风控预警 + 记忆体系
• Phase 5：系统集成 + 联调优化

---

Phase 1：基础设施 + 智能客服Agent

1.1 项目结构

严格按分层架构组织代码：

app/
├── api/                 # API路由层
│   ├── chat.py          # 对话接口
│   ├── knowledge.py     # 知识库管理
│   └── admin.py         # 管理接口
├── service/             # 业务逻辑层
│   ├── rag_service.py
│   ├── agent_service.py
│   └── memory_service.py
├── tool/                # 工具层
│   ├── document_parser.py
│   ├── embedding_tool.py
│   └── milvus_tool.py
├── model/               # 数据模型
│   ├── schemas.py       # Pydantic模型
│   └── entities.py      # 数据库ORM模型
├── config/              # 配置
│   ├── settings.py      # 环境配置
│   └── database.py      # 数据库连接
├── utils/               # 工具类
│   ├── response.py      # 统一响应格式
│   ├── exceptions.py    # 异常处理
│   └── logger.py        # 日志模块
└── main.py              # 入口文件

1.2 RAG 实现方案参考

方案	适用场景	核心代码
A — LangChain RetrievalQA（推荐）	快速上手，链式编排	retriever = Milvus.as_retriever() → qa = RetrievalQA.from_chain_type(llm, retriever=retriever)
B — LlamaIndex VectorStoreIndex	已有 LlamaIndex 经验	index = VectorStoreIndex.from_vector_store(milvus_collection) → query_engine = index.as_query_engine()
C — 自建 Pipeline	需要完全自定义控制流	Embedding → Milvus Search → Prompt拼装 → LLM调用

1.3 Embedding 模型选型

模型	维度	特点
text-embedding-ada-002（OpenAI）	1536	语义理解效果好，需 API Key
bge-large-zh（本地部署）	1024	中文效果优秀，无 API 费用，需 GPU

注意：选定后全局统一，所有 Milvus 集合使用相同维度。

1.4 分块策略

• FAQ 问答对：每个问答对作为一个独立 chunk，不分块
• 长文档：RecursiveCharacterTextSplitter（separator=\n\n，chunk_size=512，chunk_overlap=64）
• 元数据：保留标题层级信息

1.5 Redis 会话管理

import redis
import tiktoken

r = redis.Redis(host="localhost", port=6379, db=0)

# 写入消息
r.lpush(f"session:{session_id}:messages", json.dumps({"role": "user", "content": message}))

# 读取并截断
messages = r.lrange(f"session:{session_id}:messages}", 0, -1)
enc = tiktoken.get_encoding("cl100k_base")
total_tokens = 0
truncated = []
for msg in reversed(messages):
    total_tokens += len(enc.encode(msg))
    if total_tokens > 4096:
        break
    truncated.append(msg)
# truncated 为逆序，需反转回正序

---

Phase 2：客户画像 + 数据分析Agent

2.1 NL2SQL 实现参考

System Prompt 模板（动态 Schema 筛选）：

你是金融数据分析专家。根据以下数据库Schema生成SQL查询。
仅允许SELECT语句，禁止任何修改操作。
结果最多返回100行。

数据库Schema（仅包含与问题相关的表）：
CREATE TABLE sys_user (...);
CREATE TABLE fin_product (...);
...

Few-shot 示例：

自然语言	SQL
"AUM超过100万的客户数"	SELECT COUNT(*) FROM fin_customer_profile WHERE total_assets > 1000000
"客户张三的持仓"	SELECT p.product_name, h.shares, h.current_value FROM fin_holdings h JOIN fin_product p ON h.product_id = p.id WHERE h.customer_id = (SELECT id FROM sys_user WHERE real_name = '张三')

2.2 置信度计算（Phase 2 基础版）

Phase 2 阶段仅实现基础版本：来源初始值 + 简单时间衰减。完整实现见 Phase 4 F4.3。

SOURCE_INITIAL = {"风评问卷": 0.9, "AI对话提取": 0.6, "用户自述": 0.4, "默认值": 0.2}

def calc_base_confidence(source: str, age_days: int) -> float:
    base = SOURCE_INITIAL.get(source, 0.2)
    decay = max(0, 1 - age_days / 365 * 0.2)
    return max(0.0, min(1.0, base * decay))

2.3 画像研判规则

直接参考项目目录中的 D6.4.1-投资者风险画像研判规则.md，实现四维度加权打分逻辑。

---

Phase 3：知识图谱 + 投顾/业务操作Agent

3.1 Neo4j Python Driver

from neo4j import GraphDatabase

driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))

with driver.session() as session:
    result = session.run(
        "MATCH (c:Customer)-[:INVESTS_IN]->(p:Product) RETURN c.name, p.product_name"
    )
    for record in result:
        print(record["c.name"], record["p.product_name"])

3.2 GraphRAG 实现思路

1. 实体识别：LLM 或正则提取产品名、客户名、行业等实体
2. 图谱查询：构造 Cypher 语句获取关联实体和关系路径
3. 格式化为文本：将图谱查询结果格式化为自然语言描述
4. 上下文拼接：将图谱文本拼接到 RAG 检索结果中
5. 融合排序：向量 Score × 0.6 + 图谱 Score × 0.4，注入 LLM Prompt

3.3 NL2API Function Calling 示例

tools = [
    {
        "type": "function",
        "function": {
            "name": "purchase_product",
            "description": "为客户申购理财产品",
            "parameters": {
                "type": "object",
                "properties": {
                    "customer_id": {"type": "integer", "description": "客户ID"},
                    "product_code": {"type": "string", "description": "产品代码"},
                    "amount": {"type": "number", "description": "申购金额（元）"}
                },
                "required": ["customer_id", "product_code", "amount"]
            }
        }
    }
]

8 个业务 API 建议用 FastAPI 的 APIRouter 组织，每个 API 独立 endpoint，放在 app/api/operations/ 目录下。

---

Phase 4：风控预警 + 记忆体系

4.1 风控规则引擎（策略模式）

from abc import ABC, abstractmethod

class BaseRiskRule(ABC):
    rule_id: str
    rule_name: str
    risk_level: str

    @abstractmethod
    def evaluate(self, transaction: dict) -> bool: ...

class LargeTransactionRule(BaseRiskRule):
    rule_id = "R001"
    rule_name = "大额现金交易"
    risk_level = "低"
    threshold = 50000  # 5万元

    def evaluate(self, transaction: dict) -> bool:
        return transaction["amount"] >= self.threshold

class FrequentTransactionRule(BaseRiskRule):
    rule_id = "R003"
    rule_name = "频繁交易"
    risk_level = "中"

    def evaluate(self, transaction: dict) -> bool:
        # 7天内交易次数 >= 10
        return transaction["weekly_count"] >= 10

4.2 置信度计算完整实现

见需求文档 Phase 4 F4.3 中的 BaseConfidenceCalcTool、FinalConfidenceRankTool、MemoryUnitValidator 伪代码。

4.3 Redis Pub/Sub 示例

import json
import redis

# 发布（风控Agent）
redis_client.publish("event:risk_alert", json.dumps({
    "alert_id": 1001,
    "customer_id": 10001,
    "alert_level": "中",
    "trigger_rules": ["R003", "R007"]
}))

# 订阅（投顾Agent / 客服Agent）
pubsub = redis_client.pubsub()
pubsub.subscribe("event:risk_alert")
for message in pubsub.listen():
    if message["type"] == "message":
        data = json.loads(message["data"])
        handle_alert(data)

4.4 周期校准任务（APScheduler）

from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()
# 每周日凌晨3点执行全量置信度重算
scheduler.add_job(recalculate_confidence, 'cron', day_of_week='sun', hour=3)
scheduler.start()

# 手动触发接口
# POST /api/admin/recalculate-confidence

---

Phase 5：系统集成 + 联调优化

5.1 流式输出（SSE）

from fastapi.responses import StreamingResponse

@router.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def event_generator():
        async for chunk in llm.astream(prompt):
            yield f"data: {chunk.content}\n\n"
        yield "data: [DONE]\n\n"
    return StreamingResponse(event_generator(), media_type="text/event-stream")

5.2 Agent 路由（工厂模式）

AGENT_REGISTRY = {
    "customer": CustomerAgent,
    "advisor": AdvisorAgent,
    "risk": RiskAgent,
    "analyst": AnalystAgent,
    "operator": OperatorAgent,
}

def get_agent(agent_type: str):
    agent_class = AGENT_REGISTRY.get(agent_type)
    if not agent_class:
        raise ValueError(f"Unknown agent type: {agent_type}")
    return agent_class()

5.3 降级方案

组件	降级策略
LLM	OpenAI 不可用时切换到本地模型（如 Qwen）
Milvus	向量检索超时时用 MySQL LIKE 关键词查询
Neo4j	图谱查询超时时跳过图谱增强，仅使用 RAG 结果
Redis	连接失败时直连 MySQL，恢复后自动回填缓存

5.4 演示脚本建议

序号	场景	覆盖模块
1	客户A（高净值）完整旅程	开户→风评→咨询→申购→持仓
2	客户B（普通投资者）风控触发	大额交易→预警→工单处置
3	多Agent协作联动	客服→业务操作→风控联动
4	NL2SQL 数据查询	自然语言→SQL→结果解读
5	GraphRAG vs 纯RAG 对比	同一问题两种检索方式的效果差异

---

文档结束 — 开发实施引导完。本文档内容仅为技术参考，具体实现以《项目开发需求文档》中的功能需求和验收标准为准。