GLM-OCR数据处理实战如何避免代码耦合过度的设计模式你是不是也遇到过这种情况项目初期为了快速上线把GLM-OCR的调用、结果处理、数据输出这些功能一股脑儿全写在一个大函数里。刚开始跑得挺顺可一旦需求变了——比如要换个OCR模型、增加一种输出格式或者处理逻辑要调整——改起代码来简直像在拆炸弹牵一发而动全身到处都是硬编码的依赖。这就是典型的“代码耦合过度”。它让代码变得僵化、难以测试、更别提维护了。今天我们就从一个软件工程的角度聊聊怎么在GLM-OCR这类AI应用里设计出清晰、灵活、好维护的代码架构。我们不空谈理论就通过具体的反面案例和正面改造看看工厂模式、策略模式这些设计模式到底怎么用才能让代码“活”起来。1. 从“反面教材”说起一个耦合过度的OCR处理流程我们先来看一段在快速原型阶段很常见的代码。它的功能很直接调用GLM-OCR识别图片中的文字然后简单处理一下最后保存为文本文件。# 反面案例高度耦合的OCR处理函数 def process_image_with_glm_ocr(image_path, output_txt_path): 一个将所有逻辑耦合在一起的函数。 问题难以修改、难以测试、职责不清。 # 1. 加载图片硬编码依赖PIL from PIL import Image img Image.open(image_path) # 2. 调用GLM-OCR API硬编码API地址和参数 import requests api_url http://your-glm-ocr-server/v1/ocr payload {image: image_path} # 假设API接受路径 headers {Authorization: Bearer YOUR_TOKEN} response requests.post(api_url, jsonpayload, headersheaders) if response.status_code ! 200: raise Exception(fOCR API调用失败: {response.text}) # 3. 解析OCR原始结果硬编码结果结构 raw_result response.json() # 假设返回格式是 {text_blocks: [{text: ..., bbox: [...]}, ...]} text_blocks raw_result.get(text_blocks, []) # 4. 后处理简单拼接所有文本块处理逻辑固化在函数内 full_text for block in text_blocks: full_text block.get(text, ) \n # 移除多余空行简单的业务逻辑 full_text \n.join([line for line in full_text.split(\n) if line.strip()]) # 5. 输出到文件硬编码输出格式和方式 with open(output_txt_path, w, encodingutf-8) as f: f.write(full_text) print(f处理完成结果已保存至: {output_txt_path}) return full_text # 调用示例 # result process_image_with_glm_ocr(invoice.jpg, result.txt)这段代码能跑但问题一大堆。我们来拆解一下难以更换OCR服务如果你想试试另一个OCR模型比如百度的、腾讯的或者GLM-OCR的API升级了你得深入这个函数内部找到第2步和第3步小心翼翼地修改网络请求和结果解析逻辑很容易出错。难以调整处理逻辑如果业务方说“我们不需要按行拼接需要按识别出的表格结构来组织文本。”你又得去修改第4步。如果还有别的处理需求比如过滤特定关键词、计算文本置信度这个函数会像滚雪球一样越变越大。难以变更输出方式今天要输出TXT明天要输出JSON后天要直接存入数据库。每次变动你都得去改第5步甚至可能影响到前面的逻辑。几乎无法进行单元测试你怎么测试网络请求怎么模拟不同的OCR返回结果这个函数把网络I/O、文件I/O和业务逻辑死死绑在一起测试起来非常痛苦。核心问题就在于这个函数承担了太多不同的“职责”图片加载、服务调用、结果解析、业务处理、结果输出。这些职责像一团乱麻纠缠在一起形成了“耦合过度”。任何一个点的变化都可能需要你重写整个函数。2. 解耦之道用设计模式重构OCR处理流程我们的目标是把这团乱麻梳理成几条清晰的线。思路很简单分离关注点。让图片加载只关心图片服务调用只关心API处理逻辑只关心业务规则输出只关心格式。这里工厂模式和策略模式就能派上大用场。别被名字吓到它们其实就是帮你把“创建对象”和“选择算法”这两件事变得灵活、可配置的套路。2.1 第一步用工厂模式解耦OCR服务调用首先我们把“调用哪个OCR服务”这个选择从主流程里抽出来。定义一个所有OCR服务都必须遵守的“合同”接口然后用一个“工厂”来根据配置决定给你哪个服务实例。# 正面案例使用工厂模式解耦OCR服务 from abc import ABC, abstractmethod from typing import List, Dict, Any # 1. 定义OCR服务的抽象接口合同 class OcrService(ABC): OCR服务的抽象基类所有具体OCR服务都必须实现recognize方法。 abstractmethod def recognize(self, image_input) - Dict[str, Any]: 识别图片中的文字。 :param image_input: 图片路径或PIL.Image对象 :return: 包含识别结果的字典结构由具体实现定义 pass # 2. 实现具体的GLM-OCR服务 class GlmOcrService(OcrService): GLM-OCR服务的具体实现。 def __init__(self, api_base: str, api_key: str): self.api_base api_base.rstrip(/) self.api_key api_key def recognize(self, image_input) - Dict[str, Any]: import requests from PIL import Image import io import base64 # 准备图片数据这里简化处理实际可能需base64编码或传文件 if isinstance(image_input, str): # 如果是路径打开图片 with open(image_input, rb) as f: image_data base64.b64encode(f.read()).decode(utf-8) else: # 如果是PIL Image转换为base64 buffered io.BytesIO() image_input.save(buffered, formatPNG) image_data base64.b64encode(buffered.getvalue()).decode(utf-8) # 调用GLM-OCR API headers {Authorization: fBearer {self.api_key}, Content-Type: application/json} payload {image: image_data, mode: general} # 示例参数 response requests.post(f{self.api_base}/ocr, jsonpayload, headersheaders) response.raise_for_status() return response.json() # 返回原始API响应 # 3. 实现一个“模拟”的OCR服务用于测试 class MockOcrService(OcrService): 用于单元测试或开发的模拟OCR服务返回预设结果。 def __init__(self, fixed_result: Dict[str, Any]): self.fixed_result fixed_result def recognize(self, image_input) - Dict[str, Any]: print(f[Mock] 识别图片输入类型{type(image_input)}返回预设结果。) return self.fixed_result # 4. 简单的OCR服务工厂 class OcrServiceFactory: 负责创建OCR服务实例的工厂。 staticmethod def create_service(service_type: str, **kwargs) - OcrService: 根据类型创建OCR服务。 :param service_type: glm 或 mock :param kwargs: 创建服务所需的参数 :return: OcrService实例 if service_type glm: return GlmOcrService(api_basekwargs[api_base], api_keykwargs[api_key]) elif service_type mock: return MockOcrService(fixed_resultkwargs.get(fixed_result, {})) else: raise ValueError(f不支持的OCR服务类型: {service_type}) # 使用示例 config { service_type: glm, # 切换成 mock 就可以无缝使用模拟服务进行测试 api_base: http://your-glm-ocr-server/v1, api_key: YOUR_TOKEN_HERE } # 工厂根据配置创建服务主流程不关心具体是哪个服务 ocr_service OcrServiceFactory.create_service(**config) # 现在调用ocr_service.recognize(image)即可与具体实现解耦这样做的好处更换服务零成本明天要换百度OCR你只需要新增一个BaiduOcrService类实现OcrService接口然后在配置里把service_type改成baidu。主流程代码一行都不用改。测试变得极其简单在写单元测试时使用MockOcrService完全不用发起真实的网络请求测试又快又稳定。配置化服务的创建参数如API地址、密钥可以通过配置文件管理而不是硬编码在代码里。2.2 第二步用策略模式解耦结果处理逻辑OCR识别出的原始结果一堆带坐标的文本块往往需要根据业务需求进行二次加工。这个加工逻辑也应该是可插拔的。# 正面案例使用策略模式解耦结果处理逻辑 from abc import ABC, abstractmethod # 1. 定义结果处理器的抽象接口 class ResultProcessor(ABC): OCR结果处理器的抽象基类。 abstractmethod def process(self, raw_ocr_result: Dict[str, Any]) - Any: 处理原始OCR结果。 :param raw_ocr_result: recognize方法返回的原始字典 :return: 处理后的结果类型和结构由具体策略决定 pass # 2. 实现不同的处理策略 class SimpleTextConcatenator(ResultProcessor): 策略1简单地将所有识别文本按行拼接。 def process(self, raw_ocr_result: Dict[str, Any]) - str: text_blocks raw_ocr_result.get(text_blocks, []) lines [block.get(text, ).strip() for block in text_blocks if block.get(text, ).strip()] return \n.join(lines) class TableStructureExtractor(ResultProcessor): 策略2尝试根据文本框坐标提取表格结构简化示例。 def process(self, raw_ocr_result: Dict[str, Any]) - List[List[str]]: text_blocks raw_ocr_result.get(text_blocks, []) # 这里简化了真实的表格解析算法仅作演示 # 实际可能需要根据bbox的y坐标排序分行再根据x坐标分列 sorted_blocks sorted(text_blocks, keylambda b: (b[bbox][1], b[bbox][0])) # 按y, x排序 # 假设我们简单地将每行文本作为一个列表项 table_data [] current_row [] prev_y None for block in sorted_blocks: text block.get(text, ) bbox block.get(bbox, []) if prev_y is not None and abs(bbox[1] - prev_y) 10: # 简单的行判断阈值 if current_row: table_data.append(current_row) current_row [text] else: current_row.append(text) prev_y bbox[1] if current_row: table_data.append(current_row) return table_data class ConfidenceFilterProcessor(ResultProcessor): 策略3过滤掉置信度低于阈值的结果。 def __init__(self, confidence_threshold: float 0.8): self.threshold confidence_threshold def process(self, raw_ocr_result: Dict[str, Any]) - Dict[str, Any]: filtered_blocks [ block for block in raw_ocr_result.get(text_blocks, []) if block.get(confidence, 1.0) self.threshold ] # 返回结构相同但经过过滤的结果 return {**raw_ocr_result, text_blocks: filtered_blocks} # 3. 处理器上下文可选用于动态切换策略 class ProcessingPipeline: 一个简单的处理管道可以串联多个处理器。 def __init__(self): self.processors: List[ResultProcessor] [] def add_processor(self, processor: ResultProcessor): self.processors.append(processor) return self # 支持链式调用 def execute(self, raw_data: Dict[str, Any]) - Any: current_data raw_data for processor in self.processors: current_data processor.process(current_data) return current_data # 使用示例 pipeline ProcessingPipeline() # 可以灵活组合处理策略 pipeline.add_processor(ConfidenceFilterProcessor(confidence_threshold0.7)) pipeline.add_processor(SimpleTextConcatenator()) # 或者换成 TableStructureExtractor() # raw_result 是 ocr_service.recognize() 的结果 # processed_result pipeline.execute(raw_result)策略模式的威力业务逻辑灵活切换产品经理说“我们要表格数据”你就把SimpleTextConcatenator换成TableStructureExtractor。说“置信度太低的结果不要”你就加一个ConfidenceFilterProcessor。这些变化被隔离在各自的策略类里互不影响。易于扩展未来有新的处理需求比如文本翻译后处理、敏感词过滤只需要新增一个实现了ResultProcessor接口的类然后“插入”到处理管道中即可。单一职责每个处理器只做一件事代码更清晰也更容易测试。2.3 第三步解耦输出格式化与持久化最后处理好的数据怎么输出同样我们应该把“格式转换”和“保存到哪”分开。# 正面案例解耦输出格式化与持久化 from abc import ABC, abstractmethod import json # 1. 定义输出格式化器的接口 class OutputFormatter(ABC): abstractmethod def format(self, processed_data: Any) - str: 将处理后的数据格式化为字符串。 pass class TextFormatter(OutputFormatter): def format(self, processed_data: Any) - str: # 假设processed_data已经是字符串如SimpleTextConcatenator的结果 if isinstance(processed_data, str): return processed_data else: return str(processed_data) # 简单回退 class JsonFormatter(OutputFormatter): def format(self, processed_data: Any) - str: # 确保数据可被JSON序列化 import json return json.dumps(processed_data, ensure_asciiFalse, indent2) # 2. 定义输出器的接口负责把格式化后的字符串送到目的地 class OutputWriter(ABC): abstractmethod def write(self, content: str): 将内容写入目标。 pass class FileWriter(OutputWriter): def __init__(self, filepath: str): self.filepath filepath def write(self, content: str): with open(self.filepath, w, encodingutf-8) as f: f.write(content) print(f内容已写入文件: {self.filepath}) class ConsoleWriter(OutputWriter): def write(self, content: str): print(--- 输出内容 ---) print(content) print(---------------) # 3. 组合使用 def export_result(processed_data: Any, formatter: OutputFormatter, writer: OutputWriter): 通用的结果导出函数。 formatted_content formatter.format(processed_data) writer.write(formatted_content) # 使用示例可以任意组合格式化和输出方式 # export_result(text_data, TextFormatter(), FileWriter(output.txt)) # export_result(table_data, JsonFormatter(), ConsoleWriter()) # 未来增加DatabaseWriter、S3Writer等都非常容易至此我们最初那个“巨无霸”函数被拆分成了几个独立且可复用的模块OCR服务工厂、结果处理策略、输出格式化器和输出写入器。它们通过清晰的接口抽象类进行通信彼此之间没有硬依赖。3. 组装起来一个清晰、可配置的OCR处理管道现在我们把上面这些解耦的模块像乐高积木一样组装起来形成一个完整的、可配置的处理流程。# 正面案例组装后的清晰OCR处理管道 class ConfigurableOcrPipeline: 一个可配置、低耦合的OCR处理管道。 通过依赖注入DI的方式组合各个模块。 def __init__(self, ocr_service: OcrService, result_processor: ResultProcessor, output_formatter: OutputFormatter, output_writer: OutputWriter): self.ocr_service ocr_service self.result_processor result_processor self.output_formatter output_formatter self.output_writer output_writer def run(self, image_input): 执行完整的OCR处理流程。 print(开始OCR处理流程...) # 1. OCR识别 raw_result self.ocr_service.recognize(image_input) print(OCR识别完成。) # 2. 结果处理 processed_data self.result_processor.process(raw_result) print(结果处理完成。) # 3. 格式化并输出 formatted_content self.output_formatter.format(processed_data) self.output_writer.write(formatted_content) print(流程执行完毕。) return processed_data # 如何使用通过配置或依赖注入容器来组装管道 def main(): # 从配置或环境变量读取 config { ocr: {type: glm, api_base: ..., api_key: ...}, processor: {type: simple_text}, output: {format: text, destination: file, path: ./output.txt} } # 创建各个组件在实际项目中这部分通常由依赖注入框架完成 ocr_svc OcrServiceFactory.create_service(**config[ocr]) if config[processor][type] simple_text: processor SimpleTextConcatenator() elif config[processor][type] table: processor TableStructureExtractor() else: processor SimpleTextConcatenator() # 默认 if config[output][format] json: formatter JsonFormatter() else: formatter TextFormatter() if config[output][destination] console: writer ConsoleWriter() else: writer FileWriter(config[output][path]) # 组装并运行管道 pipeline ConfigurableOcrPipeline( ocr_serviceocr_svc, result_processorprocessor, output_formatterformatter, output_writerwriter ) pipeline.run(your_image.jpg) if __name__ __main__: # 在测试时可以轻松注入Mock对象 mock_pipeline ConfigurableOcrPipeline( ocr_serviceMockOcrService(fixed_result{text_blocks: [{text: 测试文本}]}), result_processorSimpleTextConcatenator(), output_formatterTextFormatter(), output_writerConsoleWriter() ) mock_pipeline.run(test.jpg) # 不依赖任何外部服务快速测试看看现在的代码是不是清爽多了每个模块各司其职通过接口松散地耦合在一起。需求变更时你只需要修改或替换其中一个“积木”而不会动摇整个建筑。4. 总结与建议回过头看我们通过引入工厂模式和策略模式成功地将一个高度耦合的“泥球”架构重构为一个清晰、灵活、可测试的“乐高”架构。核心思想就是“面向接口编程”和“依赖倒置”高层模块处理管道不依赖低层模块具体的OCR服务、处理器的实现细节而是依赖它们的抽象。这种设计带来的好处是实实在在的可维护性代码结构清晰修改一个功能不会“误伤”其他功能。可测试性每个模块都可以独立进行单元测试用Mock对象轻松模拟依赖。可扩展性新增一个OCR服务、一种处理策略或输出方式只需要添加新的类无需修改现有核心逻辑。可配置性系统的行为可以通过配置来改变甚至可以实现运行时动态切换。当然设计模式不是银弹过度设计也会增加复杂度。对于非常简单的、确定不会再变化的小脚本直接用一开始的“反面教材”写法也无可厚非。但一旦你的项目开始成长有了多个使用场景或者需要长期维护前期在架构上多花一点心思后期会省下大量的调试和重构时间。下次你在写GLM-OCR或者其他AI组件的集成代码时不妨先停下来想一想这段代码里哪些部分是可能变化的能不能把它们抽出来让它们更容易被替换养成这个习惯你的代码质量会提升一个档次。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
GLM-OCR数据处理实战：如何避免代码耦合过度的设计模式
发布时间：2026/7/6 9:41:54
GLM-OCR数据处理实战如何避免代码耦合过度的设计模式你是不是也遇到过这种情况项目初期为了快速上线把GLM-OCR的调用、结果处理、数据输出这些功能一股脑儿全写在一个大函数里。刚开始跑得挺顺可一旦需求变了——比如要换个OCR模型、增加一种输出格式或者处理逻辑要调整——改起代码来简直像在拆炸弹牵一发而动全身到处都是硬编码的依赖。这就是典型的“代码耦合过度”。它让代码变得僵化、难以测试、更别提维护了。今天我们就从一个软件工程的角度聊聊怎么在GLM-OCR这类AI应用里设计出清晰、灵活、好维护的代码架构。我们不空谈理论就通过具体的反面案例和正面改造看看工厂模式、策略模式这些设计模式到底怎么用才能让代码“活”起来。1. 从“反面教材”说起一个耦合过度的OCR处理流程我们先来看一段在快速原型阶段很常见的代码。它的功能很直接调用GLM-OCR识别图片中的文字然后简单处理一下最后保存为文本文件。# 反面案例高度耦合的OCR处理函数 def process_image_with_glm_ocr(image_path, output_txt_path): 一个将所有逻辑耦合在一起的函数。 问题难以修改、难以测试、职责不清。 # 1. 加载图片硬编码依赖PIL from PIL import Image img Image.open(image_path) # 2. 调用GLM-OCR API硬编码API地址和参数 import requests api_url http://your-glm-ocr-server/v1/ocr payload {image: image_path} # 假设API接受路径 headers {Authorization: Bearer YOUR_TOKEN} response requests.post(api_url, jsonpayload, headersheaders) if response.status_code ! 200: raise Exception(fOCR API调用失败: {response.text}) # 3. 解析OCR原始结果硬编码结果结构 raw_result response.json() # 假设返回格式是 {text_blocks: [{text: ..., bbox: [...]}, ...]} text_blocks raw_result.get(text_blocks, []) # 4. 后处理简单拼接所有文本块处理逻辑固化在函数内 full_text for block in text_blocks: full_text block.get(text, ) \n # 移除多余空行简单的业务逻辑 full_text \n.join([line for line in full_text.split(\n) if line.strip()]) # 5. 输出到文件硬编码输出格式和方式 with open(output_txt_path, w, encodingutf-8) as f: f.write(full_text) print(f处理完成结果已保存至: {output_txt_path}) return full_text # 调用示例 # result process_image_with_glm_ocr(invoice.jpg, result.txt)这段代码能跑但问题一大堆。我们来拆解一下难以更换OCR服务如果你想试试另一个OCR模型比如百度的、腾讯的或者GLM-OCR的API升级了你得深入这个函数内部找到第2步和第3步小心翼翼地修改网络请求和结果解析逻辑很容易出错。难以调整处理逻辑如果业务方说“我们不需要按行拼接需要按识别出的表格结构来组织文本。”你又得去修改第4步。如果还有别的处理需求比如过滤特定关键词、计算文本置信度这个函数会像滚雪球一样越变越大。难以变更输出方式今天要输出TXT明天要输出JSON后天要直接存入数据库。每次变动你都得去改第5步甚至可能影响到前面的逻辑。几乎无法进行单元测试你怎么测试网络请求怎么模拟不同的OCR返回结果这个函数把网络I/O、文件I/O和业务逻辑死死绑在一起测试起来非常痛苦。核心问题就在于这个函数承担了太多不同的“职责”图片加载、服务调用、结果解析、业务处理、结果输出。这些职责像一团乱麻纠缠在一起形成了“耦合过度”。任何一个点的变化都可能需要你重写整个函数。2. 解耦之道用设计模式重构OCR处理流程我们的目标是把这团乱麻梳理成几条清晰的线。思路很简单分离关注点。让图片加载只关心图片服务调用只关心API处理逻辑只关心业务规则输出只关心格式。这里工厂模式和策略模式就能派上大用场。别被名字吓到它们其实就是帮你把“创建对象”和“选择算法”这两件事变得灵活、可配置的套路。2.1 第一步用工厂模式解耦OCR服务调用首先我们把“调用哪个OCR服务”这个选择从主流程里抽出来。定义一个所有OCR服务都必须遵守的“合同”接口然后用一个“工厂”来根据配置决定给你哪个服务实例。# 正面案例使用工厂模式解耦OCR服务 from abc import ABC, abstractmethod from typing import List, Dict, Any # 1. 定义OCR服务的抽象接口合同 class OcrService(ABC): OCR服务的抽象基类所有具体OCR服务都必须实现recognize方法。 abstractmethod def recognize(self, image_input) - Dict[str, Any]: 识别图片中的文字。 :param image_input: 图片路径或PIL.Image对象 :return: 包含识别结果的字典结构由具体实现定义 pass # 2. 实现具体的GLM-OCR服务 class GlmOcrService(OcrService): GLM-OCR服务的具体实现。 def __init__(self, api_base: str, api_key: str): self.api_base api_base.rstrip(/) self.api_key api_key def recognize(self, image_input) - Dict[str, Any]: import requests from PIL import Image import io import base64 # 准备图片数据这里简化处理实际可能需base64编码或传文件 if isinstance(image_input, str): # 如果是路径打开图片 with open(image_input, rb) as f: image_data base64.b64encode(f.read()).decode(utf-8) else: # 如果是PIL Image转换为base64 buffered io.BytesIO() image_input.save(buffered, formatPNG) image_data base64.b64encode(buffered.getvalue()).decode(utf-8) # 调用GLM-OCR API headers {Authorization: fBearer {self.api_key}, Content-Type: application/json} payload {image: image_data, mode: general} # 示例参数 response requests.post(f{self.api_base}/ocr, jsonpayload, headersheaders) response.raise_for_status() return response.json() # 返回原始API响应 # 3. 实现一个“模拟”的OCR服务用于测试 class MockOcrService(OcrService): 用于单元测试或开发的模拟OCR服务返回预设结果。 def __init__(self, fixed_result: Dict[str, Any]): self.fixed_result fixed_result def recognize(self, image_input) - Dict[str, Any]: print(f[Mock] 识别图片输入类型{type(image_input)}返回预设结果。) return self.fixed_result # 4. 简单的OCR服务工厂 class OcrServiceFactory: 负责创建OCR服务实例的工厂。 staticmethod def create_service(service_type: str, **kwargs) - OcrService: 根据类型创建OCR服务。 :param service_type: glm 或 mock :param kwargs: 创建服务所需的参数 :return: OcrService实例 if service_type glm: return GlmOcrService(api_basekwargs[api_base], api_keykwargs[api_key]) elif service_type mock: return MockOcrService(fixed_resultkwargs.get(fixed_result, {})) else: raise ValueError(f不支持的OCR服务类型: {service_type}) # 使用示例 config { service_type: glm, # 切换成 mock 就可以无缝使用模拟服务进行测试 api_base: http://your-glm-ocr-server/v1, api_key: YOUR_TOKEN_HERE } # 工厂根据配置创建服务主流程不关心具体是哪个服务 ocr_service OcrServiceFactory.create_service(**config) # 现在调用ocr_service.recognize(image)即可与具体实现解耦这样做的好处更换服务零成本明天要换百度OCR你只需要新增一个BaiduOcrService类实现OcrService接口然后在配置里把service_type改成baidu。主流程代码一行都不用改。测试变得极其简单在写单元测试时使用MockOcrService完全不用发起真实的网络请求测试又快又稳定。配置化服务的创建参数如API地址、密钥可以通过配置文件管理而不是硬编码在代码里。2.2 第二步用策略模式解耦结果处理逻辑OCR识别出的原始结果一堆带坐标的文本块往往需要根据业务需求进行二次加工。这个加工逻辑也应该是可插拔的。# 正面案例使用策略模式解耦结果处理逻辑 from abc import ABC, abstractmethod # 1. 定义结果处理器的抽象接口 class ResultProcessor(ABC): OCR结果处理器的抽象基类。 abstractmethod def process(self, raw_ocr_result: Dict[str, Any]) - Any: 处理原始OCR结果。 :param raw_ocr_result: recognize方法返回的原始字典 :return: 处理后的结果类型和结构由具体策略决定 pass # 2. 实现不同的处理策略 class SimpleTextConcatenator(ResultProcessor): 策略1简单地将所有识别文本按行拼接。 def process(self, raw_ocr_result: Dict[str, Any]) - str: text_blocks raw_ocr_result.get(text_blocks, []) lines [block.get(text, ).strip() for block in text_blocks if block.get(text, ).strip()] return \n.join(lines) class TableStructureExtractor(ResultProcessor): 策略2尝试根据文本框坐标提取表格结构简化示例。 def process(self, raw_ocr_result: Dict[str, Any]) - List[List[str]]: text_blocks raw_ocr_result.get(text_blocks, []) # 这里简化了真实的表格解析算法仅作演示 # 实际可能需要根据bbox的y坐标排序分行再根据x坐标分列 sorted_blocks sorted(text_blocks, keylambda b: (b[bbox][1], b[bbox][0])) # 按y, x排序 # 假设我们简单地将每行文本作为一个列表项 table_data [] current_row [] prev_y None for block in sorted_blocks: text block.get(text, ) bbox block.get(bbox, []) if prev_y is not None and abs(bbox[1] - prev_y) 10: # 简单的行判断阈值 if current_row: table_data.append(current_row) current_row [text] else: current_row.append(text) prev_y bbox[1] if current_row: table_data.append(current_row) return table_data class ConfidenceFilterProcessor(ResultProcessor): 策略3过滤掉置信度低于阈值的结果。 def __init__(self, confidence_threshold: float 0.8): self.threshold confidence_threshold def process(self, raw_ocr_result: Dict[str, Any]) - Dict[str, Any]: filtered_blocks [ block for block in raw_ocr_result.get(text_blocks, []) if block.get(confidence, 1.0) self.threshold ] # 返回结构相同但经过过滤的结果 return {**raw_ocr_result, text_blocks: filtered_blocks} # 3. 处理器上下文可选用于动态切换策略 class ProcessingPipeline: 一个简单的处理管道可以串联多个处理器。 def __init__(self): self.processors: List[ResultProcessor] [] def add_processor(self, processor: ResultProcessor): self.processors.append(processor) return self # 支持链式调用 def execute(self, raw_data: Dict[str, Any]) - Any: current_data raw_data for processor in self.processors: current_data processor.process(current_data) return current_data # 使用示例 pipeline ProcessingPipeline() # 可以灵活组合处理策略 pipeline.add_processor(ConfidenceFilterProcessor(confidence_threshold0.7)) pipeline.add_processor(SimpleTextConcatenator()) # 或者换成 TableStructureExtractor() # raw_result 是 ocr_service.recognize() 的结果 # processed_result pipeline.execute(raw_result)策略模式的威力业务逻辑灵活切换产品经理说“我们要表格数据”你就把SimpleTextConcatenator换成TableStructureExtractor。说“置信度太低的结果不要”你就加一个ConfidenceFilterProcessor。这些变化被隔离在各自的策略类里互不影响。易于扩展未来有新的处理需求比如文本翻译后处理、敏感词过滤只需要新增一个实现了ResultProcessor接口的类然后“插入”到处理管道中即可。单一职责每个处理器只做一件事代码更清晰也更容易测试。2.3 第三步解耦输出格式化与持久化最后处理好的数据怎么输出同样我们应该把“格式转换”和“保存到哪”分开。# 正面案例解耦输出格式化与持久化 from abc import ABC, abstractmethod import json # 1. 定义输出格式化器的接口 class OutputFormatter(ABC): abstractmethod def format(self, processed_data: Any) - str: 将处理后的数据格式化为字符串。 pass class TextFormatter(OutputFormatter): def format(self, processed_data: Any) - str: # 假设processed_data已经是字符串如SimpleTextConcatenator的结果 if isinstance(processed_data, str): return processed_data else: return str(processed_data) # 简单回退 class JsonFormatter(OutputFormatter): def format(self, processed_data: Any) - str: # 确保数据可被JSON序列化 import json return json.dumps(processed_data, ensure_asciiFalse, indent2) # 2. 定义输出器的接口负责把格式化后的字符串送到目的地 class OutputWriter(ABC): abstractmethod def write(self, content: str): 将内容写入目标。 pass class FileWriter(OutputWriter): def __init__(self, filepath: str): self.filepath filepath def write(self, content: str): with open(self.filepath, w, encodingutf-8) as f: f.write(content) print(f内容已写入文件: {self.filepath}) class ConsoleWriter(OutputWriter): def write(self, content: str): print(--- 输出内容 ---) print(content) print(---------------) # 3. 组合使用 def export_result(processed_data: Any, formatter: OutputFormatter, writer: OutputWriter): 通用的结果导出函数。 formatted_content formatter.format(processed_data) writer.write(formatted_content) # 使用示例可以任意组合格式化和输出方式 # export_result(text_data, TextFormatter(), FileWriter(output.txt)) # export_result(table_data, JsonFormatter(), ConsoleWriter()) # 未来增加DatabaseWriter、S3Writer等都非常容易至此我们最初那个“巨无霸”函数被拆分成了几个独立且可复用的模块OCR服务工厂、结果处理策略、输出格式化器和输出写入器。它们通过清晰的接口抽象类进行通信彼此之间没有硬依赖。3. 组装起来一个清晰、可配置的OCR处理管道现在我们把上面这些解耦的模块像乐高积木一样组装起来形成一个完整的、可配置的处理流程。# 正面案例组装后的清晰OCR处理管道 class ConfigurableOcrPipeline: 一个可配置、低耦合的OCR处理管道。 通过依赖注入DI的方式组合各个模块。 def __init__(self, ocr_service: OcrService, result_processor: ResultProcessor, output_formatter: OutputFormatter, output_writer: OutputWriter): self.ocr_service ocr_service self.result_processor result_processor self.output_formatter output_formatter self.output_writer output_writer def run(self, image_input): 执行完整的OCR处理流程。 print(开始OCR处理流程...) # 1. OCR识别 raw_result self.ocr_service.recognize(image_input) print(OCR识别完成。) # 2. 结果处理 processed_data self.result_processor.process(raw_result) print(结果处理完成。) # 3. 格式化并输出 formatted_content self.output_formatter.format(processed_data) self.output_writer.write(formatted_content) print(流程执行完毕。) return processed_data # 如何使用通过配置或依赖注入容器来组装管道 def main(): # 从配置或环境变量读取 config { ocr: {type: glm, api_base: ..., api_key: ...}, processor: {type: simple_text}, output: {format: text, destination: file, path: ./output.txt} } # 创建各个组件在实际项目中这部分通常由依赖注入框架完成 ocr_svc OcrServiceFactory.create_service(**config[ocr]) if config[processor][type] simple_text: processor SimpleTextConcatenator() elif config[processor][type] table: processor TableStructureExtractor() else: processor SimpleTextConcatenator() # 默认 if config[output][format] json: formatter JsonFormatter() else: formatter TextFormatter() if config[output][destination] console: writer ConsoleWriter() else: writer FileWriter(config[output][path]) # 组装并运行管道 pipeline ConfigurableOcrPipeline( ocr_serviceocr_svc, result_processorprocessor, output_formatterformatter, output_writerwriter ) pipeline.run(your_image.jpg) if __name__ __main__: # 在测试时可以轻松注入Mock对象 mock_pipeline ConfigurableOcrPipeline( ocr_serviceMockOcrService(fixed_result{text_blocks: [{text: 测试文本}]}), result_processorSimpleTextConcatenator(), output_formatterTextFormatter(), output_writerConsoleWriter() ) mock_pipeline.run(test.jpg) # 不依赖任何外部服务快速测试看看现在的代码是不是清爽多了每个模块各司其职通过接口松散地耦合在一起。需求变更时你只需要修改或替换其中一个“积木”而不会动摇整个建筑。4. 总结与建议回过头看我们通过引入工厂模式和策略模式成功地将一个高度耦合的“泥球”架构重构为一个清晰、灵活、可测试的“乐高”架构。核心思想就是“面向接口编程”和“依赖倒置”高层模块处理管道不依赖低层模块具体的OCR服务、处理器的实现细节而是依赖它们的抽象。这种设计带来的好处是实实在在的可维护性代码结构清晰修改一个功能不会“误伤”其他功能。可测试性每个模块都可以独立进行单元测试用Mock对象轻松模拟依赖。可扩展性新增一个OCR服务、一种处理策略或输出方式只需要添加新的类无需修改现有核心逻辑。可配置性系统的行为可以通过配置来改变甚至可以实现运行时动态切换。当然设计模式不是银弹过度设计也会增加复杂度。对于非常简单的、确定不会再变化的小脚本直接用一开始的“反面教材”写法也无可厚非。但一旦你的项目开始成长有了多个使用场景或者需要长期维护前期在架构上多花一点心思后期会省下大量的调试和重构时间。下次你在写GLM-OCR或者其他AI组件的集成代码时不妨先停下来想一想这段代码里哪些部分是可能变化的能不能把它们抽出来让它们更容易被替换养成这个习惯你的代码质量会提升一个档次。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
