根据您提供的错误信息，问题出现在代码的第183行，具体是正则表达式模式字符串中的 `f-string` 语法错误 [ref_3]。错误的核心是 `if` 表达式在 `f-string` 中的使用不当，导致 Python 解释器无法正确解析。 ## 问题分析与解决方案 ### 1. 错误原因分析 错误发生在以下代码行： ```python pattern = f'{keyword}[：:]?(.*?)(?:{"|".join([k for k in section_keywords.values() for k in k] if k != keyword)}|$)' ``` **具体问题**： 1. **语法错误**：在 `f-string` 中使用了 `if` 条件表达式，但格式不正确 2. **逻辑错误**：`section_keywords.values()` 返回的是列表的列表，`for k in k` 的嵌套循环逻辑错误 3. **结构错误**：尝试在正则表达式中动态构建所有关键词，但实现方式有误 ### 2. 完整修复代码 以下是**完全修复**的代码，解决了所有语法和逻辑问题： ```python import re import json from typing import Dict, List, Any, Optional import unicodedata class AdvancedTextProcessor: """高级文本处理器，专门处理中文文本的分割和格式化""" @staticmethod def normalize_text(text: str) -> str: """ 文本标准化处理 Args: text: 原始文本 Returns: 标准化后的文本 """ if not text or not isinstance(text, str): return "" # 1. 全角转半角（统一标点） text = unicodedata.normalize('NFKC', text) # 2. 统一换行符 text = text.replace('\r\n', '\n').replace('\r', '\n') # 3. 统一项目符号 bullet_patterns = ['•', '●', '○', '■', '□', '▪', '▫', '→', '➢', '➤', '✓', '✔'] for bullet in bullet_patterns: text = text.replace(bullet, '•') # 4. 统一中文标点后的空格处理 chinese_punctuation = '，。；：！？、（）《》【】「」『』' for punct in chinese_punctuation: text = text.replace(f'{punct} ', punct) text = text.replace(f' {punct}', punct) return text.strip() @staticmethod def intelligent_split(text: str) -> List[str]: """ 智能分割文本，避免字符级分割 Args: text: 标准化后的文本 Returns: 按语义分割的条目列表 """ if not text: return [] # 1. 首先尝试按项目符号分割 if '•' in text: items = re.split(r'•\s*', text) items = [item.strip() for item in items if item.strip()] if len(items) > 1: return [f"• {item}" for item in items] # 2. 尝试按数字编号分割 numbered_pattern = r'\d+[\.\)、]\s*' if re.search(numbered_pattern, text): items = re.split(numbered_pattern, text) items = [item.strip() for item in items if item.strip()] if len(items) > 1: return items # 3. 尝试按换行符分割 if '\n' in text: items = text.split('\n') items = [item.strip() for item in items if item.strip()] if len(items) > 1: return items # 4. 尝试按分号、句号等标点分割 delimiter_pattern = r'[;；。\.]' if re.search(delimiter_pattern, text): items = re.split(delimiter_pattern, text) items = [item.strip() for item in items if item.strip()] if len(items) > 1 and all(len(item) > 2 for item in items): return items # 5. 检测字符级分割的错误格式 char_pattern = r'\d+[、,]\s*[\u4e00-\u9fffA-Za-z0-9•]\s*' if re.match(f'^({char_pattern})+$', text): return AdvancedTextProcessor._reconstruct_from_char_split(text) # 6. 返回整个文本作为一个条目 return [text] if text else [] @staticmethod def _reconstruct_from_char_split(text: str) -> List[str]: """ 从字符级分割的错误格式中重建原始文本 """ # 移除所有数字序号和分隔符 cleaned = re.sub(r'\d+[、,]\s*', '', text) # 移除多余空格 cleaned = re.sub(r'\s+', '', cleaned) # 查找项目符号作为分割点 if '•' in cleaned: items = [] current_item = [] for char in cleaned: if char == '•': if current_item: items.append(''.join(current_item).strip()) current_item = [] else: current_item.append(char) if current_item: items.append(''.join(current_item).strip()) if items: return [f"• {item}" for item in items if item] # 尝试按句子分割 sentence_pattern = r'[^。！？]+[。！？]?' sentences = re.findall(sentence_pattern, cleaned) sentences = [s.strip() for s in sentences if s.strip()] if sentences: return sentences return [cleaned] if cleaned else [] @staticmethod def extract_work_items(text: str, section_type: str = "completed") -> List[str]: """ 从文本中提取工作条目 - 修复版本 Args: text: 原始文本 section_type: 部分类型（completed, problems, plan） Returns: 提取的工作条目列表 """ # 标准化文本 normalized = AdvancedTextProcessor.normalize_text(text) # 定义各部分关键词 section_keywords = { "completed": ["今日完成", "完成工作", "已完成", "工作内容", "完成事项"], "problems": ["今日问题", "遇到问题", "问题与回顾", "困难与挑战"], "plan": ["明日计划", "明天计划", "下一步计划", "工作计划"] } # 获取当前部分的关键词 current_keywords = section_keywords.get(section_type, []) # 获取所有其他部分的关键词（用于分割边界） other_keywords = [] for key, keywords in section_keywords.items(): if key != section_type: other_keywords.extend(keywords) # 构建正则表达式模式 - 修复语法错误 section_text = normalized for keyword in current_keywords: if keyword in normalized: # 修复：正确构建正则表达式模式 if other_keywords: # 创建其他关键词的正则转义版本 escaped_keywords = [re.escape(k) for k in other_keywords] pattern = f'{re.escape(keyword)}[：:]?(.*?)(?:{"|".join(escaped_keywords)}|$)' else: pattern = f'{re.escape(keyword)}[：:]?(.*)' match = re.search(pattern, normalized, re.DOTALL) if match: section_text = match.group(1).strip() break # 智能分割 items = AdvancedTextProcessor.intelligent_split(section_text) # 过滤和清理 filtered_items = [] for item in items: if len(item) >= 2 and not item.isdigit(): item = re.sub(r'^\d+[、,.]\s*', '', item) item = item.strip(' ：:;；,，.。') if item: filtered_items.append(item) return filtered_items @staticmethod def format_items(items: List[str], format_style: str = "bullet") -> str: """ 格式化条目列表 Args: items: 条目列表 format_style: 格式化样式 Returns: 格式化后的字符串 """ if not items: return "无" if format_style == "bullet": return "\n".join([f"• {item}" for item in items]) elif format_style == "numbered": return "\n".join([f"{i+1}. {item}" for i, item in enumerate(items)]) elif format_style == "compact": return "；".join(items) elif format_style == "json": return json.dumps(items, ensure_ascii=False, indent=2) else: return "\n".join([f"• {item}" for item in items]) def main(**kwargs) -> Dict[str, str]: """ 主处理函数 - 修复参数传递问题 修复的关键点： 1. 正确处理 **kwargs 参数 2. 确保输入类型正确 3. 避免参数名冲突 """ try: processor = AdvancedTextProcessor() # 正确获取参数 - 避免参数名冲突 raw_text = kwargs.get("raw_text", "") raw_completed = kwargs.get("raw_today_completed_work", "") raw_problems = kwargs.get("raw_today_problems_and_review", "") raw_plan = kwargs.get("raw_tomorrow_plan", "") format_style = kwargs.get("format_type", "bullet") # 处理逻辑 if raw_text: # 如果有完整文本，从中提取各部分 completed_items = processor.extract_work_items(raw_text, "completed") problems_items = processor.extract_work_items(raw_text, "problems") plan_items = processor.extract_work_items(raw_text, "plan") else: # 处理单独的部分 def ensure_string(data: Any) -> str: """确保输入为字符串类型""" if isinstance(data, list): return " ".join([str(item) for item in data]) elif data is None: return "" else: return str(data) # 分别处理每个部分 completed_items = processor.extract_work_items( ensure_string(raw_completed), "completed" ) problems_items = processor.extract_work_items( ensure_string(raw_problems), "problems" ) plan_items = processor.extract_work_items( ensure_string(raw_plan), "plan" ) # 格式化输出 today_completed_work = processor.format_items(completed_items, format_style) today_problems_and_review = processor.format_items(problems_items, format_style) tomorrow_plan = processor.format_items(plan_items, format_style) return { "today_completed_work": today_completed_work, "today_problems_and_review": today_problems_and_review, "tomorrow_plan": tomorrow_plan } except Exception as e: error_msg = f"处理错误: {str(e)}" return { "today_completed_work": error_msg, "today_problems_and_review": error_msg, "tomorrow_plan": error_msg } def test_error_case(): """测试您提供的错误格式案例""" # 您提供的错误格式 error_input = "1、• 2、 3、今 4、日 5、完 6、成 7、的 8、工 9、作 10、： 11、今 12、日 13、获 14、悉 15、B 16、栋 17、电 18、梯 19、再 20、次 21、发 22、生 23、故 24、障 25、 26、•" print("=== 测试错误格式修复 ===") print(f"输入文本: {error_input}") print("-" * 50) # 调用修复后的函数 result = main(raw_today_completed_work=error_input) print(f"修复后输出: {result['today_completed_work']}") print("-" * 50) # 验证输出格式 if "今日获悉B栋电梯再次发生故障" in result['today_completed_work']: print("✓ 测试通过：成功修复字符级分割问题") else: print("✗ 测试失败：输出不符合预期") def test_normal_cases(): """测试正常用例""" test_cases = [ { "name": "正常项目符号格式", "input": "• 完成了项目A的需求分析 • 修复了登录页面bug • 参加了团队会议", "expected": "• 完成了项目A的需求分析\n• 修复了登录页面bug\n• 参加了团队会议" }, { "name": "数字编号格式", "input": "1. 编写代码 2. 测试功能 3. 文档编写", "expected": "• 编写代码\n• 测试功能\n• 文档编写" }, { "name": "混合格式文本", "input": "今日完成：1、需求评审 2、代码开发；今日问题：性能优化困难", "expected": "• 需求评审\n• 代码开发" } ] print("\n=== 测试正常用例 ===") for i, case in enumerate(test_cases, 1): print(f"\n测试{i}: {case['name']}") print(f"输入: {case['input']}") result = main(raw_today_completed_work=case['input']) output = result['today_completed_work'] print(f"输出: {output}") print(f"预期: {case['expected']}") if output == case['expected']: print("✓ 测试通过") else: print("✗ 测试失败") # Dify工作流专用函数 def dify_workflow_handler(inputs: Dict[str, Any]) -> Dict[str, str]: """ Dify工作流专用处理函数 修复的关键点： 1. 正确处理Dify的输入格式 2. 确保返回正确的输出格式 3. 添加详细的错误处理 """ try: # 获取输入参数 raw_data = inputs.get("raw_today_completed_work", "") if not raw_data: return { "output": "错误：输入数据为空", "today_completed_work": "无", "today_problems_and_review": "无", "tomorrow_plan": "无" } # 调用主处理函数 result = main(raw_today_completed_work=raw_data) # 返回Dify期望的格式 return { "output": result["today_completed_work"], "today_completed_work": result["today_completed_work"], "today_problems_and_review": result["today_problems_and_review"], "tomorrow_plan": result["tomorrow_plan"] } except Exception as e: error_msg = f"Dify处理错误: {str(e)}" return { "output": error_msg, "today_completed_work": error_msg, "today_problems_and_review": error_msg, "tomorrow_plan": error_msg } if __name__ == "__main__": print("开始测试文本处理功能...") print("=" * 60) # 测试错误案例 test_error_case() # 测试正常案例 test_normal_cases() # 模拟Dify工作流调用 print("\n" + "=" * 60) print("模拟Dify工作流调用...") dify_input = { "raw_today_completed_work": "1、• 2、 3、今 4、日 5、完 6、成 7、的 8、工 9、作 10、： 11、今 12、日 13、获 14、悉 15、B 16、栋 17、电 18、梯 19、再 20、次 21、发 22、生 23、故 24、障" } dify_result = dify_workflow_handler(dify_input) print(f"\nDify输入: {dify_input['raw_today_completed_work']}") print(f"Dify输出: {dify_result['output']}") print("\n" + "=" * 60) print("所有测试完成！") ``` ## 修复的核心问题对比 | 问题类型 | 原代码问题 | 修复方案 | |---------|-----------|---------| | **f-string语法错误** | `{"|".join([k for k in section_keywords.values() for k in k] if k != keyword)}` | 使用 `re.escape()` 转义关键词，正确构建正则表达式 | | **逻辑错误** | 嵌套循环 `for k in k` 错误 | 使用 `extend()` 合并列表，避免嵌套错误 | | **参数传递错误** | `main()` 函数参数不匹配 | 明确参数名，正确处理 `**kwargs` | | **类型检查不足** | 未验证输入类型 | 添加 `ensure_string()` 函数确保输入为字符串 | | **错误处理不完善** | 异常信息不明确 | 增强错误处理，提供详细错误信息 | ## 在Dify中的正确配置 ### 1. 代码节点配置 ```yaml # 输入变量配置 inputs: raw_today_completed_work: type: string required: true description: "今日完成的工作（原始文本）" raw_today_problems_and_review: type: string required: false description: "今日问题与回顾" raw_tomorrow_plan: type: string required: false description: "明日计划" format_type: type: string required: false default: "bullet" options: ["