最近在捣鼓智能客服系统，发现直接调用云服务虽然省事，但延迟、隐私和成本这几个老问题总是绕不开。尤其是想对接像B站这样的平台做点个性化服务时，云服务的灵活度常常不够用。于是，我把目光投向了本地化部署，用 ollama 这套工具链折腾了一番，效果还不错。今天就把从模型部署到和B站集成的整个实战过程梳理一下，给有类似想法的朋友做个参考。

智能客服系统概念图

1. 为什么选择本地部署？聊聊背景与痛点

最开始我也考虑过直接使用各大厂商提供的云智能客服API，但深入使用后，几个痛点越来越明显：

  • 响应延迟不可控：请求需要经过公网到达厂商服务器，处理后再返回，网络波动会直接影响用户体验，高峰期延迟尤其明显。
  • 数据隐私顾虑：用户的咨询对话内容，特别是涉及一些内部业务信息时，发送到第三方云平台总让人不那么放心。
  • 长期成本高昂：云服务通常按调用量或时间计费，随着咨询量增长，成本是线性上升的，对于想长期运营的项目来说是个不小的负担。
  • 定制化能力弱：云服务的模型和功能相对固定，想要针对特定领域（比如B站的二次元社区、游戏术语）做深度优化或集成特定业务逻辑，非常困难。

正是这些原因，促使我去研究本地化部署的方案。把大模型“请”到自己的服务器上，虽然前期部署麻烦点，但换来的是对性能、数据、成本的完全掌控。

2. 技术选型：为什么是Ollama？

决定本地部署后，面临几个主流选择：直接用 Transformers 库写推理脚本、用 TensorFlow Serving 或 Triton Inference Server 做服务化，或者用 Ollama。

这里简单对比一下：

  • Transformers 直接推理：最灵活，但需要自己处理模型加载、服务封装、并发请求等，工作量大，不适合快速生产部署。
  • TensorFlow Serving / Triton：工业级服务化框架，性能强大，支持多模型、版本管理，但配置复杂，学习曲线陡峭。
  • Ollama：它的定位就是简化大模型的本地运行与管理。一条命令就能拉取和运行模型，内置了简单的API服务器。对于快速原型验证和中小型应用来说，开箱即用的特性极具吸引力。

对于我们构建智能客服这个场景，核心需求是快速搭起来、能稳定提供服务、并且方便管理模型。Ollama 的轻量化和易用性正好切中要害。它负责了最复杂的模型运行环境管理，让我们可以更专注于业务逻辑和系统集成。

3. 核心实现三步走

3.1 第一步：Ollama 模型选型与部署

Ollama 支持很多模型，对于智能客服，我们需要一个在中文上表现良好、响应速度快的模型。参数量太大的模型对硬件要求高，响应慢；太小的模型智能程度可能不够。7B（70亿）参数量级的模型是一个比较好的平衡点。

这里我推荐 qwen2.5:7b 或 llama3.2:3b（虽然它叫3B，但最新版本能力很强）。通义千问和Llama的中文支持都不错。

部署只需要两行命令：

# 拉取模型（如果本地没有）
ollama pull qwen2.5:7b
# 运行模型服务，默认监听11434端口
ollama run qwen2.5:7b

运行后，Ollama 会启动一个本地服务，提供简单的API。但为了集成到我们的系统，我们需要一个更规范、功能更全的API层。

3.2 第二步：用 FastAPI 构建业务API层

Ollama 的原生API比较简单，我们需要用 FastAPI 包装一层，加入鉴权、日志、格式化输入输出等功能。

首先，安装依赖：

pip install fastapi uvicorn httpx python-jose[cryptography] passlib[bcrypt] redis

然后，创建 main.py：

from fastapi import FastAPI, HTTPException, Depends, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from typing import List, Optional
import httpx
import logging
from jose import JWTError, jwt
from datetime import datetime, timedelta
import redis.asyncio as redis

# --- 配置 ---
OLLAMA_BASE_URL = "http://localhost:11434"
MODEL_NAME = "qwen2.5:7b"
SECRET_KEY = "your-secret-key-here-change-in-production"  # 生产环境务必更换！
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Redis连接（用于对话历史缓存，后续会用到）
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

app = FastAPI(title="智能客服API")
security = HTTPBearer()

# --- 数据模型 ---
class ChatMessage(BaseModel):
    role: str  # user, assistant
    content: str

class ChatRequest(BaseModel):
    messages: List[ChatMessage]
    session_id: Optional[str] = None  # 用于关联对话历史
    max_tokens: Optional[int] = 512

class ChatResponse(BaseModel):
    reply: str
    session_id: Optional[str] = None

# --- 工具函数 ---
def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        # 这里可以添加更复杂的权限检查，例如从payload中取出username
        return payload
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效或过期的令牌",
            headers={"WWW-Authenticate": "Bearer"},
        )

# --- 核心聊天端点 ---
@app.post("/v1/chat", response_model=ChatResponse)
async def chat_with_model(
    request: ChatRequest,
    token_payload: dict = Depends(verify_token)  # 依赖注入，实现鉴权
):
    """
    与Ollama模型对话的核心接口。
    """
    # 1. 准备发送给Ollama的请求体
    ollama_payload = {
        "model": MODEL_NAME,
        "messages": [msg.dict() for msg in request.messages],
        "stream": False,
        "options": {"num_predict": request.max_tokens}
    }

    # 2. 调用Ollama API
    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            resp = await client.post(
                f"{OLLAMA_BASE_URL}/api/chat",
                json=ollama_payload
            )
            resp.raise_for_status()
            ollama_result = resp.json()
        except httpx.RequestError as e:
            logging.error(f"调用Ollama服务失败: {e}")
            raise HTTPException(status_code=502, detail="后端模型服务暂时不可用")
        except httpx.HTTPStatusError as e:
            logging.error(f"Ollama服务返回错误: {e.response.status_code} - {e.response.text}")
            raise HTTPException(status_code=500, detail="模型处理请求时出错")

    # 3. 提取回复
    reply_content = ollama_result.get("message", {}).get("content", "").strip()
    if not reply_content:
        reply_content = "抱歉，我暂时无法处理这个问题。"

    # 4. （可选）保存当前对话到Redis，用于后续多轮对话
    if request.session_id:
        # 简化的历史存储：将本轮问答追加到历史记录
        history_key = f"chat_history:{request.session_id}"
        # 这里仅作示例，实际应结构化存储
        await redis_client.rpush(history_key, f"User: {request.messages[-1].content}")
        await redis_client.rpush(history_key, f"Assistant: {reply_content}")
        # 设置过期时间，例如1天
        await redis_client.expire(history_key, 86400)

    return ChatResponse(reply=reply_content, session_id=request.session_id)

# --- 测试端点（无需鉴权） ---
@app.get("/health")
async def health_check():
    return {"status": "healthy", "model": MODEL_NAME}

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

这个API提供了 /v1/chat 作为聊天入口，并用JWT进行保护。你可以用下面的命令测试：

# 1. 获取一个测试Token (实际应由独立的登录接口签发)
# 这里简单演示，生产环境请勿将密钥硬编码在客户端
python -c "from jose import jwt; print(jwt.encode({'sub':'test_user'}, 'your-secret-key-here-change-in-production', algorithm='HS256'))"

# 2. 调用聊天接口
curl -X POST http://localhost:8000/v1/chat \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <上一步得到的Token>" \
  -d '{
    "messages": [{"role": "user", "content": "你好，请介绍一下你自己。"}],
    "session_id": "test_session_123"
  }'

预期返回：

{
  "reply": "你好！我是一个基于Qwen2.5模型构建的智能助手，致力于为你提供有用的信息和帮助。请问有什么可以为你效劳的吗？",
  "session_id": "test_session_123"
}
3.3 第三步：对接B站开放平台Webhook

B站开放平台允许开发者创建应用，并通过 Webhook 接收用户私信、评论@等事件。这是我们将智能客服接入B站的关键。

主要流程如下：

  1. 创建应用：在B站开放平台创建一个机器人类型的应用，获取 client_id 和 client_secret。
  2. 配置Webhook地址：在应用配置中，设置一个HTTPS URL作为Webhook接收地址。这个地址就是我们的FastAPI服务暴露的公网地址（需要内网穿透或部署到云服务器）。
  3. 验证请求：B站发送的Webhook请求会携带签名，我们需要验证签名以确保请求来源合法。
  4. 处理事件：解析Webhook JSON数据，提取用户消息和发送者信息。
  5. 调用智能客服API：将用户消息构造成 ChatRequest，调用我们上面写的 /v1/chat 接口。
  6. 回复用户：将得到的回复，通过B站开放平台提供的 “发送私信”API 回给用户。

这里给出一个处理Webhook的FastAPI端点示例：

from fastapi import Request, Header
import hashlib
import hmac

WEBHOOK_SECRET = "your-bilibili-webhook-secret"  # 在B站开放平台配置的密钥

@app.post("/bilibili/webhook")
async def handle_bilibili_webhook(
    request: Request,
    x_bili_signature: Optional[str] = Header(None),  # B站发送的签名头
    x_bili_timestamp: Optional[str] = Header(None)   # 时间戳头
):
    # 1. 验证签名
    body_bytes = await request.body()
    if not verify_bilibili_signature(body_bytes, x_bili_signature, x_bili_timestamp):
        raise HTTPException(status_code=403, detail="签名验证失败")

    # 2. 解析事件
    event_data = await request.json()
    event_type = event_data.get("type")
    # 这里以接收私信为例
    if event_type == "im.message.receive_v1":
        message_data = event_data.get("data")
        sender_uid = message_data.get("sender_uid")
        message_content = message_data.get("message", {}).get("content", "") # 实际内容需要进一步解析

        # 3. 调用我们的智能客服API
        chat_request = ChatRequest(
            messages=[ChatMessage(role="user", content=message_content)],
            session_id=f"bili_{sender_uid}"  # 用B站UID作为会话ID
        )
        # 这里需要模拟一个已认证的请求，或者内部直接调用函数
        # 简单起见，我们复用之前的逻辑，但需要处理认证绕过（内部调用）
        chat_response = await chat_with_model_internal(chat_request)

        # 4. 调用B站API回复用户 (此处省略具体HTTP请求代码)
        # await send_bilibili_private_message(sender_uid, chat_response.reply)

        return {"status": "ok", "msg": "processed"}
    return {"status": "ok", "msg": "event ignored"}

def verify_bilibili_signature(body: bytes, signature: str, timestamp: str) -> bool:
    """验证B站Webhook签名"""
    if not signature or not timestamp:
        return False
    string_to_sign = f"{timestamp}.{WEBHOOK_SECRET}.{body.decode()}"
    expected_sig = hmac.new(
        WEBHOOK_SECRET.encode(),
        string_to_sign.encode(),
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected_sig, signature)

# 一个内部调用的聊天函数，绕过JWT验证
async def chat_with_model_internal(request: ChatRequest) -> ChatResponse:
    # 这里可以复用上面chat_with_model函数的核心逻辑，但去掉token验证部分
    # 直接调用Ollama并返回结果
    pass

API对接流程示意图

4. 性能优化：让响应更快更稳

本地模型推理速度是关键。Ollama 本身已经做了优化，但我们还可以更进一步。

4.1 使用 vLLM 加速推理（高级选项）

Ollama 默认的推理后端在某些情况下可能不是最快的。如果你追求极致的吞吐量（TPS），可以考虑用 vLLM 这种高性能推理引擎来部署模型，然后让 Ollama 或你自己的 FastAPI 服务去调用它。

  1. 安装 vLLM：pip install vllm
  2. 使用 vLLM 启动模型服务： 
    vllm serve qwen2.5-7b-instruct --max-model-len 4096 --port 8001
    这会在8001端口启动一个兼容OpenAI API格式的服务。
  3. 修改 FastAPI 代码：将上面代码中调用 OLLAMA_BASE_URL 的部分，改为指向 http://localhost:8001/v1，并将请求体格式调整为 OpenAI 兼容格式（/v1/chat/completions）。vLLM 的PagedAttention等技术能显著提升并发下的推理速度。
4.2. 使用 Redis 实现对话历史缓存

智能客服需要上下文记忆。上面的代码已经简单演示了用 Redis 存储历史。一个更完善的方案是：

  • 结构化存储：不要简单用列表存字符串。可以使用 Hash 存储会话元信息，用 List 或 Sorted Set 存储有序的消息链。
  • 上下文窗口管理：模型有token限制（如4096）。当历史对话太长时，需要一种策略来裁剪或总结旧历史。一种简单策略是只保留最近N轮对话。
  • 向量缓存：对于常见问题（FAQ），可以将问题和标准答案向量化存入 Redis（配合向量数据库更佳），用户提问时先进行向量相似度检索，如果匹配度高直接返回缓存答案，避免调用大模型，极大降低响应时间和负载。

5. 避坑指南与进阶思考

5.1 中文语料微调注意事项

如果你发现模型对特定领域（如B站梗、游戏术语）理解不佳，可能需要微调。

  • 数据准备：收集高质量的客服对话数据（用户问-标准答），注意清洗和去噪。
  • 格式对齐：将数据转换成与模型预训练时一致的格式（如 [INST] 问题 [/INST] 答案 对于 Llama）。
  • 工具选择：可以使用 unsloth、Axolotl 或 LLaMA-Factory 等微调框架，它们对消费级显卡更友好。
  • 防止灾难性遗忘：微调时加入一定比例的通用语料，防止模型忘记原有的通用知识。
5.2 GPU显存不足时的量化方案

7B模型全精度（FP16）需要约14GB显存。如果显卡不够（比如只有8G），量化是必须的。

  • Ollama 内置量化：拉取模型时就可以选择量化版本，如 qwen2.5:7b-q4_K_M。q4_K_M 表示4位量化，能大幅降低显存占用（约4-5GB），精度损失在可接受范围内。
  • 使用 llama.cpp：如果你不用Ollama，可以直接用 llama.cpp 进行量化（./quantize）和推理，它在低资源环境下的效率很高。
5.3 敏感词过滤机制实现

作为公开客服，内容安全至关重要。必须在返回给用户前进行过滤。

  • 多层过滤：
    1. 本地词库：维护一个敏感词列表，使用高效的字符串匹配算法（如AC自动机）进行过滤。
    2. 模型自检：在提示词（Prompt）中明确要求模型进行自我审查，例如添加“请确保你的回复安全、合法、健康”。
    3. 后处理拦截：即使模型生成后，也要再次经过过滤层，将敏感词替换为***或触发人工审核。
  • 示例代码片段：
class ContentFilter:
    def __init__(self, bad_words_path: str):
        with open(bad_words_path, 'r', encoding='utf-8') as f:
            self.bad_words = set(line.strip() for line in f if line.strip())
        # 可以在此初始化AC自动机

    def filter(self, text: str) -> str:
        filtered_text = text
        for word in self.bad_words:
            if word in filtered_text:
                filtered_text = filtered_text.replace(word, "***")
        return filtered_text

# 在返回回复前调用
filter = ContentFilter("bad_words.txt")
safe_reply = filter.filter(model_raw_reply)

写在最后

折腾这么一圈下来，最大的感受就是“可控”带来的安心感。响应速度自己调，数据不出私服，成本就是电费和硬件折旧。虽然前期踩了不少坑，比如模型量化后效果下降、Webhook签名验证调试、对话历史管理策略等，但看到自己搭建的客服机器人在测试环境里流畅地回答问题时，成就感还是满满的。

当然，这套方案还有很多可以深挖的地方。比如，如何设计一个更优雅的多轮对话上下文管理策略？ 是简单粗暴地截断最近N条？还是尝试用更小的模型去总结历史对话？或者是将长上下文向量化存储，按相关性检索？这又是一个值得探索的话题了。

希望这篇笔记能为你搭建自己的智能客服系统提供一条清晰的路径。如果有任何问题或更好的实践，欢迎一起交流。

Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

从零开始：用Python搭建你的第一个财务数据分析智能体

AI Agent（人工智能智能体）可以理解为一个能自己“思考”并“动手做事”的AI程序。传统的AI就像一个问答机器人——你问一句，它答一句，像个只会接话的客服。而AI Agent更像一个有主见的助理——你给它一个目标（比如“帮我分析一下上季度的财务数据”），它会自己规划怎么做、调用什么工具、按什么顺序执行，最后把完整的结果交给你。传统AI是“你问什么它答什么”，AI Agent是“你给我目标，我自

avatar AI Agent技术社区

smolagents：用代码思考的AI Agent框架

Hugging Face推出的smolagents框架是一个创新的AI Agent开发工具，通过让AI直接生成Python代码而非传统JSON指令来完成任务。该框架核心代码不足千行，设计极简且支持多种主流LLM模型和工具生态，包括多模态处理能力。相比传统方法，代码执行方式可减少30%的步骤，效率更高。框架提供沙箱安全方案，并有命令行工具支持非编程使用。其核心优势在于高效代码执行、简洁可修改的设计以

avatar AI Agent技术社区

从 Prompt 到 Loop：理清 AI Agent 工程的概念演进

也就是说，模型之外的所有东西都是 Harness。你说"Harness"，可能指的是 Claude Code 这个产品我说"Harness"，可能指的是 Initializer + Coding Agent 这种设计模式他说"Harness"，可能指的是他项目里的 AGENT.md 配置文件三个人用同一个词，说的完全不是同一个东西。这就像说"软件工程"——你可能在说设计模式，也可能在说编程语言，也

avatar AI Agent技术社区