大模型实战：从 FAQ 到多轮对话的 DeepSeek 企业智能助手

在前两篇里，我们已经完成了两件事：

但这个机器人目前还是典型的：

“单轮 FAQ 问答 + 搜索式使用体验”

现实企业场景里，你更需要的是一个能：

• 记住上下文，理解“那上个月的呢？”这种指代；
• 识别意图：是查政策、查订单、还是要帮忙发起流程；
• 能主动调企业内部系统（工单、审批、订单、HR 等）完成任务；

的 多轮对话企业智能助手（Enterprise Assistant）​。

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

在前两篇 RAG+IM 的基础上，加上 对话状态管理 + 多轮对话 + Function Calling 工具调用， 从“FAQ 机器人”升级为“企业智能助手”。

一、单轮 FAQ 和 多轮企业助手，本质差别在哪？

可以用一个简单对比表来理解：

要实现右边这列，你至少需要三块基础设施：

下面我们一步步落地。

二、整体架构：在原有 RAG + IM 上加一层“对话与工具”

你现在已有架构（来自第②篇）大致是：

钉钉 / 企业微信
↓
IM 网关（FastAPI）
↓ HTTP /ask
RAG 问答服务（向量检索 + DeepSeek）

这一篇在中间加一层：

钉钉 / 企业微信
↓
IM 网关（FastAPI）
↓
对话与工具层（本篇新增：Conversation + Tools）
↓
RAG 内核（上一篇已有：VectorDB + DeepSeek）

对话与工具层负责：

• 给每个用户 / 会话一个独立的 Conversation（上下文记忆）；
• 决定当前轮要不要用工具（Function Calling）；
• 是否需要用 RAG 检索知识；
• 把所有东西“打包成 Prompt + messages”交给 DeepSeek。

三、对话状态管理：给每个用户一个 Conversation

先定义一个最简单的对话管理类（你可以放到 conversation.py）：

# conversation.py
from typing import Dict, List, Any
import uuid
import time

class Conversation:
"""管理单个对话的上下文状态"""
def __init__(self, conversation_id: str = None):
self.conversation_id = conversation_id or str(uuid.uuid4())
self.messages: List[Dict] = []
self.start_time = time.time()
self.last_active = time.time()
self.metadata = {} # 可以放 user 部门、权限等

def add_message(self, role: str, content: str, **kwargs):
"""添加一条消息"""
self.messages.append({
"role": role, # "system" | "user" | "assistant" | "tool"
"content": content,
"timestamp": time.time(),
**kwargs
})
self.last_active = time.time()

def get_messages(self, max_turns=10) -> List[Dict]:
"""获取最近若干轮对话（防止上下文过长）"""
return self.messages[-max_turns:]

def clear(self):
self.messages = []

def update_metadata(self, key: str, value: Any):
self.metadata[key] = value

然后，在 IM 网关里用一个简单的“会话表”管理它们（Demo 可以用字典，生产用 Redis）：

# conversations_pool.py
from typing import Dict
from conversation import Conversation

_conversations: Dict[str, Conversation] = {}

def get_conversation(session_id: str) -> Conversation:
"""按用户ID或(群ID+用户ID)获取对话"""
if session_id not in _conversations:
_conversations[session_id] = Conversation(session_id)
return _conversations[session_id]

在钉钉/企微回调里，就可以通过 senderId（或群ID + senderId）拿到会话了。

四、企业工具系统：Function Calling 的“手脚”

DeepSeek 支持类似 OpenAI 的 Function Calling，也就是：

先做一个简单版工具系统（tools.py）：

# tools.py
from typing import Dict, Any, List

class Tool:
"""工具基类"""
def __init__(self, name: str, description: str):
self.name = name
self.description = description

def schema(self) -> Dict[str, Any]:
"""返回 DeepSeek 兼容的 function schema"""
raise NotImplementedError

def call(self, arguments: Dict[str, Any]) -> Any:
"""执行工具调用"""
raise NotImplementedError

示例 1：HR 政策查询工具（真实环境可以对接 HR 系统 API）：

class HRPolicyTool(Tool):
"""人力资源政策查询工具"""
def __init__(self):
super().__init__("query_hr_policy", "查询公司人力资源政策")

def schema(self):
return {
"type": "function",
"function": {
"name": "query_hr_policy",
"description": "查询公司人力资源政策，如休假、报销、入职等",
"parameters": {
"type": "object",
"properties": {
"category": {
"type": "string",
"description": "政策类别: 休假、报销、入职、福利",
"enum": ["休假", "报销", "入职", "福利"],
"required": True
}
},
"required": ["category"]
}
}
}

def call(self, arguments: Dict[str, Any]):
category = arguments.get("category", "")
policies = {
"休假": {"policy": "年假、病假、调休、事假", "details": "员工每年享有5-15天带薪年假，具体根据工龄确定"},
"报销": {"policy": "差旅费报销", "details": "交通、住宿、餐饮等费用，需提供发票"},
"入职": {"policy": "入职流程", "details": "新员工需提交身份证、学历证明等材料"},
"福利": {"policy": "员工福利", "details": "包括五险一金、年度体检、节日福利"}
}
return policies.get(category, {"error": "未知的政策类别"})

示例 2：订单状态查询工具：

class OrderStatusTool(Tool):
"""订单状态查询工具"""
def __init__(self):
super().__init__("query_order_status", "查询订单状态和物流信息")

def schema(self):
return {
"type": "function",
"function": {
"name": "query_order_status",
"description": "查询订单状态和物流信息",
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "订单编号",
"required": True
}
},
"required": ["order_id"]
}
}
}

def call(self, arguments: Dict[str, Any]):
order_id = arguments.get("order_id", "")
# 真实环境这里调用订单系统
return {"order_id": order_id, "status": "已发货", "tracking_no": f"SF{order_id}20250201"}

工具管理器：

class ToolManager:
"""工具管理器"""
def __init__(self):
self.tools: Dict[str, Tool] = {}

def register(self, tool: Tool):
self.tools[tool.name] = tool

def get_schema(self) -> List[Dict[str, Any]]:
"""返回所有工具的 schema"""
return [t.schema() for t in self.tools.values()]

def get_tool(self, name: str) -> Tool:
return self.tools.get(name)

五、把这些粘在一起：EnterpriseAssistant 多轮助手

在第①篇你已经有了：

• VectorDB：向量检索
• RAGWithDeepSeek：RAG 调用逻辑
• call_deepseek：调用 DeepSeek 的封装

这一篇，我们包一层 EnterpriseAssistant，专门面向“一个会话的多轮对话”。

5.1 初始化助手

# assistant.py
from typing import Dict, Any, List
from conversation import Conversation
from tools import ToolManager, HRPolicyTool, OrderStatusTool
from rag_deepseek_demo import load_knowledge_base, VectorDB, RAGWithDeepSeek

class EnterpriseAssistant:
"""企业智能助手：多轮对话 + 工具调用 + RAG"""

def __init__(self, conversation_id: str):
self.conversation = Conversation(conversation_id)
self.tool_manager = ToolManager()
self.tool_manager.register(HRPolicyTool())
self.tool_manager.register(OrderStatusTool())

# 复用第①篇的 RAG 逻辑
docs = load_knowledge_base("data/knowledge_base.json")
vdb = VectorDB(docs)
vdb.build(n_neighbors=4)
self.rag_engine = RAGWithDeepSeek(vdb)

# 系统提示（只加一次）
system_prompt = (
"你是一个企业智能助手，负责回答员工问题。\\n"
"- 如果是关于公司政策、产品、技术文档的问题，使用知识库回答\\n"
"- 如果需要查询具体业务数据（如订单、审批），使用工具\\n"
"- 对话要友好、专业，每次回答不超过200字\\n"
"- 如果缺少必要信息，要先向用户澄清再调用工具\\n"
)
self.conversation.add_message("system", system_prompt)

5.2 构造多轮 + 知识库的 Prompt

def _build_prompt(self, question: str, docs: List[Dict]) -> str:
"""把最近几轮对话 + 检索到的知识库片段拼成一个 Prompt"""
history = self.conversation.get_messages(max_turns=6)
history_text = "\\n".join(
f"{m['role'].upper()}: {m['content']}"
for m in history if m["content"]
)
ctx_text = "\\n\\n".join(
f"【{d['title']}】\\n{d['content']}"
for d in docs
)

return f"""你是一个企业知识库和业务助手，请结合“对话历史”和“知识库内容”回答问题。

[对话历史]
{history_text}

[知识库文档]
{ctx_text}

[当前用户问题]
{question}

回答要求：
1. 优先基于知识库与工具结果，不要瞎编；
2. 不确定时请明确说“现有信息不足，建议…”；
3. 控制在200字以内。
"""

5.3 调用 DeepSeek + 处理工具调用

这里的伪代码基于 DeepSeek 官方 Function Calling 文档的结构（和 OpenAI 非常类似）：

import json

def _call_with_tools(self, messages: List[Dict]) -> Any:
"""调用 DeepSeek，并处理一次工具调用"""
resp = self.rag_engine.client.chat.completions.create(
model=self.rag_engine.model,
messages=messages,
tools=self.tool_manager.get_schema(),
temperature=0.2
)
msg = resp.choices[0].message

# 检查是否有 tool_calls
if hasattr(msg, "tool_calls") and msg.tool_calls:
tool_call = msg.tool_calls[0]
tool_name = tool_call.function.name
# 注意：生产中要用 json.loads + 异常捕获
tool_args = json.loads(tool_call.function.arguments)

# 记录这次“调用工具”的行为到对话里（方便多轮）
self.conversation.add_message(
"assistant",
None,
tool_calls=[{
"id": tool_call.id,
"type": "function",
"function": {
"name": tool_name,
"arguments": tool_call.function.arguments
}
}]
)

tool = self.tool_manager.get_tool(tool_name)
if not tool:
return {"error": f"未知工具 {tool_name}"}

result = tool.call(tool_args)
return {
"tool_name": tool_name,
"tool_args": tool_args,
"tool_result": result
}

# 如果模型没要求调用工具，就返回 None
return None

5.4 综合回答逻辑：优先工具 + RAG

def answer(self, question: str) -> str:
"""完整多轮回答逻辑"""
# 1. 记录用户发言
self.conversation.add_message("user", question)

# 2. 先用 RAG 找知识库上下文
retrieved = self.rag_engine.vector_db.search(question, top_k=3)
prompt = self._build_prompt(question, retrieved)
base_messages = self.conversation.get_messages()

# 3. 第一次调用：让模型决定是否需要工具
tool_ctx = self._call_with_tools(base_messages + [
{"role": "user", "content": prompt}
])

# 4. 如果触发了工具调用，再做一次“总结”
if tool_ctx and not tool_ctx.get("error"):
final_prompt = f"""
用户问题：{question}
工具名称：{tool_ctx['tool_name']}
工具参数：{json.dumps(tool_ctx['tool_args'], ensure_ascii=False)}
工具返回结果：{json.dumps(tool_ctx['tool_result'], ensure_ascii=False)}

请基于工具结果+对话历史，用自然语言给用户一个清晰答案。
"""
resp = self.rag_engine.client.chat.completions.create(
model=self.rag_engine.model,
messages=base_messages + [
{"role": "assistant", "content": final_prompt}
],
temperature=0.2
)
answer = resp.choices[0].message.content.strip()
else:
# 5. 不需要工具时，用 RAG+Prompt 直接生成答案
resp = self.rag_engine.client.chat.completions.create(
model=self.rag_engine.model,
messages=base_messages + [
{"role": "user", "content": prompt}
],
temperature=0.2
)
answer = resp.choices[0].message.content.strip()

# 6. 记录模型回复，更新对话
self.conversation.add_message("assistant", answer)
return answer

六、和钉钉 / 企业微信组合：从“问一次答一次”到“持续对话”

假设你在第②篇里已经有 dingtalk_bot.py，核心逻辑里从原来的“每次直接调 /ask”改成“拿到一个长期存在的 EnterpriseAssistant”。

示意（仅展示关键变化）：

# dingtalk_bot.py 中的一部分
from fastapi import FastAPI, Request
from assistant import EnterpriseAssistant
from conversations_pool import get_conversation # 这里返回的是 Conversation 或 Assistant

app = FastAPI()
assistants = {} # 简单内存版，建议用 Redis

def get_assistant(session_id: str) -> EnterpriseAssistant:
if session_id not in assistants:
assistants[session_id] = EnterpriseAssistant(session_id)
return assistants[session_id]

@app.post("/dingtalk/callback")
async def dingtalk_callback(req: Request):
body = await req.json()
msg_type = body.get("msgtype")
if msg_type != "text":
return {"msg": "ignore"}

content = body["text"]["content"].strip()
sender_id = body.get("senderStaffId", "unknown")
# 也可以用 群id+sender_id 做 session key
session_id = sender_id

# 去掉@机器人
if content.startswith("@"):
content = content.split(" ", 1)[-1].strip()

if not content:
# 返回一个提示
…
return {"msg": "empty"}

assistant = get_assistant(session_id)

try:
answer = assistant.answer(content)
except Exception as e:
answer = f"处理你的问题时出错：{e}"

# 通过钉钉 Webhook 把 answer 发回群里（参考第②篇代码）
…
return {"msg": "ok"}

效果：​

• 同一个 sender_id 连续提问时，会共享同一个 EnterpriseAssistant → 对话有上下文；
• RAG 知识库负责“说明书类问题”，Function Calling 工具负责“查数据/办事情”。

七、实战对话示例：从 FAQ 到多轮任务

示例 1：从政策说明到个性化查询

第 1 轮

用户：公司年假怎么计算？

• 触发 RAG，命中“企业请假制度”文档；
• 助手回答年假规则等（知识说明）。

第 2 轮（多轮追问）​

用户：那我现在还有几天年假？

助手多步操作：

对于用户而言，它没感知“我现在在走工具调用”，只是“我问的问题更具体了”。

示例 2：订单场景的多轮澄清

用户：帮我查一下上次给小米的那个大订单发货没？

助手可以这样多轮引导：

这里的多轮对话 + 工具调用组合，已经接近“半自动客服/商务助理”。

八、落地注意事项与优化建议

九、写在最后：系列路线回顾与下一步

到目前为止，这个系列已经完整走完一条非常典型的企业落地路径：