### PSG (Python子生成器) 的使用方法及示例 #### Python 子生成器简介 在 Python 中，子生成器是指通过 `yield from` 语句调用的协程。这种机制简化了委托生成器和子生成器之间的协作过程。当一个生成器函数在其内部调用了另一个生成器时，前者被称为委托生成器，后者则称为子生成器。 #### 使用场景 子生成器通常用于处理嵌套的数据结构或者流水线式的操作流程。比如，在数据预处理阶段，可能需要先清洗一批原始记录再传递给下一个环节继续加工；又或者是实现生产者消费者模型中的多级缓冲区管理等功能[^1]。 #### 基本语法 ```python def sub_generator(): yield 'Sub-generator started' while True: received = yield if isinstance(received, str): break yield f'Sub-generator got: {received}' def main_generator(): print('Main generator starting...') result = yield from sub_generator() print(f'Main generator finished with message: "{result}"') ``` 在这个例子中，`main_generator()` 是委托生成器而 `sub_generator()` 则是被它调用的子生成器。执行过程中，一旦遇到 `yield from` 表达式就会暂停当前上下文并把控制权交给后面的子生成器直到其结束为止。最后返回的结果会被赋值给左侧变量 `result` 并恢复到原来的上下文中去完成后续逻辑。 #### 实际应用案例 假设有一个文件读取任务，其中涉及到逐行解析以及按条件筛选有效信息两步工作，则可以通过构建两个独立却相互关联的小型生成器来分别承担这两项职责： ```python import os.path def filter_lines(file_path, predicate): """Filter lines based on a given condition.""" try: with open(os.path.expanduser(file_path)) as file_obj: for line_num, raw_line in enumerate(file_obj, start=1): stripped_line = raw_line.strip() if not predicate(stripped_line): continue yield {'line_number': line_num, 'content': stripped_line} def process_files(files_list, *predicates): """Process multiple files using one or more filters sequentially applied to each.""" for path_item in files_list: filtered_items = [] pipeline = iter(predicates) current_filter = next(pipeline) temp_results = list(filter_lines(path_item, lambda x:True)) for item in temp_results: should_include = all([p(item['content']) for p in predicates]) if should_include: final_result = yield from enrich_data(item) filtered_items.append(final_result) yield dict(source=path_item, entries=filtered_items) def enrich_data(record_dict): record_dict.update({'processed_at': datetime.now()}) return record_dict if __name__ == '__main__': sample_file_paths = ['./data/sample_01.txt', './data/sample_02.txt'] non_empty_string_checker = lambda s : bool(s and not s.isspace()) contains_keyword_checker = partial(str.find, keyword='important') results_iterator = process_files(sample_file_paths, non_empty_string_checker, contains_keyword_checker) for processed_file_info in results_iterator: pprint(processed_file_info) ``` 上述代码展示了如何利用多个生成器组合成复杂的工作流，并且借助于 `yield from` 来串联起各个部分间的交互关系。这里不仅实现了基本的功能需求——即只保留含有特定关键字的有效行，还额外增加了时间戳标记作为元数据保存下来供后期审计或其他用途所用。