如果你每天的工作和交流离不开微信和钉钉,大概率动过这个念头:能不能让 Gemini 直接待在我的聊天框里,随手发一句话,它就把翻译、写周报、解释术语、生成草稿这些事干了?我最近把 Gemini 接进了企业微信和钉钉,做成一个可以随时 @ 的 AI 助手。整个过程不复杂,核心就三件事:准备 API Key、搭一个消息接收服务、把 Gemini 的回复塞回聊天框。这篇内容就是我实际接通的完整记录,从账号后台配置到部署避坑都写了,适合有一定 Python 基础、想给团队或个人搭一个专属 AI 助手的开发者参考。
先说一个很容易踩的误区:不要想着用个人微信号去接。个人微信没有官方机器人接口,用脚本模拟登录不仅不稳定,还有各种风险,企业也不会允许。微信侧我选的是企业微信自建应用,钉钉侧用的是钉钉企业内部机器人,这两条都是官方支持的接入方式。整个项目代码量不大,但涉及签名校验、会话上下文、超时处理这些细节,我会把自己踩过的坑和取舍一起放进来。
1. 为什么是企业微信和钉钉:选型思路与整体架构
1.1 先别急着找个人微信机器人
如果你的最终目的是给自己一个人用,听到“接入微信”确实很容易想到个人微信。但个人微信的登录协议、消息收发协议都没有对外开放,市面上的“微信机器人”基本都是逆向或挂机方案。这类方案有两个致命问题:一是账号随时可能被限制,二是消息内容要经过第三方服务器,安全性没法保证。企业场景里我不建议碰。
企业微信和钉钉则完全不同。企业微信的“自建应用”允许你设置一个接收消息的服务器,成员在单聊或群里 @ 应用机器人时,企业微信服务器会把消息内容推送到你的服务端,你的服务端处理后返回给企业微信。钉钉的企业内部机器人更直接,它提供了官方的 Stream 长连接模式,甚至不需要你有公网回调地址。这两条路都是纯官方接口,权限边界清楚,数据链路可控,适合做正经的 AI 助手。
1.2 一条消息从聊天框到 Gemini 再回来的完整链路
先理解整体流转,后面的代码才有方向。
- 用户在聊天窗口发了一条文本消息。
- 企业微信或钉钉的服务器收到它,判断这条消息要推给哪个机器人。
- 平台把消息内容、发送人、会话 ID 等信息交给你的服务端。
- 你的服务端调用 Gemini API,把用户提问塞进去。
- Gemini 返回文本后,你的服务端把结果构造成一条回复消息,还给平台。
- 平台把回复展示在使用者的聊天窗口里。
企业微信和钉钉的差别主要在接入方式上。企业微信是“HTTP 回调模式”,需要你提供一个公网能访问的接口,并且要做签名验证和 AES 加解密;钉钉 Stream 模式则是 SDK 主动和钉钉服务器建立长连接,消息到了以后推给你,省掉了公网回调和加解密这一整块。所以我会把两边的代码分开写,但共用同一个 Gemini 调用函数。
1.3 两份密钥搞清楚:API Key、Token、AESKey 都是干什么的
新手最容易在密钥这里混乱。我把需要用到的凭证列成一张表,后面配置的时候对着填就行。
| 凭证 | 所属平台 | 作用 |
|---|---|---|
| GEMINI_API_KEY | Google Gemini | 访问 Gemini 模型的唯一凭证 |
| 企业 ID(CorpID) | 企业微信 | 标识你的企业 |
| AgentId、Secret | 企业微信 | 自建应用的唯一标识和访问凭证 |
| Token、EncodingAESKey | 企业微信 | 消息回调的签名校验和内容加解密 |
| Client ID(AppKey) | 钉钉 | 钉钉应用的唯一标识 |
| Client Secret | 钉钉 | 钉钉应用的访问密钥 |
可以这么理解:Gemini 的 API Key 是你从 Google 那边拿到的“模型门票”;企业微信 Token 和 EncodingAESKey 是你和腾讯之间约定好的“暗号”,防止别人伪造消息推到你服务器;钉钉的 Client ID 和 Client Secret 则管着你这个应用的登录身份。把这六七个变量放到环境变量文件里统一管理,后面代码里不写死任何密钥。
需要模型API调用? 免费领10W Token,多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 开工前准备:账号权限、模型 Key 和 Python 环境
2.1 拿到 Gemini API Key 并确认可调用
先去 Google AI Studio 的 API Key 页面创建一个 Key,这个 Key 只在生成时完整显示一次,记得立刻存到你自己的密码管理器里。创建完之后,先用一个最简单的请求验证它能不能通。
bash复制export GEMINI_API_KEY="你的APIKey"
curl "https://generativelanguage.googleapis.com/v1beta/models/gemini-2.0-flash:generateContent?key=$GEMINI_API_KEY" \
-H 'Content-Type: application/json' \
-d '{"contents":[{"parts":[{"text":"你好,请回复:我收到了"}]}]}'
如果返回里能看到 "text" 字段,说明 Key 有效。模型名我建议先用 gemini-2.0-flash,它响应快、价格便宜,特别适合对话机器人。如果你需要更强的推理能力,之后把代码里的模型名换掉即可。
注意:API Key 一定要放在服务端环境变量里,不要写进前端页面,也不要提交到 Git 仓库。项目代码里我统一用 dotenv 读取。
2.2 在企业微信管理后台创建自建应用
企业微信这边需要管理员权限,或者让管理员协助你操作。登录企业微信管理后台,进入“应用管理 - 应用 - 自建”,创建一个自建应用,填上应用 Logo 和名称,保存后会得到 AgentId 和 Secret。
紧接着进入应用的“接收消息”配置页,页面上会让你填三个东西:
- URL:你的回调接口地址,例如
https://your-domain.com/wecom/callback - Token:自己随便生成一个字符串,建议长一点
- EncodingAESKey:点击随机生成,系统会给你一个 43 位字符串
把 Token 和 EncodingAESKey 记下来,同时把企业 ID(CorpID)也找到,它在“我的企业 - 企业信息”里。这三样是后面企业微信接入的命根子。
还需要注意“企业可信 IP”配置。如果你的服务端有固定出口 IP,把它加到符合调用接口的 IP 白名单里,否则部分主动调用接口会拒绝。我们这里主要是被动接收消息,影响不大,但建议顺手配上。
2.3 在钉钉开放平台创建企业内部机器人
钉钉侧的准备工作更轻一些。进入钉钉开放平台,在“开发者后台”里创建一个企业内部应用,创建时选择“企业内部应用”,完成后你会拿到 AppKey 和 AppSecret。注意,钉钉的 AppKey 在客户端代码里也叫 Client ID,别被两个叫法搞混。
然后在应用内添加“机器人”,创建机器人时选择“自定义机器人”里的 Stream 模式,不需要填回调 URL。这也是我推荐钉钉用 Stream 模式的原因:它不需要你有一台公网可访问的服务器,开发阶段笔记本上就能联调。
权限方面,机器人通常需要申请“读取消息”和“发送消息”的权限。如果之后要主动给用户发消息,还需要申请对应权限并发布应用,企业内部应用一般审批很快。
2.4 本地开发环境与依赖清单
这个项目我用 Python 3.10 开发,操作系统没有特殊要求。先建一个虚拟环境,然后安装依赖。
bash复制python3 -m venv venv
source venv/bin/activate
pip install fastapi uvicorn python-dotenv requests google-generativeai wechatpy dingtalk-stream
依赖说明:
fastapi和uvicorn:跑企业微信回调接口google-generativeai:官方 Gemini SDKwechatpy:封装了企业微信的签名校验、XML 加解密、消息解析,比自己手写省很多事dingtalk-stream:钉钉官方 Stream 模式 SDKpython-dotenv:读取.env配置文件
安装完可以准备一个 .env 文件,把前面拿到的所有凭证放进去。
bash复制GEMINI_API_KEY=your_gemini_key
GEMINI_MODEL=gemini-2.0-flash
WECOM_CORP_ID=your_corp_id
WECOM_SECRET=your_secret
WECOM_AGENT_ID=your_agent_id
WECOM_TOKEN=your_token
WECOM_AES_KEY=your_encoding_aes_key
DINGTALK_CLIENT_ID=your_dingtalk_appkey
DINGTALK_CLIENT_SECRET=your_dingtalk_appsecret
3. 企业微信接入:回调验证、消息解密与 Gemini 回复
3.1 配置接收消息服务器的三个关键参数
企业微信的接入方式本质上是一个带签名验证的 HTTP 服务。当你在后台保存回调配置时,企业微信服务器会先向你填写的 URL 发送一个 GET 请求,带上 msg_signature、timestamp、nonce、echostr 四个参数。你的服务必须验签成功并解密 echostr 原样返回,企业微信才认为这个 URL 是你自己的。
配置里最容易错的两个点:
- Token、EncodingAESKey、CorpID 三者必须和你代码里读到的一致,任何一处不对都会验签失败。
- URL 必须是公网可达的地址,且服务必须已经启动。先启动本地服务再填后台,不然保存时验证不过去。
3.2 FastAPI 实现回调 URL 验证接口
下面是一份可以直接跑起来的最小服务。为了和真实项目匹配,我把企业微信相关的处理都放在 wechat_bot.py 里。
python复制# wechat_bot.py
import os
from fastapi import FastAPI, Request
from fastapi.responses import PlainTextResponse
from wechatpy.enterprise.crypto import WeChatCrypto
from wechatpy.exceptions import InvalidSignatureException
from wechatpy.enterprise import parse_message
from wechatpy.enterprise.replies import TextReply
from dotenv import load_dotenv
import gemini_client
load_dotenv()
app = FastAPI()
crypto = WeChatCrypto(
token=os.getenv("WECOM_TOKEN"),
encoding_aes_key=os.getenv("WECOM_AES_KEY"),
corp_id=os.getenv("WECOM_CORP_ID"),
)
@app.get("/wecom/callback")
async def verify_url(msg_signature: str, timestamp: str, nonce: str, echostr: str):
try:
if not crypto.check_signature(msg_signature, timestamp, nonce, echostr):
return PlainTextResponse("signature error", status_code=403)
return PlainTextResponse(crypto.decrypt(echostr))
except InvalidSignatureException:
return PlainTextResponse("signature error", status_code=403)
@app.post("/wecom/callback")
async def handle_message(request: Request, msg_signature: str, timestamp: str, nonce: str):
raw = await request.body()
try:
xml_text = crypto.decrypt_message(
raw.decode("utf-8"), msg_signature, timestamp, nonce
)
msg = parse_message(xml_text)
if msg.type == "text":
reply_content = gemini_client.ask_gemini(msg.content)
else:
reply_content = "我暂时只能处理文本消息,图片、语音和文件我还在学。"
reply = TextReply(content=reply_content, message=msg)
encrypted = crypto.encrypt_message(reply.render(), nonce, timestamp)
return PlainTextResponse(encrypted)
except Exception as exc:
print("handle_message error:", exc)
return PlainTextResponse("error", status_code=500)
这段代码解决的问题是:
- 别人伪造的请求在
check_signature这一层就会被拦住。 - 企业微信推送的正文是 XML 格式,并且整体被 AES 加密过,所以先解密再交给 SDK 解析。
- 构造回复时也要把 XML 重新加密,直接明文返回企微是不认的。
- 回复内容用
PlainTextResponse返回纯文本,避免 FastAPI 默认把它当 JSON 处理。
3.3 读取并解密企微消息
这里有个细节值得单独说明。企业微信的 POST 回调请求体不是 JSON,而是加密后的 XML 字符串。FastAPI 里如果你写 await request.json() 一定报错,正确方式是先读原始 body:
python复制raw = await request.body()
然后交给 crypto.decrypt_message 解密。解密后得到的 XML 会给到 parse_message,SDK 根据 XML 根节点自动判断消息类型。对文本消息来说,msg.content 就是用户发的内容,msg.source 是发送人,msg.agent_id 是应用 ID。环境比较复杂时,可以用这几个字段做权限控制。
3.4 调用 Gemini 并构造回复
Gemini 调用我单独抽成了 gemini_client.py,这样企业微信和钉钉都能复用。先写一个最基础的单轮调用版本。
python复制# gemini_client.py
import os
import google.generativeai as genai
from dotenv import load_dotenv
load_dotenv()
genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
MODEL_NAME = os.getenv("GEMINI_MODEL", "gemini-2.0-flash")
SYSTEM_PROMPT = "你是一个在企业微信和钉钉里的 AI 助手,回答简洁、准确、友好。"
model = genai.GenerativeModel(MODEL_NAME, system_instruction=SYSTEM_PROMPT)
def ask_gemini(prompt: str) -> str:
try:
response = model.generate_content(prompt)
if response.prompt_feedback.block_reason:
return "这个问题我目前无法回答,换个说法再试一次。"
return response.text.strip()
except ValueError:
return "模型没有返回内容,可能触发了安全限制。"
except Exception as exc:
print("Gemini error:", exc)
return "服务暂时开小差了,稍后再试。"
企业微信对回包时间要求比较严,正常情况下需要在 5 秒内返回。如果 Gemini 偶尔变慢,最简单的处理是让它超时后返回一句固定文案,不要直接 500。否则企业微信会判定服务异常并重新推送,同一个消息可能被触发多次。
3.5 上线前在企微工作台里试发一条消息
服务启动后,先把 FastAPI 跑起来:
bash复制uvicorn wechat_bot:app --host 0.0.0.0 --port 8000
然后在企业微信后台填写回调地址,点保存,如果看到“验证成功”的提示,说明 URL 验签、解密都通了。接下来进入企业微信工作台,找到你创建的应用,给它发一条“你好”,机器人应该会回一句 Gemini 生成的内容。
如果你在群里测试,记得先把应用机器人拉进群里,并且用 @ 的方式发消息。企微群的语义是只有被 @ 的消息才会推给机器人,这个不是代码问题,是平台规则。
4. 钉钉接入:用 Stream 模式省掉回调服务器的麻烦
4.1 钉钉机器人的 Client ID 和 Client Secret
钉钉侧我把接入方式定为 Stream 模式,因为它是我实际用过最舒服的接入方式。为什么这么说?HTTP 回调模式需要你有公网接口,还要处理签名、时间戳、nonce、AES 加解密,还没开始写业务逻辑就被安全流程磨掉一层皮。Stream 模式是 SDK 主动和钉钉服务器建立一条长连接,消息推送过来后直接触发你的处理函数,不需要公网回调,也不需要处理加解密。
配置时只需要两样东西:
- Client ID:对应钉钉开放平台上的 AppKey
- Client Secret:对应 AppSecret
注意这里叫 Client ID 而不是 AppKey,是因为 dingtalk-stream SDK 的 Credentials 参数名就叫 client_id 和 client_secret,填写时别对应错了。
4.2 Stream 模式的接入代码骨架
dingtalk_bot.py 的完整结构如下:
python复制# dingtalk_bot.py
import os
import re
import dingtalk_stream
from dingtalk_stream import AckMessage
from dotenv import load_dotenv
import gemini_client
load_dotenv()
def build_prompt(message_data: dict) -> str:
text = message_data["text"]["content"].strip()
# 群聊消息里会带 @机器人 的文本,先把 @ 去掉
at_list = message_data.get("atUsers", [])
if at_list:
text = re.sub(r"@[^\s]+", "", text).strip()
return text
class GeminiChatbotHandler(dingtalk_stream.ChatbotHandler):
async def process(self, callback: dingtalk_stream.CallbackMessage):
message_data = callback.data
session_key = message_data.get("conversationId", "default")
user_input = build_prompt(message_data)
if not user_input:
return AckMessage.STATUS_OK, "OK"
reply = gemini_client.ask_gemini(user_input)
conversation_id = message_data["conversationId"]
self.reply_text(reply, in_chat=conversation_id)
return AckMessage.STATUS_OK, "OK"
def main():
credentials = dingtalk_stream.Credentials(
client_id=os.getenv("DINGTALK_CLIENT_ID"),
client_secret=os.getenv("DINGTALK_CLIENT_SECRET"),
)
client = dingtalk_stream.DingTalkStreamClient(credentials)
client.register_callback_handler(dingtalk_stream.ChatbotHandler.TOPIC, GeminiChatbotHandler())
client.start_forever()
if __name__ == "__main__":
main()
启动方式很简单:
bash复制python dingtalk_bot.py
运行后控制台会输出连接成功之类的日志,然后去钉钉里找到这个机器人,给它发一条消息,它就会通过 Gemini 回你。
4.3 Stream 模式如何区分群聊和单聊
Stream 模式下,每条消息的数据结构里都有一个 conversationId。这个 ID 在单聊里是“用户和机器人之间的会话”,在群聊里则是“群和机器人之间的会话”。所以:
- 你想让每个用户各自拥有独立上下文,就用发送人 ID 做 key。
- 你想让一个群共享同一个上下文,用 conversationId 做 key 更合适。
- 你想让同一个人在群里和单聊里互不干扰,最好把 conversationId 和发送人 ID 拼起来。
我自己的项目用的是 conversationId,原因是企业内部使用场景里,大家更习惯在同一个群里一起问同一个助手,上下文共享反而方便。
4.4 把同一个 Gemini 处理函数复用到钉钉
你现在看到的 gemini_client.py 只有一个 ask_gemini(prompt),它不代表多轮对话。要想让机器人记住上下文,需要给 Gemini 传历史会话。这个我会在下一节展开。
复用逻辑也简单:钉钉处理函数里不需要关心企业微信的加密、签名,只需要拿到文本、调用 gemini_client、把结果 reply 回去。也就是说,你以后想接飞书、Teams,只要写出对应的消息适配层,大脑始终是同一个 gemini_client。
5. 多轮对话和指令设计:从一个“回声机器人”变成真正的助手
5.1 每个会话要有自己的上下文
单轮调用只能解决“一问一答”。但一个合格的 AI 助手至少要记得用户上一句在聊什么。Gemini SDK 里提供了 start_chat(history=...),可以把多轮消息塞进去。我维护了一个简单的内存字典,key 是会话 ID,value 是最近若干轮消息。
python复制# gemini_client.py 追加部分
from collections import defaultdict
conversation_histories = defaultdict(list)
MAX_HISTORY_LEN = 10
def get_history(session_key: str) -> list:
return conversation_histories.get(session_key, [])
def append_history(session_key: str, user_text: str, assistant_text: str):
history = conversation_histories[session_key]
history.append({"role": "user", "parts": [user_text]})
history.append({"role": "model", "parts": [assistant_text]})
if len(history) > MAX_HISTORY_LEN:
conversation_histories[session_key] = history[-MAX_HISTORY_LEN:]
def ask_gemini_with_history(prompt: str, history: list) -> str:
chat = model.start_chat(history=history)
try:
response = chat.send_message(prompt)
return response.text.strip()
except Exception as exc:
print("Gemini chat error:", exc)
return "我这边临时出错了,请再发一次。"
这个“只保留最近 10 条消息”的策略很重要。如果无限累积,请求体积会越来越大,延迟越来越明显,费用也会失控。
注意:内存字典这种方案只适合单进程、低并发的小范围使用。如果服务部署成多个 worker 进程,会话字典会各自独立,就会出现“上次记得、这次忘了”的情况。要真正解决,可以把会话存到 Redis,key 就是 session_key,value 就是一个 JSON 数组。
5.2 用 System Prompt 约束机器人角色
模型能力强,不代表它知道在聊天框里该以什么风格说话。Gemini 的 system_instruction 就是用来约束整体人设的。我实际用的 System Prompt 大概是这样的:
text复制你是公司内部的 AI 助手,名字叫小 G。
回答要简洁,默认使用中文,能用 3 句话讲清楚就不要写 3 段。
如果问题涉及公司机密或你不确定的信息,明确说不知道。
不要扮演人类,不要编造事实。
把这段写进 SYSTEM_PROMPT 后,群里有人问“今天天气怎么样”,模型就不会啰啰嗦嗦扯到你电脑里有没有天气数据,而是会告诉你它没有获取实时信息的能力。
System Prompt 别写太长,控制在几百字内就够了。它每次请求都会从头带一遍,太长不仅浪费 token,还可能影响响应速度。
5.3 怎么实现 @机器人 才回复
企业微信和钉钉在群聊场景下,默认只会把“包含 @机器人”的消息推送给机器人。所以很多情况下你不用在代码里额外判断。但钉钉的 Stream 消息里,text.content 会带上类似 @小G 这样的文本,如果直接把这段原文交给 Gemini,它会被这串符号干扰。所以我在 build_prompt 里用正则把它去掉。
对企业微信来说,SDK 解析后拿到的是纯文本消息,一般不会把 @ 串带进 content,所以处理更简单。如果你想实现“在群里只有 @ 才回复”,不需要额外写规则;如果你想实现“群里无论是否 @ 都回复”,这个就要看平台是否支持,企微和钉钉目前默认都不允许机器人接收所有群消息,避免打扰。
5.4 对话轮数、token 长度和成本控制
企业里的实际使用场景很杂,有人拿它翻译,有人拿它写周报,有人拿它追问一整段对话。控制成本可以从这几个方向同时下手:
- 限制上下文轮数:只保留最近 5 轮,而不是无限保留。
- 限制单轮输出长度:在 System Prompt 里加一句“回答控制在 200 字以内”,比在代码里做字符截断要自然得多。
- 模型分档:日常闲聊和小任务用
gemini-2.0-flash,复杂的代码审查才切到更强的 Pro 模型。 - 废消息直接拦截:空文本、纯标点、连续相同消息不调用 Gemini,省掉无效消耗。
我还在代码里加了一个简单的限流:同一会话 1 秒内最多触发一次。避免有人连按回车把 Gemini 打爆,也避免自己多花冤枉钱。
6. 部署到服务器之后:长驻进程、超时重试和典型报错排查
6.1 用 Supervisor 托管 Python 进程
本地调试没问题后,把代码推到服务器。企业微信回调接口必须保持在线,钉钉 Stream 也需要长驻进程,所以我用 Supervisor 来管这两个服务。先安装:
bash复制apt install supervisor
然后在 /etc/supervisor/conf.d/bot.conf 里写两段配置。
ini复制[program:wecom-bot]
command=/home/ubuntu/bot/venv/bin/uvicorn wechat_bot:app --host 0.0.0.0 --port 8000
directory=/home/ubuntu/bot
user=ubuntu
autostart=true
autorestart=true
stdout_logfile=/var/log/wecom-bot.log
stderr_logfile=/var/log/wecom-bot-error.log
[program:dingtalk-bot]
command=/home/ubuntu/bot/venv/bin/python dingtalk_bot.py
directory=/home/ubuntu/bot
user=ubuntu
autostart=true
autorestart=true
stdout_logfile=/var/log/dingtalk-bot.log
stderr_logfile=/var/log/dingtalk-bot-error.log
改完后重载配置:
bash复制supervisorctl reread
supervisorctl update
supervisorctl status
用 Supervisor 而不是直接在终端里 python dingtalk_bot.py,是因为长连接进程一旦断开会自动拉起,日志也会统一写到文件,出问题时不用跑到终端前面看。
6.2 消息幂等设计和 Gemini 超时重试
平台在回调超时或网络抖动时,可能会重推同一条消息。如果不做幂等,用户会看到同一个问题被 Gemini 回答两次,甚至更多次。
我在代码里加了一个简单的去重集合:
python复制processed_msg_ids = set()
def is_processed(msg_id: str) -> bool:
if msg_id in processed_msg_ids:
return True
processed_msg_ids.add(msg_id)
if len(processed_msg_ids) > 1000:
processed_msg_ids.clear()
return False
企业微信的 msg.msg_id 和钉钉的 messageData["msgId"] 都适合做这个 key。
超时方面,Gemini 偶尔会因为网络波动或模型负载返回变慢。我的做法是给 SDK 调用包一层 try/except,返回一句“服务暂时开小差了,请稍后再试”。同时记录日志,方便事后分析。
如果你希望用户在等待时能看到“正在输入”的反馈,可以结合平台能力。企业微信被动回复的 5 秒限制比较紧,更稳妥的方案是:先立即返回一个空内容或提示,让 worker 后台生成结果后调用企业微信主动推送消息接口。钉钉 Stream 模式没有这么严的时长限制,但建议也控制在 10 秒内,体验更接近真人。
6.3 我在实际测试中踩过的几个坑
我把真实遇到的坑和解决办法整理成一张表,每条都挺典型。
| 现象 | 原因 | 解决办法 |
|---|---|---|
| 企业微信后台保存回调 URL 提示验证失败 | Token、AESKey、CorpID 有一处对不上 | 逐项核对,AESKey 必须是 43 位 |
| 企业微信 URL 验证通过,但收不到消息 | 应用没有开启“接收消息”API | 在应用配置里开启接收消息,并重新保存 |
| 企业微信收到了消息但回复乱码 | 返回前没有重新加密 | 必须用 crypto.encrypt_message 加密 XML |
| 钉钉 Stream 启动后立刻断开 | Client ID 和 Secret 用反了 | 确认 Client ID 是 AppKey,不是 AppSecret |
| 多轮对话串群 | 用了全局 session key | 改用 conversationId 隔离会话 |
| Gemini 偶尔不回复 | 模型返回被安全过滤 | 捕获 block_reason,返回固定文案 |
| 群里 @ 机器人后包含 @ 文本 | DingTalk 消息 content 里有 @昵称 |
用正则去掉 @ 片段再传模型 |
| 服务重启后上下文丢失 | 会话存在内存里 | 生产环境换 Redis,或者接受 REST 场景 |
6.4 一台低配服务器能扛住多少并发
很多人的第一反应是“我要不要上高配机器”。以这个项目为例,企业微信接口是个轻量 FastAPI 服务,钉钉是个长连接客户端,整条链路的瓶颈几乎都压在 Gemini API 的响应延迟上,而不是本地 CPU 或内存。
我在自己的 2 核 4G 云主机上跑过一段时间,企业微信和钉钉两个进程加起来内存占用不到 400MB。日常十人小团队偶尔提问,完全没压力。如果并发真的上去了,优先做两件事:
- 给 Gemini 调用加并发线程池,不要用 FastAPI 的 async 函数直接同步调用模型,否则事件循环会阻塞。
- 把会话上下文从内存挪到 Redis,保证多 worker 之间上下文一致。
我的实际经验是,企业内部这种“团队共用一个机器人”的场景,一般用不到消息队列和微服务。先把日志打清楚,把重复消息拦截住,比堆一堆组件靠谱得多。
最后再分享一个小技巧:调试阶段不要直接在生产群里试,可以单独拉一个测试群,把机器人和你都拉进去,在测试群里把触发方式、回复格式、异常文案都验证一遍。这个项目看起来就是“接个 API”,真正花时间的往往是平台规则和异常边界,把这些处理顺了,后面用起来会非常省心。
