集成方案将LightOnOCR-2-1B接入企业微信机器人自动识图1. 当企业微信机器人“学会”看图识字想象一下这个场景财务同事在群里发了一张供应商的发票照片问“这笔款付了吗”。通常你需要点开图片眯着眼睛辨认模糊的金额和抬头然后手动输入到系统里查询。或者销售发来一张客户手写的需求清单你得一个字一个字地敲进CRM。这些琐碎、重复的“人肉OCR”工作每天都在消耗团队的效率。有没有一种方法能让企业微信机器人自动看懂图片里的文字并直接给出结构化的信息答案是肯定的。通过将LightOnOCR-2-1B这个专精文字识别的AI模型与企业微信的机器人API“焊接”在一起你就能打造一个7x24小时在线的“识图小助手”。它不休息、不出错收到图片秒回文字还能把结果直接推送到审批流、数据库或知识库。这篇文章我将带你一步步实现这个自动化流程。你不需要是AI专家只需要会基本的命令行操作和一点Python脚本编写能力。我们将从最核心的“如何让机器人调用OCR”讲起覆盖消息接收、图片处理、结果返回的全链路并提供可直接部署的代码。2. 核心架构消息如何流转在动手写代码前我们先理清整个系统的数据流向。理解了这个后面的每一步都会变得清晰。整个流程可以概括为“接收-识别-回复”三个核心环节用户在企业微信发送图片 ↓ 企业微信服务器将消息推送到你的回调服务器 ↓ 你的服务器程序接收到图片URL ↓ 程序下载图片并转换为Base64编码 ↓ 调用部署好的LightOnOCR-2-1B API进行识别 ↓ 获取OCR识别出的纯文本结果 ↓ 将文本结果封装成企业微信消息格式 ↓ 调用企业微信API将识别结果发送回群聊或用户关键组件说明企业微信机器人/应用负责接收和发送消息的“前台”。你需要创建一个企业微信自建应用并配置其接收消息的API地址。你的回调服务器Callback Server这是系统的“大脑”。一个运行在你服务器上的Web服务如用Flask、FastAPI搭建用于接收企业微信推送的消息并协调后续所有处理逻辑。LightOnOCR-2-1B API服务这是系统的“眼睛”。即你已经部署好的OCR模型服务监听在http://你的服务器IP:8000。网络与安全确保你的回调服务器有公网IP或域名并且企业微信能访问到。同时需要配置Token和EncodingAESKey用于消息加解密。接下来我们分步实现这三个核心环节。3. 第一步准备你的“眼睛”LightOnOCR-2-1B服务假设你已经按照官方文档部署了LightOnOCR-2-1B。我们这里快速确认服务状态并进行一个简单的连通性测试这是后续所有工作的基础。3.1 确认OCR服务健康在你的LightOnOCR服务器上执行以下命令确保两个关键端口前端7860后端API 8000都在监听ss -tlnp | grep -E “:7860|:8000”你应该看到类似下面的输出表明服务运行正常LISTEN 0 128 0.0.0.0:8000 0.0.0.0:* LISTEN 0 128 0.0.0.0:7860 0.0.0.0:*3.2 进行一次API调用测试我们用一个简单的Python脚本来测试OCR API是否工作。将下面的代码保存为test_ocr_api.py并替换server_ip为你的实际IP替换image_path为一张测试图片的路径。import requests import base64 import json def test_ocr_api(server_ip, image_path): # 1. 将图片转换为Base64 with open(image_path, “rb”) as image_file: base64_image base64.b64encode(image_file.read()).decode(‘utf-8’) # 2. 构造API请求 url f“http://{server_ip}:8000/v1/chat/completions” headers {“Content-Type”: “application/json”} payload { “model”: “/root/ai-models/lightonai/LightOnOCR-2-1B”, “messages”: [{ “role”: “user”, “content”: [{ “type”: “image_url”, “image_url”: {“url”: f“data:image/jpeg;base64,{base64_image}”} }] }], “max_tokens”: 4096 } # 3. 发送请求 try: response requests.post(url, headersheaders, datajson.dumps(payload), timeout30) response.raise_for_status() # 检查HTTP错误 result response.json() # 4. 提取并打印识别文本 extracted_text result[“choices”][0][“message”][“content”] print(“OCR识别成功”) print(“识别结果”) print(“-” * 20) print(extracted_text) print(“-” * 20) return extracted_text except requests.exceptions.RequestException as e: print(f“API请求失败: {e}”) return None except (KeyError, json.JSONDecodeError) as e: print(f“解析响应结果失败: {e}”) print(f“原始响应: {response.text}”) return None if __name__ “__main__”: # 配置你的参数 SERVER_IP “192.168.1.100” # 替换为你的OCR服务器IP TEST_IMAGE_PATH “./test_receipt.jpg” # 替换为你的测试图片路径 test_ocr_api(SERVER_IP, TEST_IMAGE_PATH)运行这个脚本python test_ocr_api.py如果看到打印出的图片文字内容恭喜你“眼睛”准备好了。如果失败请检查IP地址、网络连通性以及OCR服务日志。4. 第二步打造“大脑”企业微信回调服务器这是最核心的一步。我们需要创建一个Web服务器它能够验证企业微信的消息接收图片消息调用OCR并回复消息。我们将使用Flask这个轻量级框架因为它简单易懂。确保你的服务器已安装Python和Flask。4.1 创建企业微信应用并获取凭证登录企业微信管理后台。进入“应用管理” - “自建应用”点击“创建应用”。填写应用名称如“AI识图助手”、选择可见范围并上传Logo。创建成功后记录下以下关键信息AgentId(应用ID)Secret(应用密钥)企业ID(CorpID在“我的企业”页面查看)进入应用详情页找到“接收消息”模块点击“设置API接收”。填写你的服务器URL例如https://your-domain.com/wechat。你需要有一个支持HTTPS的公网域名或IP。生成并记录Token和EncodingAESKey。点击保存。此时企业微信会向你填写的URL发送一个验证请求我们需要在下一步的代码中处理它。4.2 编写回调服务器核心代码将下面的代码保存为wechat_ocr_bot.py。你需要填写代码开头CONFIG部分的所有信息。#!/usr/bin/env python3 # -*- coding: utf-8 -*- import hashlib import json import time import requests import base64 from flask import Flask, request, jsonify from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.backends import default_backend import xml.etree.ElementTree as ET import io app Flask(__name__) # 配置区域请务必修改 CONFIG { “OCR_SERVER_IP”: “192.168.1.100”, # LightOnOCR服务器的内网IP “OCR_SERVER_PORT”: “8000”, “CORP_ID”: “wwxxxxxx”, # 你的企业ID “AGENT_ID”: “1000002”, # 你的应用AgentId “SECRET”: “your_app_secret_here”, # 你的应用Secret “TOKEN”: “your_token_here”, # 接收消息配置的Token “EncodingAESKey”: “your_encoding_aes_key_here”, # 接收消息配置的EncodingAESKey } # # 全局变量用于存储Access Token _access_token None _token_expire_time 0 def get_access_token(): “”“获取企业微信API调用凭证带缓存机制。”“” global _access_token, _token_expire_time now int(time.time()) # 如果token未过期直接返回缓存的token if _access_token and now _token_expire_time: return _access_token url f“https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid{CONFIG[‘CORP_ID’]}corpsecret{CONFIG[‘SECRET’]}” try: resp requests.get(url, timeout10).json() if resp[“errcode”] 0: _access_token resp[“access_token”] _token_expire_time now resp[“expires_in”] - 300 # 提前5分钟过期 print(f“获取新Access Token成功: {_access_token[:10]}...”) return _access_token else: print(f“获取Access Token失败: {resp}”) return None except Exception as e: print(f“请求Access Token异常: {e}”) return None def decrypt_message(encrypt_msg): “”“解密企业微信推送的加密消息。”“” aes_key base64.b64decode(CONFIG[‘EncodingAESKey’] “”) cipher Cipher(algorithms.AES(aes_key), modes.CBC(aes_key[:16]), backenddefault_backend()) decryptor cipher.decryptor() decrypted decryptor.update(base64.b64decode(encrypt_msg)) decryptor.finalize() # PKCS#7去除填充 pad decrypted[-1] content decrypted[:-pad] # 前16字节为随机字符串需要去掉 xml_content content[16:].decode(‘utf-8’) return xml_content def encrypt_message(reply_msg, nonce, timestamp): “”“加密回复给企业微信的消息本例中未使用但保留以备需主动回复加密消息时用。”“” # 此处简化实际使用时可参考企业微信官方加密示例 pass def parse_wechat_message(xml_str): “”“解析解密后的XML消息提取关键信息。”“” root ET.fromstring(xml_str) msg_type root.find(“MsgType”).text from_user root.find(“FromUserName”).text to_user root.find(“ToUserName”).text create_time root.find(“CreateTime”).text msg_id root.find(“MsgId”).text content None pic_url None media_id None if msg_type “text”: content root.find(“Content”).text elif msg_type “image”: pic_url root.find(“PicUrl”).text media_id root.find(“MediaId”).text # 可以继续扩展其他消息类型如file, voice等 return { “msg_type”: msg_type, “from_user”: from_user, “to_user”: to_user, “create_time”: create_time, “msg_id”: msg_id, “content”: content, “pic_url”: pic_url, “media_id”: media_id, } def call_ocr_api(image_url): “”“下载图片并调用LightOnOCR-2-1B API进行识别。”“” try: # 1. 下载图片 resp requests.get(image_url, timeout15) resp.raise_for_status() image_data resp.content # 2. 转换为Base64 base64_image base64.b64encode(image_data).decode(‘utf-8’) # 3. 构造OCR请求 ocr_url f“http://{CONFIG[‘OCR_SERVER_IP’]}:{CONFIG[‘OCR_SERVER_PORT’]}/v1/chat/completions” headers {“Content-Type”: “application/json”} payload { “model”: “/root/ai-models/lightonai/LightOnOCR-2-1B”, “messages”: [{ “role”: “user”, “content”: [{ “type”: “image_url”, “image_url”: {“url”: f“data:image/jpeg;base64,{base64_image}”} }] }], “max_tokens”: 4096 } # 4. 发送请求 ocr_resp requests.post(ocr_url, headersheaders, datajson.dumps(payload), timeout30) ocr_resp.raise_for_status() result ocr_resp.json() # 5. 提取识别文本 extracted_text result[“choices”][0][“message”][“content”] return extracted_text.strip() except Exception as e: print(f“调用OCR API失败: {e}”) return f“识别图片时出错: {str(e)}” def send_wechat_message(to_user, content): “”“通过企业微信API发送文本消息。”“” access_token get_access_token() if not access_token: return False, “获取Access Token失败” url f“https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token{access_token}” payload { “touser”: to_user, “msgtype”: “text”, “agentid”: int(CONFIG[‘AGENT_ID’]), “text”: {“content”: content}, “safe”: 0, } try: resp requests.post(url, jsonpayload, timeout10).json() if resp[“errcode”] 0: return True, “发送成功” else: return False, resp[“errmsg”] except Exception as e: return False, str(e) app.route(‘/wechat’, methods[‘GET’, ‘POST’]) def handle_wechat(): “”“处理企业微信服务器推送消息的主入口。”“” if request.method ‘GET’: # 验证URL有效性企业微信首次配置回调时调用 signature request.args.get(‘msg_signature’, ‘’) timestamp request.args.get(‘timestamp’, ‘’) nonce request.args.get(‘nonce’, ‘’) echostr request.args.get(‘echostr’, ‘’) # 此处应实现签名验证逻辑为简化示例我们直接返回echostr。 # 生产环境务必按企业微信文档实现签名校验 # 参考https://developer.work.weixin.qq.com/document/path/90930 print(f“收到验证请求: signature{signature}, echostr{echostr}”) return echostr elif request.method ‘POST’: # 处理消息推送 xml_data request.data root ET.fromstring(xml_data) encrypt root.find(“Encrypt”).text # 解密消息 xml_content decrypt_message(encrypt) msg_info parse_wechat_message(xml_content) print(f“收到消息: {msg_info}”) # 仅处理图片消息 if msg_info[“msg_type”] “image”: pic_url msg_info[“pic_url”] from_user msg_info[“from_user”] # 告诉用户正在处理 send_wechat_message(from_user, “收到图片正在识别文字请稍候...”) # 调用OCR识别 ocr_result call_ocr_api(pic_url) # 将识别结果发送回用户 reply_content f“图片文字识别结果\n\n{ocr_result}” success, msg send_wechat_message(from_user, reply_content) if not success: print(f“发送识别结果失败: {msg}”) send_wechat_message(from_user, “识别完成但发送结果失败请稍后重试或联系管理员。”) else: # 非图片消息回复提示 send_wechat_message(msg_info[“from_user”], “暂不支持此消息类型。请发送图片我将为您识别其中的文字。”) # 企业微信要求返回success return “success” if __name__ ‘__main__’: # 在公网服务器运行时host应为‘0.0.0.0’并确保端口如5000开放 # 强烈建议使用Nginx反向代理并配置HTTPS app.run(host‘0.0.0.0’, port5000, debugFalse)4.3 配置与运行填写配置用你从企业微信后台获取的信息替换CONFIG字典中的所有值。安装依赖在服务器上执行pip install flask requests cryptography。运行服务python wechat_ocr_bot.py。服务将在http://0.0.0.0:5000启动。配置公网访问这是关键一步。你的本地服务器需要让企业微信能访问到。最佳实践使用云服务器如阿里云、腾讯云ECS并配置安全组开放5000端口。临时测试可以使用内网穿透工具如ngrok、frp将本地的5000端口暴露到一个公网地址。假设你的公网地址是https://your-public-domain.com。完成企业微信配置在企业微信应用后台的“接收消息”设置中将“服务器URL”设置为https://your-public-domain.com/wechat填入你代码中设置的Token和EncodingAESKey点击保存。如果验证失败请检查网络连通性企业微信能否访问你的URL。Token和EncodingAESKey是否填写正确。你的/wechatGET 接口是否正确返回了echostr参数代码中已简化处理生产环境需完整实现签名验证。5. 第三步优化与扩展让机器人更聪明基础功能跑通后我们可以从稳定性、用户体验和功能扩展上进行优化。5.1 稳定性优化异步处理与队列上面的代码是同步处理如果OCR识别耗时较长会导致企业微信服务器重试。更健壮的方式是使用消息队列。# 示例使用线程池进行异步处理 from concurrent.futures import ThreadPoolExecutor executor ThreadPoolExecutor(max_workers3) app.route(‘/wechat’, methods[‘POST’]) def handle_wechat(): # ... [之前的解密和解析代码] ... if msg_info[“msg_type”] “image”: # 立即回复“正在处理” send_wechat_message(from_user, “收到图片正在识别文字请稍候...”) # 将耗时的OCR任务提交到线程池避免阻塞主请求 executor.submit(process_image_async, pic_url, from_user) # ... return “success” def process_image_async(pic_url, from_user): “”“在后台线程中处理图片识别和结果发送。”“” try: ocr_result call_ocr_api(pic_url) reply_content f“图片文字识别结果\n\n{ocr_result}” send_wechat_message(from_user, reply_content) except Exception as e: send_wechat_message(from_user, f“识别过程发生错误: {str(e)}”)5.2 功能扩展从识别到执行单纯的文字识别只是第一步。结合识别结果机器人可以做得更多关键词触发工作流识别出“发票”、“报销”等关键词自动将图片和识别文本转发到指定的审批应用或群聊。结构化信息提取针对固定格式的票据如出租车票、增值税发票在OCR结果后接一个简单的正则表达式或规则引擎提取金额、日期、编号并格式化成表格回复。多语言翻译检测到非中文文本时自动调用翻译API如百度翻译、腾讯云翻译进行翻译中英对照回复。与内部系统对接将识别出的订单号、客户名称自动查询ERP或CRM系统并将查询结果一并返回。5.3 安全与权限管理IP白名单在企业微信应用后台配置“可信IP”只允许你自己的服务器回调地址调用API防止伪造请求。频率限制在Flask应用中添加限流机制如使用Flask-Limiter防止恶意用户刷API。内容审核如果识别内容可能涉及敏感信息可以在回复前接入内容安全API进行过滤。6. 总结让自动化成为团队的新习惯通过本文的步骤你已经成功地将一个强大的多语言OCR模型无缝集成到了企业微信这个日常协作工具中。这个“识图机器人”的价值会随着使用而不断放大对员工节省了重复的、易出错的手动录入时间体验“发张图就有文字”的便捷。对团队信息流转更快基于图片的沟通和决策效率得到提升。对管理者为后续更复杂的自动化流程如自动报销、合同归档、客户信息录入打下了坚实的数据基础。技术的集成最终是为了消除工具间的隔阂让信息自然流动。LightOnOCR-2-1B提供了精准的“视力”企业微信提供了广泛的“触达”而你的几行代码则是连接两者的“神经”。现在就从让机器人看懂第一张发票开始吧。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。