企业级实践案例（增强多轮记忆 + 无结果短路 + 上下文感知版）
本系统是基于LangGraph框架构建的HR智能体，专为企业级部署设计，具备问题拆解、多轮对话记忆、意图检测和文档检索等功能。系统采用StateGraph架构，通过子问题拆分、分类、检索和合成等步骤，高效解答HR领域相关问题。
支持加载和处理TXT、Markdown、PDF和Word等多种格式文档
将复杂HR问题拆分为最多3个独立子问题，提高回答精确性
对子问题进行入职/薪资/请假等HR领域分类，提供针对性回答
基于OpenAI嵌入模型和Chroma向量数据库实现高效语义检索
保留最近5轮对话历史，支持基于历史对话生成简洁步骤
针对无结果查询和特定模式问题实现智能短路，提高用户体验
HR智能体采用LangGraph框架的StateGraph模型构建，实现了HR问答的全流程处理。系统由文档加载、向量化存储、状态管理和智能处理流程四大部分组成。
flowchart TB
subgraph 文档处理["文档处理层"]
A1[TXT加载器] --> B[文本分割器]
A2[MD加载器] --> B
A3[PDF加载器] --> B
A4[Word加载器] --> B
B --> C[向量化]
C --> D[Chroma向量库]
end
subgraph 状态流程["LangGraph状态流程"]
E[问题拆分节点] --> F[问题分类节点]
F --> G[文档检索节点]
end
subgraph 推理层["推理层"]
H1[子问题拆分链] -.-> E
H2[子问题分类链] -.-> F
H3[回答合成链] -.-> G
H4[步骤整理链] -.-> I
end
subgraph 交互层["交互层"]
I[多轮记忆]
J1[意图检测]
J2[实体短路]
K[命令行界面]
end
D -.-> G
G --> I
J1 --> I
J2 --> I
I --> K
style 文档处理层 fill:#e1f5fe,stroke:#4fc3f7
style 状态流程 fill:#e8f5e9,stroke:#66bb6a
style 推理层 fill:#fff8e1,stroke:#ffd54f
style 交互层 fill:#f3e5f5,stroke:#ce93d8
| 组件名称 | 功能描述 |
|---|---|
| 文档加载器 | 支持TXT、MD、PDF和Word四种文件格式，自动跳过Word临时锁文件 |
| 文本分割器 | RecursiveCharacterTextSplitter，将文档切分为1000字符的片段，100字符重叠 |
| 向量化与存储 | 使用OpenAI嵌入模型，Chroma数据库持久化存储文本向量 |
| 状态图节点 | 拆分、分类和检索三个主要节点，组成完整的处理流程 |
| LLM链 | 四个专用Chain：拆分链、分类链、合成链和步骤链 |
| 多轮记忆 | 支持最近5轮对话的记忆，智能管理上下文长度 |
| 短路机制 | 针对特定模式和无结果查询的智能处理逻辑 |
HR智能体的核心是基于LangGraph的StateGraph状态流程。系统接收用户查询，经过拆分、分类和检索三个核心节点进行处理，最终生成答案。下图展示了完整的状态转换流程：
stateDiagram-v2
[*] --> 拆分节点: 接收用户查询
state 拆分节点 {
[*] --> 调用拆分链: 输入原始查询
调用拆分链 --> 拆分成功: 最多3个子问题
拆分成功 --> [*]: 更新state.subqs
}
拆分节点 --> 分类节点: 提供子问题列表
state 分类节点 {
[*] --> 遍历子问题
遍历子问题 --> 调用分类链: 每个子问题
调用分类链 --> 分类完成: 获取类别标签
分类完成 --> [*]: 更新state.categories
}
分类节点 --> 检索节点: 提供子问题和类别
state 检索节点 {
[*] --> 遍历子问题处理
遍历子问题处理 --> 执行向量检索: 每个子问题
执行向量检索 --> 检查结果: 获取TopK相关文档
检查结果 --> 无结果短路: 未找到相关文档
检查结果 --> 调用合成链: 找到相关文档
无结果短路 --> 生成抱歉回复: "抱歉，没有相关信息"
调用合成链 --> 生成回答: 基于检索文档
生成抱歉回复 --> 回答完成
生成回答 --> 回答完成
回答完成 --> [*]: 更新state.answers
}
检索节点 --> [*]: 返回最终state
系统使用TypedDict定义了HR智能体的核心状态结构，包含四个关键字段：
class HRState(TypedDict):
query: str # 用户原始输入
subqs: List[str] # 拆分后的子问题列表
categories: List[str] # 每个子问题对应的分类标签
answers: List[str] # 每个子问题对应的回答
以下时序图展示了用户与HR智能体交互的完整过程，包括不同类型查询的处理流程：
sequenceDiagram
participant 用户
participant 主循环
participant 意图检测
participant StateGraph
participant 拆分节点
participant 分类节点
participant 检索节点
participant 向量数据库
participant 步骤链
用户->>主循环: 输入查询
alt 退出命令
主循环-->>用户: 退出程序
else 操作步骤查询
主循环->>意图检测: 检测步骤/流程意图
意图检测-->>主循环: 确认是步骤请求
主循环->>步骤链: 整理历史回答
步骤链-->>主循环: 返回1-2-3步骤
主循环-->>用户: 显示操作步骤
else 实体查询短路
主循环->>意图检测: 检测"XXX是谁"模式
意图检测-->>主循环: 匹配实体查询模式
主循环-->>用户: 无相关信息提示
else 常规HR问题
主循环->>StateGraph: 初始化状态并调用
StateGraph->>拆分节点: 转发查询
拆分节点->>拆分节点: 拆分子问题
拆分节点-->>StateGraph: 返回子问题列表
StateGraph->>分类节点: 传递子问题
分类节点->>分类节点: 分类子问题
分类节点-->>StateGraph: 返回分类结果
StateGraph->>检索节点: 传递问题和分类
检索节点->>向量数据库: 检索相关文档
向量数据库-->>检索节点: 返回文档片段
alt 有检索结果
检索节点->>检索节点: 合成答案
else 无检索结果
检索节点->>检索节点: 生成无结果提示
end
检索节点-->>StateGraph: 返回答案列表
StateGraph-->>主循环: 返回最终状态
主循环->>主循环: 更新历史记忆
主循环-->>用户: 显示回答
end
HR智能体由多个关键组件组成，下面详细介绍每个关键部分的实现细节和功能。
系统支持加载多种格式的HR文档，并进行自动分割、向量化处理：
def load_all_docs(doc_dir: str = "./hr_docs") -> List[Document]:
"""加载 hr_docs 目录下 txt/md/pdf/docx 文档，跳过 Word 临时锁文件"""
docs: List[Document] = [] # 初始化文档列表
# 加载 .txt 文档
txt_loader = DirectoryLoader(
doc_dir, glob="*.txt", loader_cls=TextLoader,
loader_kwargs={"encoding": "utf-8"}
)
docs.extend(txt_loader.load()) # 添加所有 .txt
# ... 加载其他格式文档 ...
return docs # 返回文档列表
# 文本拆分与向量化
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=100)
fragments = splitter.split_documents(docs) # 将文档拆分为多个片段
embeddings = OpenAIEmbeddings() # 初始化 OpenAI 嵌入模型
vectorstore = Chroma.from_documents(
fragments, embeddings, persist_directory="./hr_chroma_db"
) # 构建并持久化向量库
retriever = vectorstore.as_retriever(search_kwargs={"k": 5}) # 设置 top-5 检索
技术要点： 系统使用了四种专用加载器处理不同文档，通过glob模式匹配文件。对于Word文档，会专门过滤以"~$"开头的临时锁文件，提高系统稳定性。
系统定义了四个专用的LLM链，分别处理问题拆分、分类、回答合成和步骤整理：
# 子问题拆分链（最多拆 3 个）
decomp_prompt = PromptTemplate(
input_variables=["query"],
template=(
"请将以下复杂HR问题拆分为独立子问题，"
"最多拆出3条，每行一个，只输出子问题列表：\n{query}"
)
)
decompose_chain = LLMChain(
llm=ChatOpenAI(model_name="gpt-4o-mini", temperature=0),
prompt=decomp_prompt
)
# 子问题分类链
classify_prompt = PromptTemplate(
input_variables=["subq"],
template=(
"请判断以下子问题属于哪个HR类别："
"入职/薪资/请假/年度定级评估/升职/离职，"
"输出对应数字标签：\n{subq}"
)
)
classify_chain = LLMChain(
llm=ChatOpenAI(model_name="gpt-4o-mini", temperature=0),
prompt=classify_prompt
)
# ... 其他链定义 ...
系统定义了三个核心节点函数，作为StateGraph的处理单元：
def decompose_node(state: HRState) -> HRState:
"""拆分用户问题为最多3个子问题"""
subs = decompose_chain.predict(query=state["query"]).splitlines() # 调用拆分链
state["subqs"] = subs # 存储拆分结果
print(f"[调试] 拆分子问题：{subs}") # 输出调试信息
return state # 返回更新后的状态
def classify_node(state: HRState) -> HRState:
"""为每个子问题打上分类标签"""
cats: List[str] = [] # 初始化分类列表
for sub in state["subqs"]: # 遍历子问题
label = classify_chain.predict(subq=sub).strip() # 预测分类标签
cats.append(label) # 添加标签
state["categories"] = cats # 存储分类结果
print(f"[调试] 分类结果：{cats}") # 输出调试信息
return state # 返回更新后的状态
def retrieve_node(state: HRState) -> HRState:
"""检索相关文档并生成回答，若无命中则短路"""
answers: List[str] = [] # 初始化答案列表
for sub in state["subqs"]: # 遍历子问题
docs_hit = retriever.get_relevant_documents(sub) # 执行检索
if not docs_hit: # 如果没有检索到任何文档
answers.append(f"抱歉，我的知识库中没有"{sub}"的相关信息。") # 添加无结果提示
print(f"[调试] 子问题"{sub}"无检索结果，已短路") # 输出调试信息
continue # 跳过后续合成
snippets = "\n".join([d.page_content[:200] for d in docs_hit]) # 拼接文档摘要
ans = synthesize_chain.predict(subq=sub, docs=snippets) # 调用合成链生成答案
answers.append(ans) # 添加答案
print(f"[调试] 子问题"{sub}"回答：{ans}") # 输出调试信息
state["answers"] = answers # 存储所有回答
return state # 返回更新后的状态
系统将上述节点组合成一个完整的状态流图：
# 编排 StateGraph
graph = StateGraph(HRState) # 创建 StateGraph
graph.add_node("decompose", decompose_node) # 注册拆分节点
graph.add_node("classify", classify_node) # 注册分类节点
graph.add_node("retrieve", retrieve_node) # 注册检索节点
graph.add_edge(START, "decompose") # START → decompose
graph.add_edge("decompose", "classify") # decompose → classify
graph.add_edge("classify", "retrieve") # classify → retrieve
graph.add_edge("retrieve", END) # retrieve → END
app = graph.compile() # 编译流程
系统实现了多轮对话记忆，能够根据用户意图提供不同处理：
def main():
"""启动命令行交互，支持多轮记忆、无结果短路与简洁步骤输出"""
history: List[List[str]] = [] # 存储每轮的回答列表
max_rounds = 5 # 最多记忆 5 轮
token_limit = 2048 # 上下文 token 上限
while True:
query = input("用户: ").strip() # 读取并清理输入
if query.lower() in ["exit", "quit"]: # 退出检测
print("退出程序") # 打印退出提示
break # 跳出循环
# 简洁步骤分支
if re.search(r"流程|步骤|怎么做|重点|简洁", query):
if not history: # 如果没有上下文
print("请先提出一个具体问题，然后再要求简洁或重点。") # 提示用户
continue # 跳过后续
# 将历史回答平铺成文本
all_text = "\n".join(ans for round_ans in history for ans in round_ans)
# 如果超出 token 限制，则丢弃最早轮次
while count_tokens(all_text) > token_limit and len(history) > 1:
history.pop(0) # 弹出最早一轮
all_text = "\n".join(ans for round_ans in history for ans in round_ans)
steps = action_chain.predict(answers=all_text) # 生成步骤列表
print("操作步骤：") # 打印标题
for line in steps.splitlines(): # 分行输出
print(line)
continue # 跳过常规流程
# 实体查询短路
if re.match(r".+是谁($|\?)", query): # 匹配"XXX是谁"模式
print(f"抱歉，我没有关于"{query}"的相关信息。") # 无资料提示
continue # 跳过后续
# 常规 StateGraph 流程
init_state: HRState = { # 初始化状态
"query": query,
"subqs": [],
"categories": [],
"answers": []
}
result_state = app.invoke(init_state) # 执行状态图流程
answers = result_state["answers"] # 获取本轮回答
# 多轮记忆管理
history.append(answers) # 添加到历史
if len(history) > max_rounds: # 超出轮数
history.pop(0) # 丢弃最早一轮
# 输出回答
print("Agent 回答:") # 打印前缀
for ans in answers: # 遍历本轮回答
print(f"- {ans}") # 分行输出
下面是HR智能体的交互方法示例。
HR智能体支持以下几种交互模式：
| 输入类型 | 示例输入 | 系统响应 |
|---|---|---|
| 常规问题 | 我需要申请产假，流程是什么？ |
- 产假申请需先填写《休假申请表》，说明预产期和计划休假时间 - 申请表需经直属上级和HR部门审批 - 正常情况下，女性员工可享受98天法定产假 |
| 复杂问题 | 我是新员工，如何入职并领取工资卡？ |
- 首先，您需要在入职当天带齐所有证件到HR部门办理入职手续 - HR会协助您填写入职表格，签订劳动合同 - 工资卡领取需填写《银行卡申请表》，财务部门会帮您办理 |
| 步骤请求 | 请给我详细步骤 |
操作步骤： 1. 准备身份证、学历证明等个人证件 2. 入职当天到HR部门完成入职登记 3. 签署劳动合同和保密协议 4. 填写《银行卡申请表》 5. 等待财务部门通知领取工资卡 |
| 无结果查询 | 张三是谁？ | 抱歉，我没有关于"张三是谁？"的相关信息。 |
| 退出命令 | exit | 退出程序 |