第十章：Worker 接入 DeepSeek（AI核心引擎）
这一章开始，系统正式从：
“能处理消息的架构系统”
升级为：
“真正具备智能回复能力的 AI 系统”
10.1 当前系统状态回顾
在上一章结束时，我们已经有：
企业微信
↓
FastAPI（Webhook）
↓
Dispatcher
↓
Redis Queue
↓
Worker（但只是打印 + 假处理）
但此时 Worker 还没有“脑子”。
❌ 当前 Worker 的问题
只能：
✔ 接收消息
✔ 打印内容
不能：
❌ 理解语义
❌ 生成回复
❌ 具备AI能力
10.2 本章目标
本章完成后：
Worker
↓
DeepSeek API
↓
Prompt处理
↓
生成回复
↓
返回企业微信
10.3 AI调用架构设计
📌 新增核心模块
app/core/
├── llm.py
├── prompt.py
└── client.py
10.4 DeepSeek API设计
我们统一封装LLM调用层：
未来可以无缝切换 OpenAI / Qwen / Claude
📌 llm.py（核心封装）
import os
import httpx
class LLMClient:
def __init__(self):
self.api_key = os.getenv("DEEPSEEK_API_KEY")
self.base_url = "https://api.deepseek.com/v1/chat/completions"
async def chat(self, prompt: str):
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-chat",
"messages": [
{"role": "system", "content": "你是一个企业AI助手"},
{"role": "user", "content": prompt}
],
"temperature": 0.7
}
async with httpx.AsyncClient(timeout=30) as client:
resp = await client.post(
self.base_url,
json=payload,
headers=headers
)
data = resp.json()
return data["choices"][0]["message"]["content"]
10.5 Prompt工程（第一版）
📌 prompt.py
def build_prompt(user_input: str) -> str:
"""
构建企业AI Prompt
"""
return f"""
你是企业微信中的AI助手。
请用简洁、专业的方式回答用户问题。
用户问题：
{user_input}
"""
10.6 Worker升级（接入AI）
📌 worker.py（核心升级版）
import json
import asyncio
from app.queue.redis_client import redis_client
from app.core.llm import LLMClient
from app.core.prompt import build_prompt
from app.services.wecom_service import send_message
QUEUE_KEY = "queue:wecom"
llm = LLMClient()
async def process(message: dict):
user = message.get("FromUserName")
content = message.get("Content")
print("AI处理:", content)
prompt = build_prompt(content)
reply = await llm.chat(prompt)
send_message(user, reply)
def run_worker():
print("Worker started...")
while True:
_, data = redis_client.brpop(QUEUE_KEY)
message = json.loads(data)
asyncio.run(process(message))
10.7 AI调用流程拆解
用户消息
↓
Worker取出
↓
Prompt构建
↓
DeepSeek请求
↓
返回AI结果
↓
发送企业微信
10.8 企业微信发送接口（补齐）
📌 wecom_service.py
import os
import httpx
class WeComService:
def __init__(self):
self.corp_id = os.getenv("WECOM_CORP_ID")
self.agent_id = os.getenv("WECOM_AGENT_ID")
self.secret = os.getenv("WECOM_SECRET")
self.access_token = None
async def get_token(self):
url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken"
params = {
"corpid": self.corp_id,
"corpsecret": self.secret
}
async with httpx.AsyncClient() as client:
resp = await client.get(url, params=params)
data = resp.json()
return data["access_token"]
async def send_message(self, user: str, content: str):
token = await self.get_token()
url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"
payload = {
"touser": user,
"msgtype": "text",
"agentid": self.agent_id,
"text": {
"content": content
}
}
async with httpx.AsyncClient() as client:
await client.post(url, json=payload)
10.9 Worker完整链路
Redis队列
↓
Worker消费
↓
Prompt构建
↓
DeepSeek调用
↓
AI返回
↓
企业微信发送
10.10 当前系统能力
✅ 已完成能力
✔ 企业微信接入
✔ 消息解密
✔ Dispatcher分发
✔ Redis异步队列
✔ Worker消费模型
✔ DeepSeek接入
✔ AI回复生成
✔ 企业微信发送API
10.11 系统第一次“智能化”
这一章完成后，你的系统已经具备：
👉 真正AI对话能力
表现为：
用户：你好
系统：你好，我是企业AI助手，很高兴为你服务
10.12 架构升级总结
企业微信
↓
Webhook（FastAPI）
↓
Dispatcher
↓
Redis Queue
↓
Worker
↓
DeepSeek
↓
WeCom Send API
↓
企业微信回复
10.13 本章核心价值
这一章完成后：
👉 系统从“消息处理系统”
升级为
👉 “AI驱动的企业机器人系统”
🚀 下一章预告（第十一章）
第十一章：AccessToken管理与企业微信稳定发送机制
下一章我们将解决一个生产级问题：
❌ access_token 每次请求都重新获取
❌ 接口不稳定
❌ 容易触发限流
下一章核心内容：
✔ Token缓存
✔ Redis存储
✔ 自动刷新机制
✔ 并发安全
✔ 企业微信稳定发送架构
👉 到这一章结束：
系统才真正进入“生产级稳定状态”。