LLM与Agent开发实战：从零构建智能对话系统的避坑指南

1次阅读

没有评论

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

传统规则引擎对话系统和基于 LLM 的智能对话系统各有优劣。规则引擎的优势在于可控性强、响应速度快，但缺点也很明显：

意图识别僵化：需要人工编写大量规则，难以覆盖用户表达的多样性
上下文管理困难：对话状态需要显式维护，容易丢失上下文
扩展成本高：每增加一个新功能都需要修改规则体系

而 LLM 方案则面临三大典型问题：

意图漂移(Intent Drift)：模型可能会误解用户意图，特别是在多轮对话中
上下文丢失(Context Lost)：受限于 token 窗口，长对话时历史信息可能被截断
响应不可控 (Uncontrolled Output)：模型可能产生幻觉(hallucination) 或不合规内容

主流 LLM API 对比（以中文场景为例）：

模型	响应速度	token 成本(千字)	中文理解	适合场景
GPT-4	中等	$0.06	★★★★★	复杂逻辑、高准确度需求
Claude 3	快	$0.04	★★★★☆	长文本处理
LLaMA 3	慢	$0.02	★★★☆☆	私有化部署、低成本
文心一言	快	¥0.03	★★★★★	中文专有领域

@startuml
component "用户输入" as input
component "路由模块(Router)" as router
component "工具集(Tools)" as tools
component "记忆模块(Memory)" as memory
component "LLM 核心" as llm
component "输出处理" as output

input --> router
router --> tools : 工具调用
router --> llm : 自然语言处理
tools --> memory : 保存结果
llm --> memory : 对话历史
memory --> router : 上下文注入
llm --> output
@enduml

对话状态机 (Dialog State Machine) 通过有限状态控制对话流程：

初始状态：等待用户输入
意图识别：通过 LLM 分类用户意图
槽位填充：提取关键参数(如查询城市)
动作执行：调用对应工具
响应生成：格式化输出
状态持久化：更新对话上下文

以下是基于 LangChain 的 WeatherAgent 示例：

from typing import List, Dict
from langchain.agents import AgentExecutor, Tool
from langchain.memory import ConversationBufferMemory
from langchain.chat_models import ChatOpenAI
from langchain.agents import initialize_agent
from tenacity import retry, stop_after_attempt

# 天气查询工具
@retry(stop=stop_after_attempt(3))
def get_weather(location: str) -> str:
    """调用天气 API（示例用模拟数据）"""
    if "北京" in location:
        return "北京天气：晴，25℃"
    return f"{location}天气数据暂不可用"

# 初始化 LLM
llm = ChatOpenAI(
    model_name="gpt-3.5-turbo",
    temperature=0.5  # 降低随机性
)

# 定义工具集
weather_tool = Tool(
    name="Weather",
    func=get_weather,
    description="查询城市天气，参数格式：' 城市名 '"
)

tools = [weather_tool]

# 记忆模块
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# 创建 Agent
agent = initialize_agent(
    tools,
    llm,
    agent="conversational-react-description",
    memory=memory,
    verbose=True  # 打印执行日志
)

# 执行示例
response = agent.run("北京明天天气怎么样？")
print(response)

关键设计点：

使用 @retry 实现异常自动重试
temperature=0.5平衡响应创造性
verbose=True方便调试
类型注解提升代码可维护性

日志脱敏：使用正则过滤手机号 / 身份证号

import re

def sanitize_log(text: str) -> str:
    text = re.sub(r'1[3-9]\d{9}', '[PHONE]', text)
    return re.sub(r'[1-9]\d{5}(19|20)\d{2}[0-9Xx]', '[ID]', text)

限流配置：

from fastapi import FastAPI, Request
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter

@app.post("/chat")
@limiter.limit("5/minute")  # 每分钟 5 次
async def chat_endpoint(request: Request):
    ...