LangGraph 多智能体协作 + RAG 知识引擎 + 审批工作流 + 自动重试
本章带你完成 阶段 5：「智控」企业级智能工作流平台 项目： 基于 LangGraph 多智能体协作 构建一个具备 智能风险评估、多级审批（Human-in-the-Loop）、自动执行与重试、实时监控 的企业级工作流平台。
interrupt_before + MemorySaver 实现工作流暂停/恢复，支持跨天审批且状态不丢失。time.sleep()，而是利用 LangGraph 的 Checkpointer 持久化状态，服务重启也不会丢失流程进度。for 循环，而是图级别的条件边回环——每次重试都经过完整的状态更新和日志记录，便于排查问题。理解了以上 4 大核心能力后，下面把它们组装成一个完整可运行的项目。
zhikong-platform/
├── config.py # 全局配置
├── workflow_models.py # State 定义（类型约束）
├── knowledge_engine.py # 极简知识引擎（关键词匹配版）
├── workflow.py # LangGraph 工作流（含所有 Agents 实现）
├── api.py # FastAPI 接口
├── main.py # 启动入口 + 演示
├── requirements.txt
└── docs/
└── procurement_rules.txt # 企业规则文档import os
# === API 配置（演示版使用默认值即可，无需真实 Key）===
# 通义千问 API Key，演示环境使用 sk-demo 即可
DASHSCOPE_API_KEY = os.getenv("DASHSCOPE_API_KEY", "sk-demo")
# 使用的聊天模型
CHAT_MODEL = "qwen-plus"
# 使用的文本嵌入模型
EMBEDDING_MODEL = "text-embedding-v3"
# === 知识库配置 ===
# 文档分块大小（字符数）
CHUNK_SIZE = 500
# 分块重叠大小（字符数）
CHUNK_OVERLAP = 50
# 检索时返回的文档数量
TOP_K = 3
# 知识库文档存放目录
KNOWLEDGE_DIR = "./docs"
# === 工作流配置 ===
# 最大重试次数
MAX_RETRY = 3
# 自动审批金额阈值（超过此金额需要人工审批）
AUTO_APPROVE_THRESHOLD = 5000from typing import TypedDict, Annotated, List, Optional
import operator
class WorkflowState(TypedDict):
"""工作流状态定义（仅做类型约束，运行时请用 dict）
注意：TypedDict 只是提供类型提示，实际运行时使用普通字典
这样可以避免序列化问题，同时保持代码可读性
"""
# ---- 流程控制字段 ----
# 日志列表：使用 operator.add 实现自动累积
logs: Annotated[List[str], operator.add]
# 当前执行阶段：analyze（分析）/ approve（审批）/ execute（执行）/ done（完成）
current_phase: str
# 风险等级：low（低）/ medium（中）/ high（高）
risk_level: str
# 审批状态：pending（待审批）/ approved（已批准）/ rejected（已驳回）
approval_status: str
# 执行重试计数器
retry_count: int
# 错误信息（可选）
error: Optional[str]
# ---- 业务数据字段 ----
# 申请唯一标识符
request_id: str
# 申请类型：procurement（采购）/ expense（费用报销）/ leave（请假）
request_type: str
# 原始申请描述
request_detail: str
# 申请金额
amount: float
# 申请部门
department: str
# 从知识库检索到的规则上下文
knowledge_context: str
# 分析报告（Agent 生成的风险评估）
analysis_report: str
# 执行结果：success（成功）/ failed（失败）/ error（错误）
execution_result: str
# 最终执行报告（供用户查看）
final_report: str """
极简知识引擎（演示版）
特点：
- 无需 FAISS / Embedding / LLM 等复杂依赖
- 用关键词匹配实现类似 RAG 的效果
- 适合快速演示和学习 LangGraph 工作流
实际项目中可以替换为：
- Chroma + DashScope Embeddings（向量检索）
- FAISS + OpenAI Embeddings
- Elasticsearch
"""
import os
from dataclasses import dataclass
from config import KNOWLEDGE_DIR
@dataclass
class KnowledgeChunk:
"""知识块数据结构"""
source: str # 文档来源路径
content: str # 文档内容
class KnowledgeEngine:
"""知识引擎：负责文档加载和检索"""
def __init__(self):
self.chunks: list[KnowledgeChunk] = [] # 存储所有知识块
def build(self, docs_dir: str = KNOWLEDGE_DIR):
"""读取 docs 目录下的 .txt 文件，构建知识库
Args:
docs_dir: 文档目录路径，默认从 config.py 读取
"""
self.chunks = []
if not os.path.isdir(docs_dir):
print(f" ⚠ 知识库目录不存在: {docs_dir}")
return
# 遍历目录，读取所有 .txt 文件
for root, _, files in os.walk(docs_dir):
for name in files:
if not name.endswith(".txt"):
continue # 只处理 .txt 文件
path = os.path.join(root, name)
try:
with open(path, "r", encoding="utf-8") as f:
content = f.read().strip()
if content:
# 创建知识块并添加到列表
self.chunks.append(KnowledgeChunk(source=path, content=content))
except Exception as e:
print(f" ⚠ 读取失败: {path} | {e}")
print(f" 知识库就绪: {len(self.chunks)} 个文档")
def search(self, query: str, top_k: int = 3) -> str:
"""关键词匹配搜索（无需向量库）
Args:
query: 查询字符串，如 "采购 50000元 技术部"
top_k: 返回最相关的文档数量
Returns:
合并的文档内容字符串
"""
if not self.chunks:
return "知识库未初始化"
# 提取查询关键词
keywords = [k for k in query.replace("/", " ").split() if k]
# 计算每个文档的关键词匹配得分
scored: list[tuple[int, KnowledgeChunk]] = []
for c in self.chunks:
score = sum(1 for kw in keywords if kw in c.content)
scored.append((score, c))
# 按得分降序排序
scored.sort(key=lambda x: x[0], reverse=True)
# 选择得分最高的文档
picked = [c for score, c in scored[:top_k] if score > 0]
if not picked:
# 兜底：如果没有匹配到任何关键词，返回第一个文档
picked = scored[:1]
# 合并文档内容
return "\n---\n".join([p[1].content for p in picked])
# 全局单例：整个应用共享一个知识引擎实例
knowledge_engine = KnowledgeEngine()"""
LangGraph 工作流编排（核心文件）
功能：
- 包含所有 Agent 实现（analyst/approval/executor/monitor）
- 定义工作流路由逻辑（条件边）
- 构建和编译 StateGraph
- 无需额外导入 agents 包，所有逻辑内联在此文件
设计理念：
- 单文件包含所有核心逻辑，便于学习和理解
- 每个函数职责单一，易于测试和维护
- 使用 MemorySaver 实现状态持久化
"""
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from workflow_models import WorkflowState
from knowledge_engine import knowledge_engine
from config import MAX_RETRY, AUTO_APPROVE_THRESHOLD
import random
# ========== Agents 实现（内联在同一文件）==========
def analyst_node(state):
"""分析 Agent：基于金额阈值做风险分级（无需 LLM）
功能：
1. 从知识库检索相关规则
2. 根据金额进行风险分级
3. 决定是否需要人工审批
4. 生成分析报告
风险分级规则（演示版）：
- < 5000元：低风险，自动通过
- 5000-50000元：中风险，需要审批
- > 50000元：高风险，需要审批
"""
amount = float(state["amount"])
# 从知识库检索相关规则（模拟 RAG 效果）
context = knowledge_engine.search(
f"{state['request_type']} {state['request_detail']} 金额{amount}元"
)
# 风险分级逻辑
if amount < 5000:
risk = "low" # 低风险
elif amount <= 50000:
risk = "medium" # 中风险
else:
risk = "high" # 高风险
# 判断是否需要人工审批
need_approval = amount > AUTO_APPROVE_THRESHOLD
# 生成分析原因
reason = "按金额阈值进行风险分级（演示版，可替换为 LLM+RAG 推理）"
return {
"risk_level": risk,
"knowledge_context": context,
"analysis_report": f"风险: {risk} | {reason}",
"current_phase": "approve" if need_approval else "execute",
"logs": [f"[分析Agent] 风险: {risk} | {'需审批' if need_approval else '自动通过'}"]
}
def approval_node(state):
"""审批 Agent：设置等待审批状态
注意：此节点会通过 interrupt_before 机制暂停工作流，
等待人工审批后再继续执行
"""
return {
"approval_status": "pending",
"logs": ["[审批Agent] 工单已提交，等待人工处理"]
}
def approve_action(approved: bool, comment: str = ""):
"""审批动作：外部调用此方法完成人工审批
Args:
approved: True 表示批准，False 表示驳回
comment: 审批意见
Returns:
用于更新工作流状态的字典
"""
if approved:
return {
"approval_status": "approved",
"current_phase": "execute",
"logs": [f"[审批Agent] ✅ 已批准 {f'备注:{comment}' if comment else ''}"]
}
return {
"approval_status": "rejected",
"current_phase": "done",
"logs": [f"[审批Agent] ❌ 已驳回 原因:{comment}"]
}
def executor_node(state):
"""执行 Agent：模拟业务执行，带自动重试
功能：
1. 模拟实际业务操作（如调用外部 API）
2. 实现失败重试机制
3. 生成执行报告
重试策略：
- 第3次必定成功（模拟最终成功）
- 或 60% 概率成功（模拟真实场景）
"""
retry = int(state.get("retry_count", 0))
try:
# 模拟执行：第3次必定成功，或 60% 概率成功
success = retry + 1 >= 3 or random.random() > 0.4
if success:
# 生成成功报告
report = (
"执行报告(演示版)\n"
f"- 事项: {state['request_detail']}\n"
f"- 金额: ¥{state['amount']}\n"
"- 状态: 已完成"
)
return {
"execution_result": "success",
"final_report": report,
"current_phase": "done",
"logs": [f"[执行Agent] ✅ 第{retry+1}次成功"],
}
# 执行失败，准备重试
return {
"execution_result": "failed",
"retry_count": retry + 1,
"logs": [f"[执行Agent] ⚠ 第{retry+1}次失败，准备重试"],
}
except Exception as e:
# 异常处理
return {
"execution_result": "error",
"retry_count": retry + 1,
"error": str(e),
"logs": [f"[执行Agent] ❌ 异常: {e}"],
}
def monitor_node(state):
"""监控节点：流程结束汇总
功能：
1. 判断最终执行状态
2. 生成结束日志
3. 标记流程完成
"""
status = "成功" if state.get("execution_result") == "success" else "失败"
return {
"current_phase": "done",
"logs": [f"[监控] 流程结束 | 状态:{status} | 重试:{state.get('retry_count', 0)}次"]
}
# ========== 路由函数（条件边）==========
def route_after_analysis(state):
"""分析后路由：决定下一步走向
路由逻辑：
- 如果当前阶段是 execute：直接到执行节点
- 否则：到审批节点
"""
return "executor" if state.get("current_phase") == "execute" else "approval"
def route_after_approval(state):
"""审批后路由：根据审批结果决定下一步
路由逻辑：
- 审批通过：到执行节点
- 审批驳回：到监控节点（结束流程）
"""
return "executor" if state.get("approval_status") == "approved" else "monitor"
def route_after_execution(state):
"""执行后路由：根据执行结果决定下一步
路由逻辑：
- 执行成功：到监控节点（结束流程）
- 达到最大重试次数：到监控节点（结束流程）
- 执行失败但未超限：回到执行节点（重试）
"""
if state.get("execution_result") == "success":
return "monitor"
if state.get("retry_count", 0) >= MAX_RETRY:
return "monitor"
return "executor"
# ========== 工作流构建 ==========
def build_workflow():
"""构建完整的 LangGraph 工作流
工作流结构：
START → analyst → (条件路由) → approval/executor → (条件路由) → monitor → END
特性：
- 使用 MemorySaver 实现状态持久化
- 在 approval 节点前暂停（interrupt_before）
- 支持条件边路由
"""
# 创建状态图
graph = StateGraph(WorkflowState)
# 注册节点
graph.add_node("analyst", analyst_node) # 分析节点
graph.add_node("approval", approval_node) # 审批节点
graph.add_node("executor", executor_node) # 执行节点
graph.add_node("monitor", monitor_node) # 监控节点
# 构建边（定义节点之间的连接）
graph.add_edge(START, "analyst") # 开始到分析
graph.add_conditional_edges("analyst", route_after_analysis) # 分析后的条件路由
graph.add_conditional_edges("approval", route_after_approval) # 审批后的条件路由
graph.add_conditional_edges("executor", route_after_execution) # 执行后的条件路由
graph.add_edge("monitor", END) # 监控到结束
# 编译工作流
# - checkpointer=memory：启用状态持久化
# - interrupt_before=["executor"]：在 executor 节点前暂停（人工审批）
memory = MemorySaver()
return graph.compile(checkpointer=memory, interrupt_before=["executor"])
# 全局工作流实例：整个应用共享一个编译好的工作流
workflow_app = build_workflow()"""
演示入口：一键运行完整工作流
特点：
- 无需配置 API Key，纯规则演示
- 展示完整的 Human-in-the-Loop 流程
- 包含详细的中英文日志输出
- 适合快速理解 LangGraph 工作流机制
运行方式：
python main.py
"""
from workflow import workflow_app, approve_action
from knowledge_engine import knowledge_engine
def demo():
"""运行完整的演示流程"""
print("=" * 55)
print(" 智控 · 企业级智能工作流平台 v1.0（演示版）")
print("=" * 55)
# 步骤1：初始化知识库（读取 docs/procurement_rules.txt）
print("\n📚 初始化知识库...")
knowledge_engine.build()
# 步骤2：配置工作流
# thread_id 用于标识唯一的工作流实例
config = {"configurable": {"thread_id": "WF-DEMO-001"}}
# 步骤3：设置初始状态
# 注意：TypedDict 只做类型约束，实际运行时使用普通字典
state = {
# 流程控制字段
"logs": [],
"current_phase": "analyze",
"risk_level": "",
"approval_status": "",
"retry_count": 0,
"error": None,
# 业务数据字段
"request_id": "WF-DEMO-001",
"request_type": "procurement",
"request_detail": "采购50台MacBook Pro用于研发团队",
"amount": 500000.0, # 50万，高金额，需要审批
"department": "技术部",
"knowledge_context": "",
"analysis_report": "",
"execution_result": "",
"final_report": "",
}
# 步骤4：提交工作流
print("\n🚀 提交工作流...")
result = workflow_app.invoke(state, config=config)
# 打印初始执行日志
print("\n📋 执行日志:")
for log in result.get("logs", []):
print(f" {log}")
# 步骤5：处理审批暂停（如果需要）
# 如果走到 approval 节点，工作流会暂停（interrupt_before）
if result.get("current_phase") == "approve":
print(f"\n⏸ 工作流已暂停（风险: {result.get('risk_level')}），等待审批...")
print("\n📄 分析报告:")
print(result.get('analysis_report', 'N/A'))
# 模拟人工审批
input("\n按回车模拟「审批通过」...")
# 使用 update_state 更新审批状态，保持原有 state 完整
workflow_app.update_state(config, approve_action(True, "同意采购"))
# 用 None 恢复执行，LangGraph 会从 checkpoint 继续
print("\n🔄 恢复工作流执行...")
result = workflow_app.invoke(None, config=config)
# 步骤6：输出最终结果
print("\n✅ 工作流完成")
print("\n📋 完整日志:")
for log in result.get("logs", []):
print(f" {log}")
print(f"\n📄 最终报告:")
print(result.get('final_report', 'N/A'))
if __name__ == "__main__":
demo()"""
FastAPI 接口层（可选）
功能：
- 提供 HTTP API 接口供外部系统调用
- 支持工作流提交和状态查询
- 可集成到现有系统中
运行方式：
uvicorn api:app --reload --port 8000
测试接口：
curl -X POST "http://localhost:8000/workflow/submit" \
-H "Content-Type: application/json" \
-d '{
"thread_id": "test-001",
"request_type": "procurement",
"request_detail": "采购测试设备",
"amount": 10000.0,
"department": "技术部"
}'
"""
from fastapi import FastAPI
from pydantic import BaseModel
from workflow import workflow_app
# 创建 FastAPI 应用实例
app = FastAPI(title="zhikong-platform")
class SubmitRequest(BaseModel):
"""工作流提交请求的数据模型"""
thread_id: str # 工作流唯一标识
request_type: str # 申请类型
request_detail: str # 申请详情
amount: float # 申请金额
department: str = "" # 申请部门（可选）
@app.post("/workflow/submit")
def submit(req: SubmitRequest):
"""提交工作流接口
Args:
req: 工作流提交请求
Returns:
包含执行结果的响应
"""
# 配置工作流
config = {"configurable": {"thread_id": req.thread_id}}
# 构建初始状态
state = {
# 流程控制字段
"logs": [],
"current_phase": "analyze",
"risk_level": "",
"approval_status": "",
"retry_count": 0,
"error": None,
# 业务数据字段
"request_id": req.thread_id,
"request_type": req.request_type,
"request_detail": req.request_detail,
"amount": req.amount,
"department": req.department,
"knowledge_context": "",
"analysis_report": "",
"execution_result": "",
"final_report": "",
}
# 执行工作流
result = workflow_app.invoke(state, config=config)
return {
"status": "success",
"data": result
}
@app.get("/workflow/status/{thread_id}")
def get_status(thread_id: str):
"""获取工作流状态接口
Args:
thread_id: 工作流唯一标识
Returns:
当前工作流状态
"""
config = {"configurable": {"thread_id": thread_id}}
try:
# 获取当前状态快照
snapshot = workflow_app.get_state(config)
return {
"status": "success",
"data": snapshot.values
}
except Exception as e:
return {
"status": "error",
"message": str(e)
}
@app.post("/workflow/approve/{thread_id}")
def approve(thread_id: str, approved: bool, comment: str = ""):
"""审批工作流接口
Args:
thread_id: 工作流唯一标识
approved: 是否批准
comment: 审批意见
Returns:
审批结果
"""
config = {"configurable": {"thread_id": thread_id}}
try:
# 更新审批状态
from workflow import approve_action
workflow_app.update_state(config, approve_action(approved, comment))
# 恢复执行
result = workflow_app.invoke(None, config=config)
return {
"status": "success",
"data": result
}
except Exception as e:
return {
"status": "error",
"message": str(e)
}# 核心依赖（演示版最小集合）
# LangGraph：工作流编排框架
langgraph>=0.0.50
# Pydantic：数据验证和序列化
pydantic>=2.0
# Streamlit 可视化界面（推荐，体验完整交互）
# 用于构建 Web 界面，实时展示工作流状态
streamlit>=1.28.0
# API 服务（可选，运行 api.py 才需要）
# FastAPI：构建 REST API
fastapi
# Uvicorn：ASGI 服务器
uvicorn[standard]
# 数据处理（Streamlit 导出功能需要）
# pandas：数据处理和 CSV 导出
pandas>=1.5.0采购管理制度（2024版）
# 第一条 采购审批权限
## 金额分级审批规则：
- 单笔金额 5,000 元以下：部门经理审批
- 单笔金额 5,000-50,000 元：总监审批
- 单笔金额 50,000 元以上：VP 审批 + 财务复核
# 第二条 紧急采购
## 紧急采购特殊流程：
- 紧急采购需标注「紧急」，可走快速通道
- 快速通道仍需至少一级审批，事后补齐材料
- 紧急采购不得超过季度预算的 10%
# 第三条 供应商管理
## 供应商选择规则：
- 优先选择合格供应商名录中的供应商
- 新供应商需提前完成资质审核（3个工作日）
- IT 设备类采购需经 IT 部门技术评审
# 第四条 IT 设备
## IT 设备特殊要求：
- IT 设备类采购需经 IT 部门技术评审
- 超出季度预算 10% 需 CFO 特批
- 优先选择已认证供应商
# 第五条 预算管理
## 预算超支处理：
- 超出部门预算 10% 以内：需部门经理说明原因
- 超出部门预算 10-20%：需 VP 特批
- 超出部门预算 20% 以上：需 CFO 特批
- 超出季度总预算：需董事会审批
💡 为什么需要 Streamlit 界面？
① 可视化工作流：实时展示 LangGraph 的节点执行过程、状态变化
② Human-in-the-Loop 交互：体验人工审批暂停/恢复机制
③ 状态持久化演示：查看 MemorySaver 的 checkpoint 快照
④ 项目完整性：前后端分离，体现工程化思维
"""
Streamlit 可视化界面 - 智控工作流平台（完整增强版）
运行方式：streamlit run app.py
核心功能：
1. 工作流提交与实时状态展示
2. 人工审批交互（Human-in-the-Loop）
3. 执行日志可视化
4. 工作流历史记录与统计
5. 参数动态调整
6. 知识库查询功能
7. CSV 数据导出
"""
import streamlit as st
import time
import uuid
from datetime import datetime
from workflow import workflow_app, approve_action
from knowledge_engine import knowledge_engine
# 页面配置
st.set_page_config(
page_title="智控 · 企业级智能工作流平台",
page_icon="🎯",
layout="wide",
initial_sidebar_state="expanded"
)
# 初始化知识库
if 'knowledge_initialized' not in st.session_state:
knowledge_engine.build()
st.session_state.knowledge_initialized = True
# 初始化工作流历史
if 'workflow_history' not in st.session_state:
st.session_state.workflow_history = []
# 主标题
st.markdown("""
🎯 智控 · 企业级智能工作流平台
基于 LangGraph 的多 Agent 协作系统 | Human-in-the-Loop | 状态持久化
""", unsafe_allow_html=True)
# 侧边栏 - 系统配置
with st.sidebar:
st.markdown("### ⚙️ 系统配置")
# 使用 tabs 组织侧边栏内容
tab1, tab2, tab3 = st.tabs(["📊 统计", "🔧 参数", "📚 知识库"])
with tab1:
# 工作流统计
if 'workflow_stats' not in st.session_state:
st.session_state.workflow_stats = {
'total': 0,
'success': 0,
'pending': 0,
'failed': 0
}
st.markdown("#### 今日统计")
col1, col2 = st.columns(2)
with col1:
st.metric("总请求", st.session_state.workflow_stats['total'])
st.metric("成功", st.session_state.workflow_stats['success'])
with col2:
st.metric("待审批", st.session_state.workflow_stats['pending'])
st.metric("失败", st.session_state.workflow_stats['failed'])
# 成功率
if st.session_state.workflow_stats['total'] > 0:
success_rate = (st.session_state.workflow_stats['success'] /
st.session_state.workflow_stats['total'] * 100)
st.progress(success_rate / 100)
st.caption(f"成功率: {success_rate:.1f}%")
with tab2:
st.markdown("#### 参数调整")
# 审批阈值调整
st.info("💡 修改参数后需要重新提交工作流才会生效")
approval_threshold = st.slider(
"审批金额阈值（元）",
min_value=1000,
max_value=100000,
value=5000,
step=1000,
help="超过此金额需要人工审批"
)
st.caption(f"当前设置：超过 ¥{approval_threshold:,} 需要审批")
# 显示不同金额的处理方式
st.markdown("**金额处理规则：**")
st.write(f"- < ¥{approval_threshold:,}: 自动通过")
st.write(f"- ≥ ¥{approval_threshold:,}: 需要审批")
if st.button("💾 应用参数", use_container_width=True):
st.session_state.approval_threshold = approval_threshold
st.success("✅ 参数已保存")
with tab3:
st.markdown("#### 知识库管理")
if st.button("🔄 重新加载", use_container_width=True):
knowledge_engine.build()
st.success("✅ 知识库已重新加载")
st.caption(f"文档数量: {len(knowledge_engine.chunks)}")
# 知识库查询功能
st.markdown("**知识库查询：**")
query = st.text_input("输入查询内容", placeholder="例如：采购审批流程")
if st.button("🔍 查询", use_container_width=True):
if query:
result = knowledge_engine.search(query, top_k=3)
st.text_area("查询结果", result, height=200)
else:
st.warning("请输入查询内容")
# 显示知识库内容预览
with st.expander("📄 查看文档列表"):
for i, chunk in enumerate(knowledge_engine.chunks):
st.markdown(f"**文档 {i+1}**: `{chunk.source}`")
st.caption(chunk.content[:100] + "...")
if i >= 2: # 只显示前3个
break
st.markdown("---")
st.markdown("#### 🔧 快速操作")
col_a, col_b = st.columns(2)
with col_a:
if st.button("🗑️ 清空", use_container_width=True):
st.session_state.workflow_history = []
st.session_state.workflow_stats = {'total': 0, 'success': 0, 'pending': 0, 'failed': 0}
st.rerun()
with col_b:
if st.button("📥 导出", use_container_width=True):
if st.session_state.workflow_history:
import pandas as pd
df = pd.DataFrame(st.session_state.workflow_history)
csv = df.to_csv(index=False, encoding='utf-8-sig')
st.download_button(
"⬇️ 下载",
csv,
f"workflow_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv",
"text/csv",
use_container_width=True
)
else:
st.warning("暂无数据")
# 主界面 - 三列布局
col1, col2, col3 = st.columns([2, 2, 1.5])
# 左侧 - 工作流提交
with col1:
st.markdown("### 📝 提交采购申请")
# 快速金额选择（在表单外面）
st.markdown("**💡 快速选择金额：**")
quick_cols = st.columns(4)
if quick_cols[0].button("¥3,000", help="低金额-自动通过", key="quick_3k"):
st.session_state.quick_amount = 3000.0
if quick_cols[1].button("¥8,000", help="中等金额-需审批", key="quick_8k"):
st.session_state.quick_amount = 8000.0
if quick_cols[2].button("¥60,000", help="高金额-需审批", key="quick_60k"):
st.session_state.quick_amount = 60000.0
if quick_cols[3].button("¥150,000", help="超高金额-需审批", key="quick_150k"):
st.session_state.quick_amount = 150000.0
with st.form("workflow_form"):
request_type = st.selectbox(
"申请类型",
["procurement", "expense", "leave"],
format_func=lambda x: {"procurement": "采购申请", "expense": "费用报销", "leave": "请假申请"}[x]
)
request_detail = st.text_area(
"申请详情",
placeholder="例如：采购50台MacBook Pro用于研发团队",
height=100
)
col_amt, col_dept = st.columns(2)
with col_amt:
# 使用 session_state 中的快速金额（如果有）
default_amount = st.session_state.get('quick_amount', 50000.0)
amount = st.number_input(
"金额（元）",
min_value=0.0,
max_value=10000000.0,
value=default_amount,
step=1000.0
)
with col_dept:
department = st.selectbox(
"申请部门",
["技术部", "市场部", "财务部", "人力资源部", "运营部"]
)
submitted = st.form_submit_button("🚀 提交工作流", use_container_width=True, type="primary")
if submitted:
if not request_detail:
st.error("❌ 请填写申请详情")
else:
# 生成唯一 thread_id
thread_id = f"WF-{datetime.now().strftime('%Y%m%d%H%M%S')}-{str(uuid.uuid4())[:8]}"
# 初始化状态
state = {
"logs": [],
"current_phase": "analyze",
"risk_level": "",
"approval_status": "",
"retry_count": 0,
"error": None,
"request_id": thread_id,
"request_type": request_type,
"request_detail": request_detail,
"amount": amount,
"department": department,
"knowledge_context": "",
"analysis_report": "",
"execution_result": "",
"final_report": "",
}
config = {"configurable": {"thread_id": thread_id}}
# 执行工作流（带进度条）
progress_bar = st.progress(0)
status_text = st.empty()
start_time = time.time()
try:
status_text.text("🔄 正在分析申请...")
progress_bar.progress(25)
time.sleep(0.3)
result = workflow_app.invoke(state, config=config)
progress_bar.progress(75)
status_text.text("✅ 工作流已提交")
end_time = time.time()
execution_time = end_time - start_time
# 保存到 session_state
if 'current_workflow' not in st.session_state:
st.session_state.current_workflow = {}
st.session_state.current_workflow = {
'thread_id': thread_id,
'config': config,
'result': result,
'status': 'pending' if result.get('current_phase') == 'approve' else 'running',
'start_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'execution_time': execution_time
}
# 添加到历史记录
st.session_state.workflow_history.append({
'thread_id': thread_id,
'type': request_type,
'detail': request_detail[:50] + '...' if len(request_detail) > 50 else request_detail,
'amount': f"¥{amount:,.0f}",
'department': department,
'status': result.get('current_phase'),
'risk_level': result.get('risk_level', 'N/A'),
'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'execution_time': f"{execution_time:.2f}s"
})
# 更新统计
st.session_state.workflow_stats['total'] += 1
if result.get('current_phase') == 'approve':
st.session_state.workflow_stats['pending'] += 1
progress_bar.progress(100)
time.sleep(0.5)
progress_bar.empty()
status_text.empty()
st.success(f"✅ 工作流已提交！ID: {thread_id}")
time.sleep(0.5)
st.rerun()
except Exception as e:
progress_bar.empty()
status_text.empty()
st.error(f"❌ 执行失败: {str(e)}")
st.session_state.workflow_stats['failed'] += 1
# 中间 - 工作流状态
with col2:
st.markdown("### 📊 工作流状态")
if 'current_workflow' in st.session_state and st.session_state.current_workflow:
wf = st.session_state.current_workflow
result = wf['result']
# 状态卡片
st.markdown(f"""
🆔 {wf['thread_id']}
当前阶段：{result.get('current_phase', 'N/A')}
风险等级：{result.get('risk_level', 'N/A')}
审批状态：{result.get('approval_status', 'N/A')}
重试次数：{result.get('retry_count', 0)}
申请金额：¥{result.get('amount', 0):,.0f}
申请部门：{result.get('department', 'N/A')}
""", unsafe_allow_html=True)
# 显示申请详情
with st.expander("📄 查看申请详情", expanded=False):
st.write(f"**申请类型：** {result.get('request_type', 'N/A')}")
st.write(f"**申请内容：** {result.get('request_detail', 'N/A')}")
if result.get('knowledge_context'):
st.write("**知识库检索结果：**")
st.text(result.get('knowledge_context', ''))
if result.get('analysis_report'):
st.write(f"**分析报告：** {result.get('analysis_report', '')}")
# 人工审批区域（修复显示逻辑）
# 只要当前阶段是 approve，就显示审批按钮
if result.get('current_phase') == 'approve':
st.markdown("#### 🔔 等待人工审批")
st.warning(f"⚠️ 风险等级：{result.get('risk_level')} | 金额：¥{result.get('amount')}")
# 审批意见输入
approval_comment = st.text_input(
"审批意见（可选）",
placeholder="请输入审批意见...",
key="approval_comment_input"
)
col_a, col_b = st.columns(2)
with col_a:
if st.button("✅ 批准", use_container_width=True, type="primary", key="approve_btn"):
with st.spinner("处理中..."):
try:
# 更新审批状态
workflow_app.update_state(wf['config'], approve_action(True, approval_comment or "同意采购"))
# 恢复执行
result = workflow_app.invoke(None, config=wf['config'])
st.session_state.current_workflow['result'] = result
st.session_state.current_workflow['status'] = 'completed'
st.session_state.workflow_stats['pending'] -= 1
st.session_state.workflow_stats['success'] += 1
st.success("✅ 已批准，工作流继续执行")
time.sleep(0.5)
st.rerun()
except Exception as e:
st.error(f"❌ 操作失败: {str(e)}")
with col_b:
if st.button("❌ 驳回", use_container_width=True, key="reject_btn"):
with st.spinner("处理中..."):
try:
workflow_app.update_state(wf['config'], approve_action(False, approval_comment or "金额超预算"))
result = workflow_app.invoke(None, config=wf['config'])
st.session_state.current_workflow['result'] = result
st.session_state.current_workflow['status'] = 'rejected'
st.session_state.workflow_stats['pending'] -= 1
st.session_state.workflow_stats['failed'] += 1
st.warning("❌ 已驳回")
time.sleep(1)
st.rerun()
except Exception as e:
st.error(f"❌ 操作失败: {str(e)}")
# 执行日志
st.markdown("#### 📋 执行日志")
logs = result.get('logs', [])
if logs:
for log in logs:
st.markdown(f'{log}', unsafe_allow_html=True)
else:
st.info("暂无日志")
# 最终报告
if result.get('final_report'):
st.markdown("#### 📄 执行报告")
st.code(result['final_report'], language="text")
# 刷新按钮
if st.button("🔄 刷新状态", use_container_width=True):
try:
snapshot = workflow_app.get_state(wf['config'])
st.session_state.current_workflow['result'] = snapshot.values
st.rerun()
except Exception as e:
st.error(f"刷新失败: {str(e)}")
else:
st.info("👈 请先提交工作流申请")
# 右侧 - LangGraph 可视化
with col3:
st.markdown("### 🔍 LangGraph 节点")
# 节点状态可视化
nodes = [
("START", "🎬", "开始"),
("analyst", "🔍", "分析"),
("approval", "✋", "审批"),
("executor", "⚙️", "执行"),
("monitor", "📊", "监控"),
("END", "🏁", "结束")
]
if 'current_workflow' in st.session_state and st.session_state.current_workflow:
current_phase = st.session_state.current_workflow['result'].get('current_phase', '')
for node_id, icon, name in nodes:
# 判断节点状态
if node_id == "START":
status = "✅"
color = "#34d399"
elif node_id == "analyst" and current_phase in ["approve", "execute", "done"]:
status = "✅"
color = "#34d399"
elif node_id == "approval" and current_phase in ["execute", "done"]:
status = "✅"
color = "#34d399"
elif node_id == "approval" and current_phase == "approve":
status = "⏸️"
color = "#fbbf24"
elif node_id == "executor" and current_phase == "done":
status = "✅"
color = "#34d399"
elif node_id == "monitor" and current_phase == "done":
status = "✅"
color = "#34d399"
elif node_id == "END" and current_phase == "done":
status = "✅"
color = "#34d399"
else:
status = "⚪"
color = "#64748b"
st.markdown(f"""
{icon}
{name}
{status}
""", unsafe_allow_html=True)
else:
for node_id, icon, name in nodes:
st.markdown(f"""
{icon}
{name}
⚪
""", unsafe_allow_html=True)
# 底部 - 核心亮点说明
st.markdown("---")
st.markdown("### 💡 核心技术亮点")
col1, col2, col3, col4 = st.columns(4)
with col1:
st.markdown("""
🔄
状态持久化
MemorySaver 实现工作流中断恢复
""", unsafe_allow_html=True)
with col2:
st.markdown("""
✋
Human-in-Loop
interrupt_before人工审批暂停
""", unsafe_allow_html=True)
with col3:
st.markdown("""
🔁
自动重试
条件边路由失败自动重试
""", unsafe_allow_html=True)
with col4:
st.markdown("""
📊
RAG 驱动
知识库检索规则驱动决策
""", unsafe_allow_html=True)pip install langgraph pydantic streamlitmkdir zhikong-platform
cd zhikong-platform
mkdir docs
# 复制上面所有代码文件到对应位置
# - config.py
# - workflow_models.py
# - knowledge_engine.py
# - workflow.py
# - main.py
# - app.py
# - requirements.txt
# - docs/procurement_rules.txtstreamlit run app.py
✅ 浏览器自动打开 http://localhost:8501，即可体验完整的可视化工作流
python main.py✅ 在终端中体验工作流执行过程
💡 体验建议：
① 先提交一个低金额申请（如 3000 元），观察自动执行流程
② 再提交一个高金额申请（如 60000 元），体验人工审批暂停/恢复
③ 查看右侧节点状态变化，理解 LangGraph 的执行流程
④ 观察执行日志，了解每个 Agent 的决策过程
第1步：理解工作流结构
workflow.py，理解 StateGraph 的构建过程第2步：体验状态持久化
interrupt_before=["executor"] 的作用update_state 如何更新审批状态第3步：修改和扩展
AUTO_APPROVE_THRESHOLD 调整审批阈值analyst_node 中添加更复杂的风险评估逻辑executor_node 中集成真实的业务系统 API第4步：面试准备
面试官最关心的不是"你用了什么框架"，而是"你解决了什么问题、做了什么取舍"。 下面把「智控」项目的 4 大核心难点 逐一拆解，每个难点都给出 问题 → 原因 → 方案 → 效果 的完整闭环。
┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ 采购申请提交 │──▶│ Analyst Agent│──▶│ Approval │──▶│ Executor │
│ (用户输入) │ │ RAG检索+分析 │ │ Human-in-Loop│ │ 执行+重试 │
└──────────────┘ └──────┬───────┘ └──────────────┘ └──────────────┘
│ │
┌──────▼───────┐ ┌──────▼───────┐
│ 知识库(Chroma)│ │ MemorySaver │
│ 企业规章制度 │ │ 状态持久化 │
└──────────────┘ └──────────────┘
┌────────────────────────────────────────────────────┐
│ LangGraph StateGraph 编排层 │
│ 条件边 + 循环重试 + 中断恢复 + 状态快照 │
└────────────────────────────────────────────────────┘
面试时画出这张图，展示你对多 Agent 协作架构的端到端理解，而不仅仅是"调了个 API"。
WorkflowState 包含所有字段（申请信息、风险评估、审批结果、执行日志等）；② 每个 Agent 只返回需要更新的字段，LangGraph 自动合并到全局 State；③ 使用 Annotated[List, operator.add] 让日志字段累加而非覆盖，保证完整的操作记录。time.sleep() 等待，会阻塞线程且服务重启后流程丢失。interrupt_before=["executor"]，工作流自动暂停（进行人工审批）；② 用 MemorySaver 将当前状态持久化到内存/数据库；③ 审批人通过 API 提交审批结果后，用 thread_id 恢复工作流并继续执行。即使服务重启，只要 thread_id 存在，流程就能无缝恢复。if 50000 <= amount <= 500000: risk="high"，金额阈值改变时必须改代码；而 RAG 方案只需更新知识库文档。
for i in range(3): try...except 无法记录每次重试的状态、无法可视化重试过程、无法灵活调整重试策略。以下提供 完整的简历项目经验写法，按照「项目名称 → 技术栈 → 项目描述 → 核心职责 → 项目成果」的标准格式撰写，可直接复制到简历中使用。
interrupt_before + MemorySaver/PostgresSaver 实现工作流暂停与恢复，支持跨天审批且状态不丢失，审批超时自动提醒与上报。Q: 为什么不用一个大 Agent？
A: 单一 Agent Prompt 过长导致指令遵循下降，拆分后每个 Agent 专注一件事，准确率更高，支持独立替换升级。
Q: MemorySaver 是内存级别的，生产环境怎么办？
A: 换成 SqliteSaver 或 PostgresSaver，只改一行代码，业务逻辑完全不变。LangGraph 的 Checkpointer 是接口化设计。
Q: 和传统工作流引擎（Activiti/Camunda）有什么区别？
A: 传统引擎处理固定规则流程，智控的优势是 AI 驱动的智能决策——风险评估由 LLM 完成，规则检索由 RAG 完成，给工作流加上了"大脑"。
Q: 性能瓶颈在哪？如何优化？
A: 瓶颈是 LLM 调用延迟。优化：1) 意图分类用更小的模型；2) FAISS 加 LRU 缓存；3) SSE 流式输出降低感知延迟；4) 异步并发。