企业AI应用安全实践：构建钉钉与OpenAI间的敏感信息过滤代理

1. 项目概述：为什么企业需要这道“AI防火墙”？

最近在帮几个客户做企业内部AI应用落地的方案，发现一个挺普遍但容易被忽视的问题：当员工通过钉钉这类办公平台，便捷地调用OpenAI这类大模型API来处理工作内容时，公司内部那些不该外传的信息，会不会在不知不觉中“溜”出去？比如，一份正在讨论中的产品定价策略、一个未公开的客户名单，甚至是一段涉及商业机密的代码片段，如果被员工无意中粘贴给了AI，就可能造成信息泄露。这可不是危言耸听，而是实实在在的风险。

所以，“配置Dingtalk-OpenAI敏感词过滤”这个事，本质上是在企业的AI应用入口处，加装一道智能的“安检门”和“过滤器”。它的核心目标不是限制创新或阻碍效率，而是在享受AI带来的生产力红利的同时，构筑一道基本的信息安全防线。这就像给公司的数据上了把“智能锁”，只有经过检查、确认安全的内容，才能被发送到外部的AI服务进行处理，处理结果回来时，同样要经过过滤，防止AI生成的内容中包含不合规信息。

这个配置适合所有正在或计划将OpenAI API（或类似大模型服务）集成到钉钉、飞书、企业微信等办公平台的企业IT管理员、安全运维人员以及应用开发者。无论你是想通过钉钉机器人实现智能问答、自动生成会议纪要，还是构建更复杂的审批流与数据分析应用，只要涉及外部AI服务，这套过滤机制都应该是你方案设计中的标配环节。

2. 整体架构与核心组件解析

要实现钉钉与OpenAI之间的敏感词过滤，并不是在钉钉或OpenAI的官方设置里直接打个勾就能完成的。它需要我们构建一个中间代理层。这个代理层扮演着“守门人”和“翻译官”的双重角色，是整个方案的核心。

2.1 核心架构：为什么必须是“代理模式”？

直接让钉钉应用调用OpenAI API是行不通的，因为我们无法在客户端（钉钉）或服务端（OpenAI）直接插入自定义的过滤逻辑。因此，标准的架构是：钉钉应用 -> 自建代理服务器 -> OpenAI API。

1. 请求拦截与检查：当员工在钉钉中向AI机器人发送消息时，消息首先被发送到你部署的代理服务器，而不是直接到达OpenAI。代理服务器对消息内容进行敏感词扫描和过滤。
2. 安全转发与响应：通过检查的“纯净”消息，才会被代理服务器按照OpenAI API的格式要求，转发给OpenAI。OpenAI返回的文本结果，同样先回到代理服务器，进行二次内容安全审核（防止AI生成敏感信息），审核通过后再返回给钉钉机器人呈现给用户。
3. 日志与审计：所有经过代理的请求和响应，都可以被完整记录，包括用户ID、时间、原始内容、过滤后内容、触发的规则等，为安全审计提供数据支撑。

这种模式的优点在于自主可控。过滤规则、审核逻辑、日志存储完全掌握在企业自己手中，可以根据自身业务特点（如金融、医疗、法律等行业有特殊合规要求）进行深度定制。

2.2 关键组件选型与考量

搭建这个代理服务，主要涉及以下几个部分的技术选型：

1. 后端服务框架

• 推荐选择：Python的FastAPI或 Node.js的Express/Koa。对于此类中间件代理，FastAPI是当前非常热门的选择，因为它异步性能好、编写简洁，并且自动生成API文档，便于调试。如果团队更熟悉JavaScript/TypeScript技术栈，Express或Koa也是成熟稳定的选项。
• 为什么不选Django/Spring Boot？这类全功能框架略显“重”了。我们的代理服务核心功能是路由、过滤和转发，属于轻量级、高并发的网络中间件，轻量级框架更合适。

2. 敏感词过滤引擎

• 核心需求：高性能、低延迟、支持动态更新词库。过滤操作发生在每次API调用的关键路径上，必须快速。
• 推荐算法：前缀树（Trie树）或双数组Trie树（DAT）。这是敏感词过滤领域的标准算法，能在O(n)时间复杂度内完成单次扫描，并支持模糊匹配（如忽略大小写、全半角）。
• 开源库推荐：
  • Python:flashtext（关键词提取库，适用于精确匹配，速度极快）或自建Trie树。
  • Node.js:node-ahocorasick（Aho-Corasick自动机算法，适合多模式匹配）。
  • 通用方案：也可以使用DFA（确定有限状态自动机）算法自行实现，灵活性最高。

3. 词库管理与存储

• 静态基础词库：包含通用的违法违规、暴恐、色情低俗等词汇。可以从公开的安全词库初始化，但务必根据自身业务进行清洗和补充。
• 动态业务词库：这是核心。需要存储你公司的特有敏感信息，例如：
  • 项目代号：“泰山计划”、“北极星”
  • 核心数据字段名：“customer_salary”、“product_roadmap_2025”
  • 内部系统地址：“http://internal-admin.corp.com”
  • 高管姓名/特定客户名（在特定上下文中需脱敏）
• 存储方案：建议使用Redis缓存热词库，实现毫秒级读取和动态更新（如通过管理后台添加新词）。MySQL/PostgreSQL 作为词库的持久化存储，记录词条、分类、创建人、时间等元数据。

4. 钉钉交互与OpenAI客户端

• 钉钉：需要创建一个“自定义机器人”或“企业内部应用”，获取其Webhook地址或AppKey/AppSecret，用于接收钉钉用户的消息。
• OpenAI：需要使用官方openaiSDK（Python）或openainpm包（Node.js），并配置你的API Key。代理服务需要模拟一个兼容OpenAI API格式的端点。

实操心得一：词库的“灰度发布”直接全量更新敏感词库是危险的，一个误判就会导致业务中断。我们的做法是，任何新添加的敏感词，先进入“观察词库”，仅打日志告警，不实际拦截。观察一段时间（如24小时），确认日志中的触发都是合理的敏感信息后，再将其正式加入“拦截词库”。这能有效避免误杀正常业务沟通。

3. 分步实现：从零搭建过滤代理服务

下面我将以 Python + FastAPI 技术栈为例，详细拆解搭建过程。假设你已经具备基本的Python开发环境和服务器（云主机或容器）。

3.1 环境准备与依赖安装

首先，创建一个新的项目目录并初始化虚拟环境。

mkdir dingtalk-openai-filter-proxy && cd dingtalk-openai-filter-proxy python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate

安装核心依赖库：

pip install fastapi uvicorn openai httpx redis pymysql # fastapi/uvicorn: Web框架和服务器 # openai: 官方OpenAI SDK # httpx: 用于异步HTTP请求（替代requests） # redis/pymysql: 用于连接词库存储

3.2 构建敏感词过滤引擎

我们实现一个基于Trie树的过滤服务。创建一个filter_engine.py文件。

import redis import json from typing import List, Set import re class SensitiveWordFilter: def __init__(self, redis_client=None): """ 初始化过滤器，支持从Redis加载词库。 """ self.trie = {} # 前缀树字典 self.delimiters = set(' ,.!?;:\n\t\r') # 词边界分隔符 self.redis_client = redis_client self._load_words() def _add_word(self, word: str): """向前缀树中添加一个敏感词""" node = self.trie for char in word.lower(): # 统一转为小写，实现忽略大小写 node = node.setdefault(char, {}) node['__end__'] = True # 标记词尾 def _load_words(self): """从Redis加载敏感词。如果没有Redis，则从文件加载。""" if self.redis_client: try: # 假设敏感词以集合形式存储在Redis的 key: 'sensitive_words:active' 中 words = self.redis_client.smembers('sensitive_words:active') for word in words: self._add_word(word.decode('utf-8')) print(f"从Redis加载了 {len(words)} 个敏感词。") except Exception as e: print(f"从Redis加载词库失败: {e}，将尝试从文件加载。") self._load_from_file() else: self._load_from_file() def _load_from_file(self): """从本地JSON文件加载敏感词（备用方案）。""" try: with open('sensitive_words.json', 'r', encoding='utf-8') as f: words = json.load(f).get('words', []) for word in words: self._add_word(word) print(f"从文件加载了 {len(words)} 个敏感词。") except FileNotFoundError: print("未找到敏感词文件，词库为空。") def contains_sensitive(self, text: str) -> (bool, List[str]): """ 检查文本是否包含敏感词。 返回：(是否包含, 匹配到的敏感词列表) """ text_lower = text.lower() found_words = [] i = 0 while i < len(text_lower): node = self.trie j = i matched_word = None while j < len(text_lower) and text_lower[j] in node: node = node[text_lower[j]] j += 1 if '__end__' in node: # 检查词边界：当前字符是分隔符或已是文本末尾 if j == len(text_lower) or text_lower[j] in self.delimiters: matched_word = text_lower[i:j] if matched_word: found_words.append(matched_word) i = j # 跳过已匹配的词 else: i += 1 return len(found_words) > 0, found_words def filter_text(self, text: str, replace_char='*') -> str: """过滤文本，将敏感词替换为指定字符。""" sensitive, words = self.contains_sensitive(text) if not sensitive: return text result = text for word in words: # 构建正则表达式，忽略大小写进行替换 pattern = re.compile(re.escape(word), re.IGNORECASE) result = pattern.sub(replace_char * len(word), result) return result # 初始化一个全局过滤器实例 filter_engine = SensitiveWordFilter()

这个引擎实现了基础的前缀树匹配和替换功能。contains_sensitive方法用于检查并返回匹配到的词，filter_text方法用于执行替换。我们将其设计为可以从Redis动态加载词库。

3.3 实现FastAPI代理服务器

创建主应用文件main.py。

from fastapi import FastAPI, HTTPException, Header, Request from fastapi.responses import JSONResponse import openai from openai import OpenAI import httpx import asyncio from filter_engine import filter_engine import logging import time from pydantic import BaseModel from typing import Optional # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) app = FastAPI(title="Dingtalk-OpenAI Filter Proxy") # 配置（应从环境变量读取，此处为示例） OPENAI_API_KEY = "your-openai-api-key-here" OPENAI_BASE_URL = "https://api.openai.com/v1" # 或第三方兼容端点 DINGTALK_ROBOT_SECRET = "your-dingtalk-robot-secret" # 钉钉机器人加签密钥（如果有） PROXY_TIMEOUT = 30 # 初始化OpenAI客户端 client = OpenAI(api_key=OPENAI_API_KEY, base_url=OPENAI_BASE_URL) class DingTalkRequest(BaseModel): """钉钉机器人回调消息格式（简化示例，实际更复杂）""" text: dict msgtype: str conversationId: Optional[str] = None senderId: Optional[str] = None class OpenAIRequest(BaseModel): """兼容OpenAI API的请求体格式""" model: str = "gpt-3.5-turbo" messages: list max_tokens: Optional[int] = 1000 temperature: Optional[float] = 0.7 @app.post("/dingtalk/webhook") async def dingtalk_webhook(request: DingTalkRequest, x_dingtalk_signature: Optional[str] = Header(None)): """ 钉钉机器人的Webhook入口。 1. 验证签名（可选但推荐）。 2. 提取用户消息。 3. 进行敏感词过滤。 4. 调用内部OpenAI代理接口。 """ # 1. 签名验证（略，需根据钉钉文档实现） # if not verify_signature(x_dingtalk_signature, request): # raise HTTPException(status_code=403, detail="Invalid signature") user_message = request.text.get('content', '').strip() user_id = request.senderId or "unknown" logger.info(f"收到来自用户 {user_id} 的消息: {user_message[:50]}...") # 2. 敏感词检查（请求过滤） is_sensitive, hit_words = filter_engine.contains_sensitive(user_message) if is_sensitive: logger.warning(f"消息拦截！用户 {user_id} 触发了敏感词: {hit_words}") # 返回友好提示给钉钉用户 return JSONResponse(content={ "msgtype": "text", "text": {"content": f"您的问题中包含敏感词汇，请重新组织语言。触发的词：{', '.join(hit_words)}"} }) # 3. 构造OpenAI请求 openai_req_body = OpenAIRequest( messages=[{"role": "user", "content": user_message}], model="gpt-3.5-turbo" ) # 4. 调用内部代理端点（下一步实现） try: async with httpx.AsyncClient(timeout=PROXY_TIMEOUT) as client_http: response = await client_http.post( "http://localhost:8000/openai/chat/completions", # 指向本服务的另一个端点 json=openai_req_body.dict(), headers={"Content-Type": "application/json"} ) response.raise_for_status() ai_response = response.json() except Exception as e: logger.error(f"调用内部AI服务失败: {e}") return JSONResponse(content={ "msgtype": "text", "text": {"content": "AI服务暂时不可用，请稍后再试。"} }) ai_content = ai_response['choices'][0]['message']['content'] # 5. 敏感词检查（响应过滤）- 防止AI生成敏感内容 filtered_ai_content = filter_engine.filter_text(ai_content) logger.info(f"用户 {user_id} 的请求处理完成，原始AI回复长度: {len(ai_content)}， 过滤后长度: {len(filtered_ai_content)}") # 6. 返回结果给钉钉 return JSONResponse(content={ "msgtype": "text", "text": {"content": filtered_ai_content} }) @app.post("/openai/chat/completions") async def openai_proxy(request: OpenAIRequest, authorization: Optional[str] = Header(None)): """ 内部使用的OpenAI代理端点。 1. 可在此处添加额外的认证（如内部API Key）。 2. 直接转发请求至真实的OpenAI API。 """ # 1. 内部认证（可选） # if not valid_internal_token(authorization): # raise HTTPException(status_code=401, detail="Unauthorized") logger.info(f"转发OpenAI请求，模型: {request.model}") try: # 2. 使用OpenAI官方SDK调用 start_time = time.time() response = await client.chat.completions.create( model=request.model, messages=request.messages, max_tokens=request.max_tokens, temperature=request.temperature ) elapsed = time.time() - start_time # 3. 格式化响应以兼容OpenAI格式 resp_data = { "id": response.id, "object": response.object, "created": response.created, "model": response.model, "choices": [{ "index": choice.index, "message": { "role": choice.message.role, "content": choice.message.content }, "finish_reason": choice.finish_reason } for choice in response.choices], "usage": { "prompt_tokens": response.usage.prompt_tokens, "completion_tokens": response.usage.completion_tokens, "total_tokens": response.usage.total_tokens } } logger.info(f"OpenAI API调用成功，耗时: {elapsed:.2f}s, 消耗token: {resp_data['usage']['total_tokens']}") return resp_data except openai.APIConnectionError as e: logger.error(f"连接OpenAI失败: {e}") raise HTTPException(status_code=504, detail="Upstream service connection error") except openai.APIStatusError as e: logger.error(f"OpenAI API返回错误状态: {e.status_code} - {e.response}") raise HTTPException(status_code=e.status_code, detail=f"OpenAI error: {e.message}") except Exception as e: logger.error(f"处理OpenAI请求时未知错误: {e}") raise HTTPException(status_code=500, detail="Internal server error") @app.on_event("startup") async def startup_event(): """服务启动时，可以重新加载词库或初始化连接池""" # 例如，初始化Redis连接并更新过滤器 # redis_client = redis.Redis(host='localhost', port=6379, db=0) # filter_engine.redis_client = redis_client # filter_engine._load_words() logger.info("Dingtalk-OpenAI过滤代理服务启动完成。") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

这个main.py文件创建了两个核心端点：

1. /dingtalk/webhook: 钉钉机器人配置的Webhook地址。它接收用户消息，进行请求过滤，然后调用内部代理。
2. /openai/chat/completions: 内部代理端点。它负责将过滤后的请求转发给真实的OpenAI API，并将返回的结果交给上一个端点进行响应过滤。

实操心得二：异步与超时设置注意我们使用了async/await和httpx.AsyncClient。因为网络I/O是主要耗时操作，异步能显著提升并发处理能力。PROXY_TIMEOUT超时设置至关重要（这里设为30秒），必须小于钉钉机器人的超时时间（通常也是30秒），否则会导致钉钉侧请求失败。如果OpenAI API响应慢，可以考虑在代理层设置更短的超时，并返回“服务繁忙”的友好提示。

3.4 配置钉钉机器人

1. 在钉钉群（或企业内部应用）中，添加一个“自定义机器人”。
2. 在安全设置中，选择“加签”或“IP白名单”（推荐加签，更安全）。获取到WebhookURL 和Secret。
3. 将WebhookURL 配置为你的代理服务器公网地址 +/dingtalk/webhook，例如https://your-proxy.com/dingtalk/webhook。
4. 在你的代理服务代码中（上面示例中注释掉了），实现钉钉的签名验证逻辑，确保请求来源合法。

3.5 部署与运行

1. 本地测试：直接在项目目录运行python main.py，服务将在http://localhost:8000启动。使用ngrok或localtunnel等工具将本地端口暴露到公网，获得一个临时HTTPS地址，填入钉钉机器人Webhook进行测试。
2. 服务器部署：
   • 将代码上传至云服务器（如阿里云ECS、腾讯云CVM）。
   • 使用gunicorn或uvicorn作为生产级ASGI服务器。例如：gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app --bind 0.0.0.0:8000。
   • 使用Nginx作为反向代理，配置SSL证书（HTTPS是钉钉Webhook的强制要求），并设置负载均衡和静态文件服务。
   • 使用systemd或Supervisor管理进程，保证服务持续运行。
   • 将敏感配置（API Key、数据库密码）写入环境变量或配置文件，切勿硬编码在代码中。

4. 高级策略与优化实践

基础过滤搭建完成后，为了应对更复杂的场景和提高可用性，需要考虑以下进阶策略。

4.1 语义理解与上下文过滤

简单的关键词匹配存在明显局限：

• 误杀：公司名“金盾”可能包含敏感字“盾”，但本身是合法名称。
• 漏杀：通过拼音、谐音、拆字、插入无关字符等方式绕过（如“敏gan词”）。

应对策略：

1. 同义词/近义词扩展：在词库中，不仅存储“机密”，也存储“绝密”、“秘密”、“内部资料”等。可以利用词向量模型（如Word2Vec、BERT）自动挖掘近义词。
2. 正则表达式模式：针对特定模式，如电话号码、身份证号、银行卡号，使用正则表达式进行匹配和脱敏。例如，将文本中所有11位手机号替换为[手机号已屏蔽]。
3. NLP模型辅助：对于高安全等级场景，可以引入轻量级文本分类模型。将用户问题向量化后，判断其整体意图是否涉及“数据索取”、“代码生成”、“内容总结”（可能涉及泄密），而不仅仅是字面匹配。可以基于BERT微调一个小型分类器，作为关键词过滤的补充。

4.2 动态词库与分级管理

词库不能是静态的，需要一套管理流程。

1. 分级词库：

   • 黑名单：明确禁止的词汇，触发即拦截。
   • 灰名单：疑似或需要关注的词汇，触发后只告警不拦截，并通知安全管理员复核。适用于新业务术语或不确定的词汇。
   • 白名单：明确放行的词汇，即使包含敏感字符也直接通过（如公司产品名“微信助手”）。

2. 管理后台：开发一个简单的Web管理界面，允许安全管理员：

   • 添加、删除、修改敏感词，并指定所属名单和分类。
   • 查看实时拦截/告警日志。
   • 对灰名单告警进行“确认为误报”或“升级为黑名单”操作。

3. 词库热更新：通过Redis的Pub/Sub功能，当管理后台更新词库时，广播消息给所有运行中的代理服务实例，触发它们重新从Redis加载词库，实现秒级生效，无需重启服务。

4.3 审计、监控与告警

安全机制必须有可观测性。

1. 全量日志记录：记录每一次请求的元数据。

   { "timestamp": "2023-10-27T10:00:00Z", "user_id": "dingtalk_user_123", "request_content": "原始用户消息（已脱敏）", "response_content_preview": "AI回复前50字...", "sensitive_words_hit": ["项目代号A", "内部价格"], "action": "blocked", // 或 "passed", "warning" "openai_request_id": "chatcmpl-xxx", "processing_time_ms": 1200 }

   这些日志应存入Elasticsearch或专门的日志平台，便于检索和分析。

2. 关键监控指标：

   • 拦截率：触发敏感词过滤的请求比例。突然升高可能意味着新的信息泄露风险或词库误判严重。
   • 平均响应延迟：代理服务引入的额外延迟。确保在可接受范围内（如<500ms）。
   • OpenAI API调用错误率：监控上游服务健康状态。
   • 高频触发用户/关键词：发现潜在风险点或需要优化的词条。

3. 告警机制：当出现以下情况时，及时通知管理员（通过钉钉机器人、邮件等）：

   • 短时间内同一敏感词被大量触发。
   • 拦截率超过阈值。
   • 服务本身出现错误或宕机。

5. 常见问题与故障排查实录

在实际部署和运维中，你肯定会遇到各种问题。以下是我和团队踩过的一些坑以及解决方案。

5.1 性能瓶颈与优化

问题现象：服务响应变慢，尤其在消息较长或并发量稍高时。

排查与解决：

1. 检查过滤算法：最原始的循环遍历字符串匹配性能极差。必须使用Trie树或AC自动机。用cProfile或py-spy工具分析Python代码，确认耗时主要在过滤函数。
2. 词库加载方式：每次请求都从数据库读词库是不可接受的。必须使用内存缓存（如Redis），并且服务启动时全量加载到进程内存中的Trie树结构。我们的方案是Redis缓存+内存Trie树，通过Pub/Sub实现热更新。
3. 异步处理：确保整个请求链路是异步的（如使用httpx.AsyncClient调用OpenAI），避免在过滤等CPU密集型操作上使用异步（这不会提升性能，反而可能因事件循环阻塞而降低）。对于CPU密集的过滤，如果确实成为瓶颈，可以考虑将其放入单独的线程池执行。
4. 硬件与扩容：如果经过上述优化仍不满足，考虑横向扩展。使用Nginx进行负载均衡，部署多个代理服务实例。数据库（词库）和Redis也需要相应升级。

5.2 误拦截与漏拦截

问题现象：正常业务对话被阻断，或者明显的敏感信息却没有被识别。

解决方案：

1. 建立误报反馈通道：在钉钉机器人返回拦截信息时，提供一个简单的反馈入口，例如“如果这是误报，请点击[这里]反馈”。将反馈信息自动录入工单系统或通知管理员。
2. 实施灰名单机制：如前所述，对于新词或不确定的词，先进入灰名单观察。这是降低误报影响最有效的手段。
3. 定期词库审计：每周或每半月，安全团队需要复审拦截日志。对于高频触发但最终被判定为误报的词，考虑将其加入白名单或修改匹配规则（如改为更精确的正则）。
4. 对抗漏报：
   • 定期渗透测试：让安全团队或可信员工尝试用各种方式（谐音、拆字、代码注释、图片OCR后粘贴等）测试绕过过滤，不断补充对抗样本到词库。
   • 关注AI生成内容：漏报不仅发生在用户输入，也可能发生在AI回复。务必确保响应过滤逻辑同样严密，并且对AI可能生成的“请提供更多XX信息”这类诱导性话语保持警惕，可以将其加入词库。

5.3 服务稳定性保障

问题现象：代理服务宕机，导致所有AI功能不可用。

高可用设计：

1. 无状态服务：代理服务本身不应保存会话状态。这样任何一台实例宕机，流量可以立刻被其他实例接管。
2. 健康检查与负载均衡：在Kubernetes或负载均衡器（如Nginx）上配置/health健康检查端点，定期探测，自动剔除不健康的实例。
3. 降级策略：
   • 监控OpenAI API状态：如果检测到OpenAI API长时间不可用或错误率飙升，代理服务可以自动切换到一个简单的“服务降级”模式，例如返回预设的提示：“AI服务维护中，请稍后尝试”。
   • 过滤服务降级：如果Redis宕机导致词库无法加载，可以降级为使用本地备份的静态词库文件，并记录告警。切勿在词库加载失败时直接放行所有请求，这会导致安全防线彻底失效。
4. 限流与熔断：在代理入口实施限流（如使用slowapi或redis做令牌桶），防止异常流量打垮服务或消耗过多OpenAI API额度。对OpenAI的调用配置熔断器（如pybreaker），在连续失败多次后自动熔断，避免雪崩。

5.4 钉钉交互相关错误

问题现象：钉钉机器人收不到回复，或提示“消息发送失败”。

排查步骤：

1. 检查签名：这是最常见的问题。确保你的签名算法与钉钉文档完全一致，注意时间戳的同步性。建议在代码中实现签名验证，并在日志中打印验证结果。
2. 检查超时：钉钉机器人发送消息到你的Webhook，默认超时是30秒。确保你的代理服务整体处理时间（过滤+调用OpenAI+响应过滤）远小于30秒。如果OpenAI响应慢，考虑在代理层设置更短的超时（如15秒），并提前返回“思考超时”的友好提示。
3. 检查响应格式：钉钉要求返回特定格式的JSON。务必严格按照钉钉开放平台的文档来构造响应体，包括msgtype、text等字段。一个字段错误就可能导致钉钉侧解析失败。
4. 检查网络与HTTPS：确保你的代理服务器地址是公网可访问的，并且配置了有效的SSL证书（钉钉强制要求HTTPS）。可以使用curl或Postman手动向你的Webhook发送一条测试消息，查看返回。

实操心得三：测试驱动开发在开发阶段，就为过滤引擎编写完备的单元测试，覆盖各种边界情况：空字符串、超长字符串、混合大小写、包含标点、Unicode字符、重叠词（如“中国人”和“中国”）。为代理服务的API编写集成测试，模拟钉钉的请求和OpenAI的响应。这能极大减少上线后的诡异问题。我们曾因为一个全角空格匹配问题，导致某个重要产品型号被误拦截，正是通过补充测试用例才彻底修复。