在上一篇里，我们已经实现了一个基于 DeepSeek + 本地向量检索 的企业知识库问答 Demo。
但如果只能在命令行或 Web 页面里用，还离“真正落地”有点远——大部分同事日常待的地方，其实是：

• 钉钉群聊 / 个人聊天
• 企业微信群聊 / 个人聊天

这一篇，我们就做一件事：

把第一篇做好的 DeepSeek RAG 知识库问答服务，接入到 钉钉 / 企业微信 里，让同事在群里 @ 机器人就能问公司内部知识。

---

一、整体思路 & 架构

无论是钉钉还是企业微信，本质都是：

1. 用户在群里 @ 机器人 / 发消息
2. IM 平台把这条消息转发到你配置的服务端接口（HTTP 或 WebSocket）
3. 你的服务解析消息 → 调用 RAG（向量检索 + DeepSeek）→ 生成答案
4. 把答案按平台要求的格式返回给 IM 平台
5. 平台把这条消息以“机器人发言”的形式展示给用户

可以抽象成：

钉钉 / 企业微信
  ↓   （消息推送）
你的后端服务（RAG 网关）
  ↓   （本地向量检索 + DeepSeek API）
DeepSeek RAG 问答引擎（上一篇已经实现）

所以这篇的核心不是再造 RAG，而是实现一个“IM 消息 ↔ RAG 引擎”的中间层。

---

二、准备工作（承接第 1 篇）

默认你已经有了下面这些：

1. 上一篇文章中的代码文件 rag_deepseek_demo.py，里面有：
   • load_knowledge_base
   • VectorDB
   • RAGWithDeepSeek
2. 一个能跑通的 RAG 问答脚本（本地测试没有问题）
3. 已申请好 DeepSeek 的 API Key，并在环境变量中配置：

   export DEEPSEEK_API_KEY="your_real_key"

4. 你有一台可以部署服务的机器（公司内网、云服务器都可以）

本篇我们只在这基础上“加一层 IM 接入网关”。

---

三、把 RAG 封装成一个 HTTP 服务

不论是钉钉还是企业微信，最终都会调用一个 HTTP / HTTPS 接口，所以先把 RAG 封装成一个简单的 HTTP 服务。

这里用 FastAPI 做例子（Flask / Django 都可以自行替换）：

pip install fastapi uvicorn

创建 rag_service.py：

# -*- coding: utf-8 -*-
"""
RAG HTTP 服务：对外暴露一个 /ask 接口
"""
import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

from rag_deepseek_demo import load_knowledge_base, VectorDB, RAGWithDeepSeek

# ====== 初始化 RAG 引擎（与上一节保持一致） ======
KB_PATH = "data/knowledge_base.json"
docs = load_knowledge_base(KB_PATH)
vdb = VectorDB(docs)
vdb.build(n_neighbors=4)
rag = RAGWithDeepSeek(vdb)

class AskRequest(BaseModel):
    user_id: str
    question: str

class AskResponse(BaseModel):
    answer: str
    refs: list

app = FastAPI()

@app.post("/ask", response_model=AskResponse)
def ask(req: AskRequest):
    """
    统一问答接口：
    - user_id：来自钉钉 / 企业微信的用户 ID（方便做限流、权限）
    - question：用户问题
    """
    result = rag.answer(req.question, top_k=3)
    return AskResponse(
        answer=result["answer"],
        refs=[d["title"] for d in result["retrieved_docs"]]
    )

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=9000)

启动：

python rag_service.py

现在你已经有一个统一的 HTTP RAG 问答服务：

POST http://your_host:9000/ask
{
  "user_id": "some_id",
  "question": "DeepSeek-Reasoner 适合做什么？"
}

---

四、接入钉钉：企业内部机器人 + Webhook

钉钉有很多接入方式，这里选 企业内部机器人 + Webhook，优点是实现简单、文档多。

4.1 在钉钉后台创建机器人

1. 打开 https://open.dingtalk.com/

2. 「应用开发 → 企业内部开发 → 机器人 → 创建应用」

3. 填写名称、图标，创建后进入“开发管理”

4. 找到「机器人配置」或「消息推送」，添加一个 自定义机器人 / 自定义关键词机器人

5. 获取一个类似这样的 Webhook 地址：

   https://oapi.dingtalk.com/robot/send?access_token=xxxxxx

6. 在“安全设置”中，建议打开 加签，会生成一个 secret

记住这两个值：

• WEBHOOK_URL
• SECRET（加签密钥）

4.2 编写钉钉机器人转发服务

这里我们做的是“被动通知”的模式：
钉钉群里 @ 机器人 → 你收到消息 → 调用 RAG → 再用 Webhook 把答案发回群里。

安装依赖：

pip install requests

新建 dingtalk_bot.py：

# -*- coding: utf-8 -*-
"""
钉钉群机器人：接收群消息 → 调用 RAG 服务 → 通过 Webhook 回答
"""
import os
import time
import hmac
import hashlib
import base64
import urllib.parse
import requests
from fastapi import FastAPI, Request

# 这些在钉钉机器人后台获取
WEBHOOK_URL = os.getenv("DINGTALK_WEBHOOK", "https://oapi.dingtalk.com/robot/send?access_token=xxxx")
SECRET = os.getenv("DINGTALK_SECRET", "your_secret")

RAG_URL = os.getenv("RAG_URL", "http://127.0.0.1:9000/ask")

app = FastAPI()

def sign():
    """
    生成当前时间戳对应的加签字符串
    """
    timestamp = str(round(time.time() * 1000))
    secret_enc = SECRET.encode('utf-8')
    string_to_sign = '{}\n{}'.format(timestamp, SECRET).encode('utf-8')
    hmac_code = hmac.new(secret_enc, string_to_sign, digestmod=hashlib.sha256).digest()
    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return timestamp, sign

def send_dingtalk_markdown(title: str, text: str):
    """
    通过自定义机器人 Webhook 发送 markdown 消息
    """
    timestamp, sign_str = sign()
    url = f"{WEBHOOK_URL}&timestamp={timestamp}&sign={sign_str}"
    headers = {"Content-Type": "application/json;charset=utf-8"}
    data = {
        "msgtype": "markdown",
        "markdown": {
            "title": title,
            "text": text
        }
    }
    resp = requests.post(url, json=data, headers=headers, timeout=10)
    return resp.json()

@app.post("/dingtalk/callback")
async def dingtalk_callback(req: Request):
    """
    钉钉群里 @机器人 后，配置的“自定义回调地址”会收到消息
    注意：这里假设你已经在钉钉后台配置了这个 URL
    """
    body = await req.json()
    # body 结构参考钉钉文档，这里只处理文本消息
    msg_type = body.get("msgtype")
    if msg_type != "text":
        return {"msg": "ignore non-text"}

    content = body["text"]["content"].strip()
    sender_id = body.get("senderStaffId", "unknown")

    # 把 @机器人 的前缀去掉，只保留真正的问题
    # 例如 "@知识库机器人 DeepSeek-Reasoner 适合做什么？"
    if content.startswith("@"):
        content = content.split(" ", 1)[-1].strip()

    if not content:
        send_dingtalk_markdown("提示", "请直接输入你的问题，例如：\n\n> DeepSeek-Reasoner 适合做什么？")
        return {"msg": "empty"}

    # 调用本地 RAG 服务
    try:
        resp = requests.post(
            RAG_URL,
            json={"user_id": sender_id, "question": content},
            timeout=30
        )
        resp.raise_for_status()
        data = resp.json()
        answer = data["answer"]
        refs = data.get("refs", [])
    except Exception as e:
        send_dingtalk_markdown("系统错误", f"处理你的问题时出错：{e}")
        return {"msg": "error"}

    # 组织 Markdown 答案
    ref_text = ""
    if refs:
        ref_text = "\n\n---\n**引用文档**：\n" + "\n".join([f"- {r}" for r in refs])

    md = f"""**你的问题：​**  
> {content}

**答案：​**  
{answer}{ref_text}
"""
    send_dingtalk_markdown("知识库问答", md)
    return {"msg": "ok"}

启动：

uvicorn dingtalk_bot:app --host 0.0.0.0 --port 9100

然后在钉钉机器人配置里把「自定义回调地址」配置为：

https://你的域名或IP:9100/dingtalk/callback

（内网环境可配反向代理 / 内网穿透，这里不展开）

此时，在群里输入：

@知识库机器人 DeepSeek API 怎么调用？

就会触发我们上面的逻辑 → 调用 RAG → 机器人以 Markdown 形式把答案发回群里。

---

五、接入企业微信：自建应用 + 回调 URL

企业微信这边稍微复杂一点，需要处理 消息签名 & 解密。
为了不让文章被“加解密细节”淹没，这里给一个 简化版思路 + 伪代码结构，你落地时可以直接换成官方 SDK 或社区封装。

5.1 在企业微信后台创建应用

1. 登录 https://work.weixin.qq.com/
2. 「应用管理 → 自建应用 → 创建应用」
3. 记录以下信息：
   • 企业 ID（CorpID）
   • 应用 ID（AgentId）
   • 应用 Secret
4. 在该应用的“接收消息”中：
   • 填写服务器 URL（例如 https://your-domain.com/wecom/callback ）
   • 设置 Token（自定义）
   • 设置 EncodingAESKey（自动生成）
5. 保存后，企业微信会向你的 URL 发送一个验证请求，你需要按文档要求原样返回 echostr

5.2 企业微信 ↔ RAG 中转服务

依赖：

pip install fastapi

新建 wecom_bot.py（把签名/解密部分用占位函数说明，实际用官方 SDK 替换）：

# -*- coding: utf-8 -*-
"""
企业微信自建应用：回调消息 → 调用 RAG → 主动回复
"""
import os
import time
import requests
from fastapi import FastAPI, Request, Response

# 这些从企业微信后台获取
CORP_ID = os.getenv("WECOM_CORP_ID", "your_corp_id")
AGENT_ID = os.getenv("WECOM_AGENT_ID", "1000001")
SECRET = os.getenv("WECOM_SECRET", "your_secret")
TOKEN = os.getenv("WECOM_TOKEN", "your_token")
ENCODING_AES_KEY = os.getenv("WECOM_AES_KEY", "your_encoding_aes_key")

RAG_URL = os.getenv("RAG_URL", "http://127.0.0.1:9000/ask")

app = FastAPI()

# ====== 工具函数（简化版，占位）======

def verify_signature(signature: str, timestamp: str, nonce: str, echostr: str) -> bool:
    """
    校验 URL 签名（简化版思路：按 Token + timestamp + nonce 排序后 SHA1）
    实际使用请参考企业微信官方文档 / SDK
    """
    import hashlib
    tmp_list = [TOKEN, timestamp, nonce]
    tmp_list.sort()
    tmp_str = "".join(tmp_list)
    hashcode = hashlib.sha1(tmp_str.encode("utf-8")).hexdigest()
    return hashcode == signature

def decrypt_message(xml: str) -> dict:
    """
    解密并解析企业微信消息（简化，实际请用官方 SDK）
    返回一个 dict，至少包含：
    - FromUserName（发送人）
    - MsgType（消息类型）
    - Content（文本内容）
    """
    # TODO: 替换成真实解密逻辑
    # 这里可以先用“明文模式”调通流程，再逐步换成加密模式
    import xmltodict
    data = xmltodict.parse(xml)["xml"]
    return data

def build_text_reply(to_user: str, from_user: str, content: str) -> str:
    """
    构造一个最简单的文本回复 XML
    """
    now = int(time.time())
    reply = f"""
<xml>
  <ToUserName><![CDATA[{to_user}]]></ToUserName>
  <FromUserName><![CDATA[{from_user}]]></FromUserName>
  <CreateTime>{now}</CreateTime>
  <MsgType><![CDATA[text]]></MsgType>
  <Content><![CDATA[{content}]]></Content>
</xml>
""".strip()
    return reply

# ====== 回调入口 ======

@app.get("/wecom/callback")
async def wecom_verify(signature: str, timestamp: str, nonce: str, echostr: str):
    """
    企业微信验证 URL 有效性时会发 GET 请求到这里
    """
    if verify_signature(signature, timestamp, nonce, echostr):
        # 正常实现里，echostr 需要用 AES 解密后再返回，这里简化
        return Response(echostr)
    else:
        return Response("invalid signature", status_code=401)

@app.post("/wecom/callback")
async def wecom_callback(request: Request):
    """
    接收用户消息
    """
    xml = await request.body()
    msg = decrypt_message(xml.decode("utf-8"))

    msg_type = msg.get("MsgType")
    if msg_type != "text":
        return Response("success")  # 非文本消息先忽略

    user_id = msg.get("FromUserName")
    content = msg.get("Content", "").strip()

    # 可选：去掉“@机器人”的前缀
    if content.startswith("@"):
        content = content.split(" ", 1)[-1].strip()

    if not content:
        return Response("success")

    # 调用 RAG 服务
    try:
        resp = requests.post(
            RAG_URL,
            json={"user_id": user_id, "question": content},
            timeout=30
        )
        resp.raise_for_status()
        data = resp.json()
        answer = data["answer"]
        refs = data.get("refs", [])
    except Exception as e:
        answer = f"处理你的问题时出错：{e}"
        refs = []

    ref_text = ""
    if refs:
        ref_text = "\n\n---\n引用文档：\n" + "\n".join([f"- {r}" for r in refs])

    full_answer = f"你的问题：{content}\n\n{answer}{ref_text}"

    reply_xml = build_text_reply(
        to_user=msg["FromUserName"],
        from_user=msg["ToUserName"],
        content=full_answer
    )

    # 实际加密模式下，需要对 reply_xml 再包一层加密，这里先返回明文流程
    return Response(reply_xml, media_type="application/xml")

启动：

pip install xmltodict uvicorn wecom_bot:app --host 0.0.0.0 --port 9200

然后在企业微信后台，把回调 URL 配置为：

https://你的域名或IP:9200/wecom/callback

先用“明文模式”跑通（安全模式 / 兼容模式的加解密，用官方 SDK 即可对照替换）。

---

六、几个实战细节建议

1. 问题前缀规范

为了方便统计和后期扩展，你可以约定：

• “#产品 开头的问题走产品知识库”
• “#技术 开头的问题走技术知识库”
• “#人事 开头的问题走 HR 知识库”

在 RAG 侧可以维护多套 VectorDB，根据前缀路由：

if content.startswith("#产品"):
    engine = product_rag
    question = content.replace("#产品", "").strip()
elif content.startswith("#技术"):
    engine = tech_rag
    question = content.replace("#技术", "").strip()
else:
    engine = default_rag
    question = content

2. 权限与敏感信息控制

• 通过 user_id / senderStaffId 查询用户部门（调用钉钉 / 企业微信通讯录接口）
• 为不同部门加载不同的知识库索引
• 对包含关键词（如“薪资”“财务报表”“密码”）的问题直接拦截或转人工

3. 限流与稳定性

• 钉钉自定义机器人有发送频率限制：每分钟最多 20 条，企业微信也有 QPS 限制
• 可以在 IM 网关层做：
  • 单用户 QPS 限流（例如每分钟最多 10 次）
  • RAG 调用失败时，返回“系统繁忙，请稍后再试”而不是把异常抛给用户

4. 日志与埋点

建议至少记录：

• platform（dingtalk / wecom）
• user_id
• question
• answer 片段（可脱敏）
• 命中的文档 ID / 标题
• DeepSeek 调用耗时、Token 用量（用于成本核算）

后续可以：

• 统计 FAQ
• 优化知识库内容和向量模型
• 针对常见问题提前缓存答案，减轻模型压力

---

七、从 Demo 到企业应用：一个可行的路线图

1. 第 0 步：本地 RAG Demo 跑通
   完成第一篇中的向量检索 + DeepSeek 问答脚本。

2. 第 1 步：抽象出统一的 /ask HTTP 接口
   用 FastAPI / Flask 封一层，让任何客户端都能通过 HTTP 调用 RAG。

3. 第 2 步：先接入一个平台（建议钉钉 Webhook）​

   • 用上面 dingtalk_bot.py 的逻辑
   • 在一个测试群做试用，看问题形式、效果、延时

4. 第 3 步：再接入企业微信

   • 先用明文模式（方便调试）
   • 基本流程稳定后，再切到安全模式 + SDK 加解密

5. 第 4 步：做企业级增强

   • 多知识库路由、权限控制、日志分析、缓存、监控
   • 往“智能客服 / 智能助理”方向演进

6. 第 5 步：产品化 / 平台化

   • 管理后台：配置知识库、查看问答日志、手动标注问题
   • 多租户：支持多个 BU / 子公司共用一套系统

---

八、总结

这一篇，我们把上一集的 DeepSeek RAG 知识库 Demo，真正“搬进了工作 IM”：

• 用一个统一的 /ask HTTP 接口，把 本地向量检索 + DeepSeek 封装成服务
• 通过 钉钉自定义机器人 + Webhook 实现“群里 @机器人 → 知识库问答”
• 通过 企业微信自建应用 + 回调 URL 实现类似能力
• 给出了可直接改造的核心代码结构，重点放在“消息转发 → 调 RAG → 格式化回复”的完整链路

你可以直接按文中的结构把代码拆成三层：

1. RAG 内核层：向量检索 + DeepSeek 调用（上一篇已有）
2. RAG HTTP 服务层：统一的 /ask 接口
3. IM 适配层：钉钉 / 企业微信 / 甚至飞书、Slack 的适配器