第02章 客户画像系统：用Hermes建立"了解你的客户"数据库
第02章 客户画像系统：用Hermes建立"了解你的客户"数据库
“你不能管理你不了解的事情。了解客户，是一切服务的起点。” —— 客户关系管理原则
“了解你的客户”（Know Your Customer，KYC）在金融行业是法规要求，但优秀的理财经理早在 KYC 成为合规条文之前，就已经把深入了解客户当作服务的核心。问题在于：如何在服务 200 位客户时，保持和服务 20 位时同样深入的"了解"？
Hermes 的客户画像系统就是答案。
2.1 客户画像的四个维度
# ====================================================
# 客户画像数据模型设计
# ====================================================
from dataclasses import dataclass, field
from typing import List, Optional, Dict
from enum import Enum
from datetime import date
import json
class RiskLevel(str, Enum):
"""客户风险承受等级（中国监管要求的五级分类）"""
CONSERVATIVE = "保守型" # R1 - 保本产品
CAUTIOUS = "谨慎型" # R2 - 固收+
BALANCED = "稳健型" # R3 - 均衡配置
AGGRESSIVE = "积极型" # R4 - 权益为主
SPECULATIVE = "激进型" # R5 - 高风险产品
class AssetClass(str, Enum):
"""资产类型"""
STOCK = "股票"
FUND = "基金"
BOND = "债券"
INSURANCE = "保险"
TRUST = "信托"
FOREIGN_EXCHANGE = "外汇"
REAL_ESTATE = "房产"
CASH = "现金及存款"
@dataclass
class AssetHolding:
"""单项资产持仓"""
asset_class: AssetClass
name: str # 产品/股票名称
code: str # 代码
amount: float # 持仓金额（万元）
cost_price: float # 成本价
current_price: float # 当前价
purchase_date: date
@property
def return_rate(self) -> float:
"""当前收益率"""
return (self.current_price - self.cost_price) / self.cost_price
@property
def current_value(self) -> float:
"""当前市值（万元）"""
return self.amount * (1 + self.return_rate)
@dataclass
class LifeEvent:
"""重要生命节点（主动服务的触发器）"""
event_type: str # "生日/结婚纪念/子女升学/退休计划/年终奖"
date: date
notes: str = ""
@dataclass
class ClientPreference:
"""客户偏好（决定沟通方式和内容）"""
preferred_channel: str # "微信/电话/邮件"
best_contact_time: str # "工作日上午/周末下午/..."
report_format: str # "详细PDF/简洁摘要/图表为主"
communication_frequency: str # "每周/每两周/每月"
interested_topics: List[str] # 感兴趣的市场话题
disliked_topics: List[str] # 不喜欢讨论的话题
language_style: str # "专业术语/通俗易懂"
@dataclass
class ClientProfile:
"""
完整客户画像
这是 Hermes Agent 所有决策的核心数据结构
"""
# 基本信息
client_id: str
name: str
age: int
occupation: str
city: str
# 财务状况
total_aum: float # 总资产管理规模（万元）
annual_income: Optional[float] # 年收入（万元）
holdings: List[AssetHolding] = field(default_factory=list)
# 风险画像
risk_level: RiskLevel = RiskLevel.BALANCED
risk_score: int = 60 # 0-100，量化风险承受度
investment_horizon: str = "3-5年" # 投资期限
# 偏好
preferences: Optional[ClientPreference] = None
# 生命节点
life_events: List[LifeEvent] = field(default_factory=list)
# 关系维护
relationship_manager: str = "" # 负责的理财经理
onboarding_date: Optional[date] = None
last_contact_date: Optional[date] = None
notes: str = "" # 理财经理手动备注
@property
def portfolio_summary(self) -> Dict:
"""资产组合摘要"""
by_class = {}
for holding in self.holdings:
cls = holding.asset_class.value
if cls not in by_class:
by_class[cls] = 0.0
by_class[cls] += holding.current_value
total = sum(by_class.values()) or 1
return {
cls: {"金额（万）": round(amount, 2), "占比": f"{amount/total:.1%}"}
for cls, amount in sorted(by_class.items(), key=lambda x: -x[1])
}
@property
def days_since_contact(self) -> Optional[int]:
"""距上次联系的天数"""
if self.last_contact_date:
return (date.today() - self.last_contact_date).days
return None
# 创建一个示例客户
example_client = ClientProfile(
client_id="C001",
name="张先生",
age=48,
occupation="民营企业主",
city="上海",
total_aum=850.0,
annual_income=200.0,
risk_level=RiskLevel.BALANCED,
risk_score=65,
investment_horizon="5-10年",
holdings=[
AssetHolding(
asset_class=AssetClass.FUND,
name="易方达蓝筹精选混合",
code="005827",
amount=200.0,
cost_price=2.50,
current_price=2.35,
purchase_date=date(2023, 6, 15),
),
AssetHolding(
asset_class=AssetClass.TRUST,
name="XX信托计划",
code="T2023001",
amount=300.0,
cost_price=1.0,
current_price=1.058,
purchase_date=date(2023, 1, 10),
),
AssetHolding(
asset_class=AssetClass.CASH,
name="活期存款+货基",
code="CASH",
amount=350.0,
cost_price=1.0,
current_price=1.015,
purchase_date=date(2024, 1, 1),
),
],
preferences=ClientPreference(
preferred_channel="微信",
best_contact_time="工作日上午10-11点",
report_format="简洁摘要+图表",
communication_frequency="每两周",
interested_topics=["A股市场", "美联储政策", "房产市场"],
disliked_topics=["加密货币"],
language_style="通俗易懂",
),
life_events=[
LifeEvent("生日", date(1976, 9, 15)),
LifeEvent("子女高考", date(2026, 6, 7), "孩子准备出国留学"),
],
relationship_manager="李小姐",
last_contact_date=date(2025, 6, 1),
notes="企业主，现金流充裕但不喜欢高波动产品。今年计划子女留学，需要美元资产配置建议。",
)
print("=== 客户画像示例 ===")
print(f"客户：{example_client.name}，{example_client.age}岁，{example_client.occupation}")
print(f"总资产：{example_client.total_aum} 万元")
print(f"风险等级：{example_client.risk_level.value}（评分：{example_client.risk_score}/100）")
print(f"\n资产组合：")
for asset_class, info in example_client.portfolio_summary.items():
print(f" {asset_class}: {info['金额（万）']}万 ({info['占比']})")
print(f"\n距上次联系：{example_client.days_since_contact} 天")
2.2 数据整合：从多个来源构建统一画像
# ====================================================
# 客户数据整合服务
# ====================================================
"""
数据来源整合（实际项目中的常见情况）：
来源1：核心系统（CRM/交易系统）
- 基本信息、KYC 数据
- 账户持仓、交易记录
- 合同、协议文件
来源2：外部市场数据
- 持仓产品的实时/历史行情
- 基金净值更新
来源3：沟通记录系统
- 历史通话录音（已转写）
- 微信/邮件沟通记录
- 理财经理的会议备忘
来源4：手动更新（理财经理输入）
- 客户透露的信息（家庭变化、未来计划）
- 会面印象
- 特殊需求记录
"""
import asyncio
from datetime import datetime, date
from typing import Optional
class ClientProfileService:
"""
客户画像服务
负责整合多源数据，维护客户画像的最新状态
"""
def __init__(self, db_session, market_data_client):
self.db = db_session
self.market = market_data_client
async def build_profile(self, client_id: str) -> ClientProfile:
"""
从多个数据源构建完整客户画像
（实际项目中各 fetch 替换为真实 API 调用）
"""
# 并行获取基础数据（无依赖关系的查询并行执行）
basic_info, holdings_raw, preferences = await asyncio.gather(
self._fetch_basic_info(client_id),
self._fetch_holdings(client_id),
self._fetch_preferences(client_id),
)
# 更新持仓的当前价格（串行，依赖 holdings_raw）
updated_holdings = await self._update_holding_prices(holdings_raw)
# 获取生命节点（可并行）
life_events = await self._fetch_life_events(client_id)
return ClientProfile(
client_id=client_id,
name=basic_info["name"],
age=basic_info["age"],
occupation=basic_info["occupation"],
city=basic_info["city"],
total_aum=sum(h.current_value for h in updated_holdings),
risk_level=RiskLevel(basic_info["risk_level"]),
risk_score=basic_info["risk_score"],
holdings=updated_holdings,
preferences=preferences,
life_events=life_events,
relationship_manager=basic_info["rm_name"],
last_contact_date=basic_info.get("last_contact_date"),
notes=basic_info.get("notes", ""),
)
async def update_contact_record(
self,
client_id: str,
channel: str,
summary: str,
rm_notes: str = "",
):
"""
记录每次客户沟通（让画像保持最新）
"""
record = {
"client_id": client_id,
"timestamp": datetime.now().isoformat(),
"channel": channel,
"summary": summary,
"rm_notes": rm_notes,
}
# 存储到数据库
await self.db.execute(
"INSERT INTO contact_records VALUES (:client_id, :timestamp, :channel, :summary, :rm_notes)",
record,
)
# 更新 last_contact_date
await self.db.execute(
"UPDATE clients SET last_contact_date = :today WHERE client_id = :client_id",
{"today": date.today(), "client_id": client_id},
)
async def add_life_event(
self,
client_id: str,
event_type: str,
event_date: date,
notes: str = "",
):
"""
添加生命节点（由理财经理手动输入，或从沟通记录中提取）
"""
await self.db.execute(
"""INSERT INTO life_events (client_id, event_type, event_date, notes)
VALUES (:client_id, :event_type, :event_date, :notes)""",
{
"client_id": client_id,
"event_type": event_type,
"event_date": event_date.isoformat(),
"notes": notes,
},
)
# -------- 私有方法（与实际系统集成时替换） --------
async def _fetch_basic_info(self, client_id: str) -> dict:
"""从 CRM 系统获取基本信息"""
# 实际实现：await self.db.fetchone("SELECT ... FROM clients WHERE ...")
return {
"name": "张先生",
"age": 48,
"occupation": "民营企业主",
"city": "上海",
"risk_level": "稳健型",
"risk_score": 65,
"rm_name": "李小姐",
"last_contact_date": date(2025, 6, 1),
"notes": "今年计划子女留学",
}
async def _fetch_holdings(self, client_id: str) -> list:
"""从交易系统获取持仓"""
return [] # 实际实现中返回 AssetHolding 列表
async def _update_holding_prices(self, holdings: list) -> list:
"""批量更新持仓当前价格"""
# 实际实现：调用市场数据 API 批量更新
return holdings
async def _fetch_preferences(self, client_id: str) -> ClientPreference:
"""获取客户偏好设置"""
return ClientPreference(
preferred_channel="微信",
best_contact_time="工作日上午10-11点",
report_format="简洁摘要+图表",
communication_frequency="每两周",
interested_topics=["A股市场", "美联储政策"],
disliked_topics=["加密货币"],
language_style="通俗易懂",
)
async def _fetch_life_events(self, client_id: str) -> list:
"""获取客户生命节点"""
return [
LifeEvent("生日", date(1976, 9, 15)),
LifeEvent("子女高考", date(2026, 6, 7)),
]
print("客户画像服务初始化完成")
print("支持：基本信息·持仓·偏好·生命节点 四维数据整合")
2.3 Hermes 如何使用客户画像
# ====================================================
# Hermes Agent 使用客户画像进行个性化决策
# ====================================================
import asyncio
import json
from openai import AsyncOpenAI
client = AsyncOpenAI()
def build_client_context(profile: ClientProfile) -> str:
"""
将客户画像转换为 Hermes 可以使用的上下文字符串
这是 System Prompt 的一部分
"""
upcoming_events = [
f"{e.event_type}（{e.date}）"
for e in profile.life_events
if (e.date - date.today()).days <= 30 # 30天内的事件
]
context = f"""
## 当前服务客户信息
**基本信息**
- 姓名：{profile.name}，{profile.age}岁，{profile.occupation}
- 城市：{profile.city}
- 风险等级：{profile.risk_level.value}（评分 {profile.risk_score}/100）
- 投资期限：{profile.investment_horizon}
**资产状况**
- 总资产规模：{profile.total_aum:.0f} 万元
- 主要持仓：{list(profile.portfolio_summary.keys())[:3]}
**沟通偏好**
- 首选渠道：{profile.preferences.preferred_channel if profile.preferences else '未设置'}
- 最佳联系时间：{profile.preferences.best_contact_time if profile.preferences else '未设置'}
- 报告风格：{profile.preferences.report_format if profile.preferences else '未设置'}
- 感兴趣话题：{', '.join(profile.preferences.interested_topics) if profile.preferences else '未设置'}
**重要备注**
{profile.notes}
**近期生命节点**
{', '.join(upcoming_events) if upcoming_events else '未来30天无特殊节点'}
**跟进状态**
- 距上次联系：{profile.days_since_contact} 天
- 建议联系频率：{profile.preferences.communication_frequency if profile.preferences else '每两周'}
"""
return context.strip()
async def hermes_personalized_response(
client_profile: ClientProfile,
user_query: str,
market_context: str = "",
) -> str:
"""
基于客户画像生成个性化回复
展示 Hermes 如何用画像定制每位客户的回复
"""
client_ctx = build_client_context(client_profile)
system_prompt = f"""你是一位专业的理财顾问 AI 助理，正在为特定客户服务。
{client_ctx}
服务原则：
1. 根据客户的风险等级和偏好定制建议，不推荐超出其风险承受能力的产品
2. 使用客户偏好的语言风格（{client_profile.preferences.language_style if client_profile.preferences else '通俗易懂'}）
3. 所有投资建议标注"仅供参考，最终决策请与您的理财顾问确认"
4. 不透露其他客户的信息
5. 复杂问题引导客户联系理财经理李小姐进行深入沟通"""
response = await client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_query},
],
max_tokens=600,
)
return response.choices[0].message.content
# 演示
async def demo():
query = "最近市场跌了不少，我的基金是不是该赎回？"
print(f"客户询问：{query}\n")
# 构建客户上下文
context = build_client_context(example_client)
print("=== 客户上下文（Hermes 看到的） ===")
print(context[:400], "...\n")
print("基于画像的个性化回复将根据张先生的：")
print(" - 风险等级（稳健型）：不建议激进操作")
print(" - 子女留学计划：考虑流动性需求")
print(" - 语言偏好（通俗易懂）：避免过多专业术语")
asyncio.run(demo())
本章小结
-
客户画像是 Hermes 做出好决策的基础：没有完整的客户画像，AI 只能给出通用的模板化建议；有了画像，才能真正做到"了解你的客户"，给出与其风险偏好、资产情况、生活阶段相匹配的个性化服务。
-
四维画像结构：基本信息与资产状况、风险偏好、沟通偏好、生命节点——这四个维度共同构成一个完整的客户视角。其中最容易被忽视的是"生命节点"，但恰恰是主动服务的最大机会来源。
-
数据整合必须是渐进式的：不要试图一次性完美地收集所有数据。从基本信息+持仓开始，在每次互动中逐渐补全偏好和生命节点信息。Hermes 能从沟通记录中自动提取新信息，持续丰富画像。
-
沟通偏好决定服务体验：同样的投资信息，用不同的渠道、不同的语言风格、在不同的时间发送，效果可以天壤之别。记录并遵守每位客户的偏好，是从"服务合格"到"服务优秀"的关键。
-
画像维护需要人机协作：自动化可以更新持仓行情、追踪市场信号，但很多重要的客户信息（最近换工作、刚离婚、打算移民）只有理财经理才知道。设计好"人工补充备注"的界面和流程，让画像真正保持最新。
# 核心行动：为你最重要的 10 位客户创建数字画像
# 从以下字段开始（15分钟内可以完成一位客户）
QUICK_PROFILE_TEMPLATE = {
"姓名": "",
"年龄": 0,
"职业": "",
"总资产（万）": 0,
"风险等级": "稳健型",
"主要持仓": [], # 最大的3项资产
"首选联系方式": "", # 微信/电话/邮件
"最佳联系时间": "",
"最近一次重要对话要点": "", # 记录客户说过的重要信息
"未来3个月的重要节点": [], # 生日/年终奖/子女考试等
}
本章提示词模板
【模板1：客户画像提取提示词】
以下是我和客户的一段沟通记录（已脱敏）：
{conversation_text}
请帮我提取以下信息，以JSON格式输出：
{
"风险偏好线索": [], // 客户说的话中透露的风险态度
"生活阶段信息": [], // 家庭、工作、子女等信息
"投资关注点": [], // 客户关心的市场或产品
"顾虑和担忧": [], // 客户提到的不安或担忧
"行动意向": [], // 客户有意向做的事
"下次跟进建议": "" // 建议的下次沟通重点
}
【模板2：画像完整性检查提示词】
我有一位客户的以下画像信息：
{client_profile_summary}
请评估这份画像的完整性：
1. 哪些关键信息是缺失的？（按重要性排序）
2. 基于现有信息，有哪些主动服务机会？
3. 下次与客户沟通时，建议收集哪些信息来补充画像？
4. 当前画像能支持哪些类型的个性化服务？哪些还不能支持？