AI助手Agent开发实战：从零构建智能对话系统的完整指南

1次阅读

没有评论

共计 2926 个字符，预计需要花费 8 分钟才能阅读完成。

刚接触 AI 助手 Agent 开发时，最让人头疼的三个问题往往是：

上下文保持困难 ：用户说 ” 价格能再优惠吗 ” 时，Agent 得记住前文讨论的是哪款产品
多轮对话逻辑复杂 ：预订酒店场景中，处理 ” 先看房型→比价→修改日期 ” 的跳转会变得像走迷宫
API 调用成本失控 ：当用户连续发送 10 条消息时，不小心就会触发上百次 LLM API 调用

这些痛点不解决，做出来的 Agent 要么答非所问，要么开发成本高得离谱。下面我们就用 Python 手把手搭建一个经济实用的解决方案。

优点：完全自托管，对话规则可精确控制
缺点：需要大量标注数据训练 NLU 模型
典型场景：银行客服等需要严格流程控制的场景

优点：谷歌提供的可视化流程设计器
缺点：中文意图识别准确率约 85% 左右
典型场景：电商 FAQ 机器人等标准化场景

优点：轻松集成 LLM 且支持自定义工具链
缺点：学习曲线较陡峭
典型场景：需要结合知识库的智能助手

个人建议 ：从 LangChain 起步，既能利用现成 LLM 能力，又保留足够灵活性。

用 Sentence-Transformer 生成语义 embedding，比直接调用 LLM 便宜 90%：

from sentence_transformers import SentenceTransformer
import numpy as np

# 初始化模型（首次运行会自动下载）encoder = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

# 示例意图库
intent_db = {"greeting": ["你好", "嗨", "早上好"],
    "booking": ["我想订酒店", "有空房吗", "预订"],
    "complaint": ["投诉", "不满意", "服务太差"]
}

# 预计算所有意图的 embedding
intent_embeddings = {k: encoder.encode(v) for k,v in intent_db.items()}

def detect_intent(text, threshold=0.75):
    query_embed = encoder.encode(text)
    scores = {}

    # 计算余弦相似度（时间复杂度 O(n)）for intent, embeds in intent_embeddings.items():
        sims = [cosine_similarity(query_embed, e) for e in embeds]
        scores[intent] = max(sims)

    best_intent = max(scores, key=scores.get)
    return best_intent if scores[best_intent] > threshold else "unknown"

用枚举类实现酒店预订的状态流转：

from enum import Enum, auto

class BookingState(Enum):
    START = auto()
    SELECT_DATE = auto()
    SELECT_ROOM = auto()
    CONFIRM = auto()
    PAYMENT = auto()
    COMPLETE = auto()

class DialogManager:
    def __init__(self):
        self.state = BookingState.START
        self.context = {}

    def handle(self, user_input):
        if self.state == BookingState.START:
            if "预订" in user_input:
                self.state = BookingState.SELECT_DATE
                return "请问您要入住哪一天？"

        elif self.state == BookingState.SELECT_DATE:
            self.context["date"] = extract_date(user_input)  # 假设已实现日期提取
            self.state = BookingState.SELECT_ROOM
            return show_room_types()

        # 其他状态处理省略...

        return "抱歉，我没听懂"

用 Redis 实现请求缓冲，防止突发流量打爆 API：

import redis
import json
from threading import Thread

r = redis.Redis(host='localhost', port=6379)

class MessageQueue:
    @staticmethod
    def consume():
        while True:
            _, msg = r.brpop("agent_queue", timeout=30)
            data = json.loads(msg)
            process_message(data)  # 实际处理函数

# 启动消费线程
Thread(target=MessageQueue.consume, daemon=True).start()

# 生产消息示例
r.lpush("agent_queue", json.dumps({"user_id": 123, "text": "我想订房"}))

存储前必须过滤手机号 / 身份证号：

import re

def sanitize_text(text):
    text = re.sub(r'1[3-9]\d{9}', '[PHONE]', text)  # 手机号
    text = re.sub(r'[1-9]\d{5}(19|20)\d{2}[0-9Xx]', '[ID]', text)  # 身份证
    return text

使用令牌桶算法控制 LLM 调用频率：

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(
    app=current_app,
    key_func=get_remote_address,
    default_limits=["5 per minute"]  # 每个 IP 每分钟 5 次
)

意图识别模型初期可以这样改进：