# 补充专题 5：Agent 工作流的代码骨架 > **本专题定位**：把"识别 → 决策 → 工具"三步法转成 Python 代码片段。 > **创建时间**：2026-08-03 > **状态**：v1.0 占位 · 计划在 v1.0.1 补充 --- ## 1. 总体架构 ``` [用户输入] ↓ [Step 1: 识别（NLP 引擎）] → 11 场景分类 ↓ [Step 2: 决策（规则引擎）] → 风险评级 + 法律依据 ↓ [Step 3: 工具（工具箱调用）] → 证据清单 + 话术 + 程序路径 ↓ [输出] ``` ## 2. Step 1：识别（11 场景分类器） ```python class SceneClassifier: """11 场景指纹识别器。""" SCENE_FINGERPRINTS = { 'scene_01': { 'name': '入职不签合同', 'keywords': ['没签合同', '签了空白合同', '入职没合同', '没给劳动合同'], 'priority': 'high', }, 'scene_02': { 'name': '试用期陷阱', 'keywords': ['试用期', '试用期解除', '试用期工资低', '试用期超长'], 'priority': 'medium', }, 'scene_03': { 'name': '996 / 奋斗者协议', 'keywords': ['996', '奋斗者', '自愿加班', '加班没加班费', '周末加班'], 'priority': 'high', }, 'scene_04': { 'name': '末位淘汰', 'keywords': ['末位', '末位淘汰', '考核排名', '绩效淘汰'], 'priority': 'high', }, 'scene_05': { 'name': '调岗 / 调薪', 'keywords': ['调岗', '调薪', '调部门', '调城市', '降职', '降薪'], 'priority': 'medium', }, 'scene_06': { 'name': '股权激励', 'keywords': ['期权', '限制性股票', '股权', '回购', '离职股权'], 'priority': 'medium', }, 'scene_07': { 'name': '竞业限制', 'keywords': ['竞业', '竞业限制', '竞业补偿', '同业禁止'], 'priority': 'medium', }, 'scene_08': { 'name': '工伤 / 职业病', 'keywords': ['工伤', '受伤', '职业病', '工作受伤', '上下班途中'], 'priority': 'high', }, 'scene_09': { 'name': '三期', 'keywords': ['怀孕', '产假', '哺乳期', '三期', '孕期'], 'priority': 'high', }, 'scene_10': { 'name': '软暴力', 'keywords': ['软暴力', '排挤', '冷板凳', '取消门禁', '暗示辞职'], 'priority': 'medium', }, 'scene_11': { 'name': '平台用工', 'keywords': ['外卖', '滴滴', '骑手', '网约车', '主播', '平台'], 'priority': 'medium', }, } def classify(self, user_input: str) -> list[dict]: """输入案情，输出可能匹配的 1-3 个场景。""" matches = [] for scene_id, info in self.SCENE_FINGERPRINTS.items(): score = sum(1 for kw in info['keywords'] if kw in user_input) if score > 0: matches.append({ 'scene_id': scene_id, 'name': info['name'], 'priority': info['priority'], 'score': score, }) # 按分数排序 matches.sort(key=lambda x: x['score'], reverse=True) return matches[:3] # 返回 top 3 ``` ## 3. Step 2：决策（风险评级） ```python class DecisionEngine: """风险评级 + 法律依据。""" SCENE_DECISION = { 'scene_01': { # 入职不签合同 'risk_level': 'red', 'risk_label': '🔴 红线', 'legal_basis': '《劳动合同法》§10、§82', 'compensation': '每月 2 倍工资（最高 11 个月）', 'next_step': '立即行动：申请仲裁', }, 'scene_03': { # 996 'risk_level': 'red', 'risk_label': '🔴 红线', 'legal_basis': '《劳动法》§36-44', 'compensation': '1.5/2/3 倍加班费', 'next_step': '立即行动：申请仲裁', }, 'scene_04': { # 末位淘汰 'risk_level': 'red', 'risk_label': '🔴 红线', 'legal_basis': '《劳动合同法》§40', 'compensation': '2N（违法解除）', 'next_step': '立即行动：申请仲裁', }, 'scene_08': { # 工伤 'risk_level': 'red', 'risk_label': '🔴 红线', 'legal_basis': '《工伤保险条例》', 'compensation': '工伤待遇（医疗费 + 伤残津贴）', 'next_step': '30 日内申请工伤认定', }, 'scene_09': { # 三期 'risk_level': 'red', 'risk_label': '🔴 红线', 'legal_basis': '《劳动合同法》§42', 'compensation': '2N 或恢复劳动关系', 'next_step': '立即行动：申请仲裁', }, # ... 其它场景 } def decide(self, scene_id: str, fact: dict) -> dict: """根据场景 + 事实，输出风险评级。""" decision = self.SCENE_DECISION.get(scene_id, {}) return { 'scene_id': scene_id, 'risk_level': decision.get('risk_level', 'unknown'), 'risk_label': decision.get('risk_label', '待定'), 'legal_basis': decision.get('legal_basis', '需进一步分析'), 'compensation': decision.get('compensation', '需进一步评估'), 'next_step': decision.get('next_step', '收集证据 + 评估方案'), } ``` ## 4. Step 3：工具（证据清单 + 话术 + 程序） ```python class ToolBox: """11 场景的工具箱。""" EVIDENCE_LISTS = { 'scene_01': [ # 入职不签合同 '工牌 / 工作证', '考勤记录（打卡、签到）', '工资条 / 银行流水', '工作记录（邮件、聊天记录、文件）', '同事证人证言', ], 'scene_03': [ # 996 '打卡记录', '加班审批流程', '工作邮件 / 聊天记录（显示非工作时间）', '同事证人证言', ], # ... 其它场景 } def get_evidence(self, scene_id: str) -> list[str]: """返回场景的证据清单。""" return self.EVIDENCE_LISTS.get(scene_id, []) def get_dialogue(self, scene_id: str) -> list[str]: """返回场景的应对话术。""" return [ '请公司提供 [X] 书面材料', '如果 [X]，我将通过 [Y] 途径维护权利', '请在 [N] 日内书面答复', ] def get_procedure(self, scene_id: str) -> list[str]: """返回程序路径（仲裁 / 投诉 / 起诉）。""" return [ '第一步：收集证据', '第二步：申请劳动仲裁（仲裁时效 1 年）', '第三步：如不服仲裁可起诉到法院', ] ``` ## 5. 主流程：3 步法串起来 ```python class LaborLawAgent: """劳动法顾问 Agent 主类。""" def __init__(self): self.classifier = SceneClassifier() self.decision_engine = DecisionEngine() self.toolbox = ToolBox() def handle(self, user_input: str) -> dict: """处理用户输入，输出完整响应。""" # Step 1：识别 scenes = self.classifier.classify(user_input) if not scenes: return { 'status': 'unmatched', 'message': '未能匹配到典型场景，请补充案情细节', } # Step 2：决策（取 top 1） top_scene = scenes[0] decision = self.decision_engine.decide( top_scene['scene_id'], fact={'input': user_input} ) # Step 3：工具 evidence = self.toolbox.get_evidence(top_scene['scene_id']) dialogue = self.toolbox.get_dialogue(top_scene['scene_id']) procedure = self.toolbox.get_procedure(top_scene['scene_id']) return { 'status': 'success', 'scene': top_scene, 'decision': decision, 'evidence': evidence, 'dialogue': dialogue, 'procedure': procedure, } ``` ## 6. 实际使用示例 ```python agent = LaborLawAgent() # 用户输入 user_input = "老板把我辞退了，理由是考核末位，没给补偿" result = agent.handle(user_input) # 输出 print(result) # { # 'status': 'success', # 'scene': {'scene_id': 'scene_04', 'name': '末位淘汰', 'priority': 'high', 'score': 1}, # 'decision': { # 'risk_level': 'red', # 'risk_label': '🔴 红线', # 'legal_basis': '《劳动合同法》§40', # 'compensation': '2N（违法解除）', # 'next_step': '立即行动：申请仲裁' # }, # 'evidence': [ # '考核制度', # '考核结果', # '培训记录', # '调岗记录', # '解除通知', # ], # 'dialogue': [...], # 'procedure': [...], # } ``` ## 7. 未来扩展 - **v1.1**：集成 LLM 做更精准的"识别"（用 LLM 替代关键词匹配） - **v1.2**：加入"算法管理"场景（平台用工 2.0） - **v1.3**：加入"跨境用工"场景（数字游民） - **v2.0**：与"四把尺子理论"对接，形成跨理论体系 --- **v1.0 占位 · 待补充**：完整的 Python 包 + 单元测试 + RAG 集成