LLMuserprofile

基于LLM的用户画像系统

课程目标

掌握基于LLM的客户画像系统设计
掌握基于langchain与langgraph进行客户画像系统开发的能力

LLM与用户画像系统

在数据驱动的时代，深刻理解用户是业务增长的核心。传统用户画像系统在处理和理解海量、复杂的非结构化数据时面临瓶颈，而大语言模型（LLM）的出现为此带来了革命性的突破。本届课程我们将系统地阐述如何构建一个基于 LLM 的新一代用户画像系统，实现从“知道用户行为”到“理解用户内心”的跨越。

特征	传统用户画像	LLM 增强的用户画像
数据源	以结构化数据为主	结构化 + 海量非结构化数据（文本、语音）
标签生成	依赖规则和手动定义	自动化、深层次、语义化的标签提取
洞察深度	行为的“是什么”（What）	行为背后的“为什么”（Why）和意图
画像性质	相对静态、更新慢	动态演进、“鲜活”的，实时更新
交互方式	依赖数据分析师和固定报表	支持自然语言查询，人人可用

基于LLM的客户画像系统完整架构

数据采集与整合层

系统的准确性高度依赖于输入数据的质量和广度。这一层是所有后续分析的基础。

结构化数据

交易数据：购买历史、订单金额、购买频率、产品偏好等。
用户基本信息：年龄、性别、地理位置、会员等级等。
行为数据：App/网站浏览历史、点击流、停留时长、互动记录（点赞、收藏、分享）。

非结构化数据

文本数据：客服聊天记录、产品评论、社交媒体帖子、在线问卷的开放式回答等。这是 LLM 发挥最大价值的地方，能从中挖掘出深度的观点和情感。
语音数据：客服通话录音（需预处理为文本或者通过多模态模型进行处理）。
视频、图片等等

数据整合

数据清洗：处理缺失值、异常值和重复数据，保证数据质量。
身份识别（ID-Mapping）：通过统一的ID（如用户ID、手机号）将来自不同渠道的数据关联到同一个用户身上，形成一个 360度的用户视图。
数据预处理：对文本、语音等非结构化数据进行预处理，例如语音转文本、文本分句等，为后续 LLM 处理做准备。

画像加工层

这是系统的核心引擎，负责将原始数据转化为有意义的用户标签和洞察。

基础标签提取

显式标签

从用户注册信息、问卷等直接获取的明确信息。示例：年龄段：25 - 30，性别：女。

隐式标签

通过 LLM 从非结构化数据中提炼。

实体识别（NER）：识别对话中提到的品牌、产品、人物、地点等。“我下周想去上海迪士尼玩”→提炼出目的地：上海迪士尼。
关键词提取：抓取用户在高频讨论或搜索的核心词汇。

深度意图与偏好分析

意图识别

分析用户近期行为和言论，判断其短期意图。用户搜索“XX 手机测评”和“XX 手机价格”→ LLM 判断用户处于购买决策阶段，意图是购买手机。

兴趣偏好

长期、持续地分析用户产生的内容，挖掘其深层兴趣点。用户持续发表徒步、登山相关内容的帖子 → LLM 可打上兴趣：户外运动、子兴趣：登山的标签。

消费能力与习惯

结合用户的消费记录、讨论的品牌、生活方式等，推断其大致的消费水平。LLM 可推断用户是价格敏感型还是品质追求型。

情绪与价值观分析

情绪分析

精准识别用户在评论、反馈中的情绪（喜、怒、哀、惊、中立等），并可进行多维度情绪分析。

价值观与性格推断

这是 LLM 的进阶能力。通过分析用户的语言风格、关注的社会议题、对事物的看法，可以推断其价值观和性格。示例：环保主义者、科技乐观派，甚至推断其 MBTI性格类型等。

用户分群与动态画像

智能分群

基于 LLM 生成的丰富标签，进行多维度的自动化用户分群，而非传统的手动规则分群。

画像动态更新

用户画像不是静止的。LLM 可以持续“阅读”用户的新数据，实时更新其标签和状态。一个用户的 “学生” 标签，可能在几个月后根据其言论（如“找工作”、“写论文”）更新为 “准毕业生”。

用户生命周期预测

结合历史数据和当前状态，预测用户所处的生命周期阶段（如新用户、成长期、流失预警期）。

应用层

让画像数据真正服务于业务，产生实际效益。

个性化推荐

内容推荐

基于用户深度兴趣标签，推荐更精准的文章、视频、音乐。

商品推荐

结合用户意图和消费偏好，进行商品推荐，甚至生成个性化的推荐语。

智能营销

精准广告投放

根据用户画像，在合适的渠道、合适的时间推送最相关的广告。

个性化营销文案生成

利用 LLM 的文本生成能力，为不同用户群体自动生成不同风格、不同侧重点的营销文案。

可视化与查询

画像仪表盘

以卡片、图表等形式，直观展示单个用户的立体画像。

自然语言查询

支持运营人员使用自然语言进行用户查询，极大降低数据使用门槛。业务人员可以直接在查询框输入：“筛选出最近一个月对‘价格’有负面情绪，并且是‘户外运动爱好者’的用户”LLM 会将此自然语言转化为数据查询指令并返回结果。

用户画像生成

核心技术栈

环境依赖： uv +python3.13+
LLM应用框架： langchain + langgraph

用户画像生成工作流设计

以下是整理后的全部文字和代码内容：

初始化状态

Langgraph的工作流核心状态，用于保存每个节点分析后的整体分析结果

class ProfileWorkflowState(TypedDict):
    """工作流状态定义"""
    user_id: str
    input_data: Dict[str, Any]
    extracted_tags: List[UserTag]
    analyzed_intents: List[UserIntent]
    analyzed_emotions: List[UserEmotion]
    user_profile: UserProfile | None
    error_messages: List[str]
    workflow_status: str
    messages: Annotated[List, add_messages]

节点 - validate_input

数据输入校验节点，对一些非法数据进行一定程度的安全校验，并且将错误信息保存到state中

def _validate_input(state: ProfileWorkflowState) -> ProfileWorkflowState:
    """验证输入数据"""
    user_id = state["user_id"]
    input_data = state["input_data"]

    state["messages"].append(HumanMessage(content=f"开始为用户 {user_id} 生成画像"))

    if not user_id:
        state["error_messages"].append("用户ID不能为空")
        state["workflow_status"] = "error"
        return state

    if not input_data:
        state["error_messages"].append("输入数据不能为空")
        state["workflow_status"] = "error"
        return state

    state["messages"].append(AIMessage(content="输入数据验证通过"))
    return state

节点 - extract_explicit_tags

显式标签提取，用于根据结构化的数据进行具体的标签提取

def _extract_explicit_tags(self, state: ProfileWorkflowState) -> ProfileWorkflowState:
    """提取显式标签"""
    logger.info("开始提取显式标签")
    try:
        input_data = state["input_data"]

        state["messages"].append(AIMessage(content="开始提取显式标签"))

        # 提取显式标签
        explicit_tags = self.tag_extractor.extract_tags(
            input_data, TagType.EXPLICIT
        )

        state["extracted_tags"].extend(explicit_tags)
        state["messages"].append(AIMessage(
            content=f"成功提取 {len(explicit_tags)} 个显式标签"
        ))
    except Exception as e:
        logger.exception(f"显式标签提取失败: {str(e)}")
        state["error_messages"].append(f"显式标签提取失败: {str(e)}")
        state["messages"].append(AIMessage(content=f"显式标签提取出错: {str(e)}"))

    return state

节点 - extract_implicit_tags

隐式标签提取，通过大模型分析语义，来获取语义背后的深层次标签

def _extract_implicit_tags(self, state: ProfileWorkflowState) -> ProfileWorkflowState:
    """提取隐式标签"""
    logging.info("开始提取隐式标签")
    try:
        input_data = state["input_data"]

        state["messages"].append(AIMessage(content="开始提取隐式标签"))
        # 提取隐式标签
        implicit_tags = self.tag_extractor.extract_tags(input_data, TagType.IMPLICIT)
        state["extracted_tags"].extend(implicit_tags)
        state["messages"].append(AIMessage(
            content=f"成功提取 {len(implicit_tags)} 个隐式标签"
        ))

    except Exception as e:
        logger.exception(f"隐式标签提取失败: {str(e)}")
        state["error_messages"].append(f"隐式标签提取失败: {str(e)}")
        state["messages"].append(AIMessage(content=f"隐式标签提取出错: {str(e)}"))

    return state

节点 - analyze_intents

用户意图分析，基于llm来根据用户的一些交流数据，来分析用户的一些意图标签

def _analyze_intents(self, state: ProfileWorkflowState) -> ProfileWorkflowState:
    """分析用户意图"""
    logging.info("开始分析用户意图")
    try:
        input_data = state["input_data"]
        state["messages"].append(AIMessage(content="开始分析用户意图"))

        # 分析用户意图
        intents = self.intent_analyzer.analyze_user_intent(input_data)
        state["analyzed_intents"].extend(intents)

        state["messages"].append(AIMessage(
            content=f"成功分析 {len(intents)} 个用户意图和相关偏好标签"
        ))

    except Exception as e:
        logger.exception(f"用户意图分析失败: {str(e)}")
        state["error_messages"].append(f"意图分析失败: {str(e)}")
        state["messages"].append(AIMessage(content=f"意图分析出错: {str(e)}"))

    return state

节点 - analyze_emotions

用户情绪分析，基于llm来根据用户的一些评论数据，分析用户表达的情感倾向

def _analyze_emotions(self, state: ProfileWorkflowState) -> ProfileWorkflowState:
    """分析用户情绪"""
    logging.info("开始分析用户情绪")
    try:
        input_data = state["input_data"]

        state["messages"].append(AIMessage(content="开始分析用户情绪"))

        # 分析用户情绪
        emotions = self.sentiment_analyzer.analyze_user_emotions(input_data)
        state["analyzed_emotions"].extend(emotions)
        state["messages"].append(AIMessage(
            content=f"成功分析 {len(emotions)} 条情绪记录"
        ))

    except Exception as e:
        logger.exception(f"用户情绪分析失败: {str(e)}")
        state["error_messages"].append(f"情绪分析失败: {str(e)}")
        state["messages"].append(AIMessage(content=f"情绪分析出错: {str(e)}"))

    return state

节点 - analyze_values

基于llm进行用户的价值观分析

def _analyze_values(self, state: ProfileWorkflowState) -> ProfileWorkflowState:
    """分析价值观"""
    try:
        input_data = state["input_data"]

        state["messages"].append(AIMessage(content="开始分析用户价值观"))

        # 分析价值观
        value_tags = self.sentiment_analyzer.analyze_user_values(input_data)
        state["extracted_tags"].extend(value_tags)

        state["messages"].append(AIMessage(
            content=f"成功分析价值观，生成 {len(value_tags)} 个价值观标签"
        ))

    except Exception as e:
        state["error_messages"].append(f"价值观分析失败: {str(e)}")
        state["messages"].append(AIMessage(content=f"价值观分析出错: {str(e)}"))

    return state

节点 - generate_profile

结合前面所有的数据分析节点，生成用户画像

def _generate_profile(self, state: ProfileWorkflowState) -> ProfileWorkflowState:
    """生成用户画像"""
    try:
        user_id = state["user_id"]
        extracted_tags = state["extracted_tags"]
        analyzed_intents = state["analyzed_intents"]
        analyzed_emotions = state["analyzed_emotions"]

        state["messages"].append(AIMessage(content="开始生成用户画像"))

        # 计算画像完整度
        completeness = self._calculate_completeness(extracted_tags)

        # 创建用户画像
        user_profile = UserProfile(
            user_id=user_id,
            tags=extracted_tags,
            emotions=analyzed_emotions,
            intents=analyzed_intents,
            segments=[],
            profile_completeness=completeness,
            last_activity=datetime.now(),
            created_at=datetime.now(),
            updated_at=datetime.now(),
            metadata={"workflow_version": "1.0", "generation_time": datetime.now().isoformat()}
        )

        state["user_profile"] = user_profile
        state["messages"].append(AIMessage(
            content=f"成功生成用户画像，包含 {len(extracted_tags)} 个标签，完整度 {completeness:.2f}"
        ))

    except Exception as e:
        logger.exception(f"画像生成失败: {str(e)}")
        state["error_messages"].append(f"画像生成失败: {str(e)}")
        state["messages"].append(AIMessage(content=f"画像生成出错: {str(e)}"))

    return state

节点 - segment_user

基于用户画像，进行用户画像分群

def _segment_user(self, state: ProfileWorkflowState) -> ProfileWorkflowState:
    """用户分群"""
    try:
        user_profile = state["user_profile"]
        if not user_profile:
            state["error_messages"].append("用户画像为空，无法进行分群")
            return state

        state["messages"].append(AIMessage(content="开始进行用户分群"))

        # 更新用户分群
        segments = self.segmentation_engine.update_user_segments(user_profile)
        user_profile.segments = segments

        state["user_profile"] = user_profile
        state["messages"].append(AIMessage(
            content=f"用户分群完成，分配到 {len(segments)} 个分群"
        ))

    except Exception as e:
        logger.exception(f"用户分群失败: {str(e)}")
        state["error_messages"].append(f"用户分群失败: {str(e)}")
        state["messages"].append(AIMessage(content=f"用户分群出错: {str(e)}"))

    return state

以下是图片中的实体信息（代码及说明文字）整理：

节点 - validate_profile

对整体生成的用户画像进行校验，分析其完整性和生成质量

def _validate_profile(self, state: ProfileWorkflowState) -> ProfileWorkflowState:
    """验证用户画像"""
    try:
        user_profile = state["user_profile"]

        if not user_profile:
            state["error_messages"].append("用户画像生成失败")
            state["workflow_status"] = "error"
            return state

        state["messages"].append(AIMessage(content="开始验证用户画像"))

        # 验证画像质量
        validation_results = self._validate_profile_quality(user_profile)

        if validation_results["is_valid"]:
            state["workflow_status"] = "completed"
            state["messages"].append(AIMessage(content="用户画像验证通过，生成完成"))
        else:
            state["error_messages"].extend(validation_results["errors"])
            state["workflow_status"] = "error"
            state["messages"].append(AIMessage(
                content=f"用户画像验证失败：{validation_results['errors']}"
            ))

    except Exception as e:
        state["error_messages"].append(f"画像验证失败：{str(e)}")
        state["workflow_status"] = "error"
        state["messages"].append(AIMessage(content=f"画像验证出错：{str(e)}"))

    return state

节点 - error_handler

用于在节点最后进行所有的异常统一处理，并更新到全局状态中

def _error_handler(state: ProfileWorkflowState) -> ProfileWorkflowState:
    """错误处理"""
    state["workflow_status"] = "error"
    error_summary = "; ".join(state["error_messages"])
    state["messages"].append(AIMessage(content=f"工作流执行失败：{error_summary}"))
    return state

核心处理逻辑分析 - 标签提取

显式标签提取：基于特定的规则来生成标签。与传统标签提取的方式比较类似

这些内容主要是用户画像生成工作流中，画像校验、异常处理节点的代码实现，以及核心逻辑里标签提取的相关说明。

Name		Name	Last commit message	Last commit date
Latest commit History 6 Commits
.cursor		.cursor
src		src
.gitignore		.gitignore
.python-version		.python-version
README.md		README.md
pyproject.toml		pyproject.toml
uv.lock		uv.lock

taoboy/LLMuserprofile

Folders and files

Latest commit

History

Repository files navigation