从零构建Agent意图识别工具：调用流程与实战避坑指南

1次阅读

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

意图识别是 NLP 领域的关键技术，主要分为三个阶段：

输入预处理 ：清理用户输入的文本（如去除特殊字符、纠错、分词等）。例如用户输入 ” 我想订明天去北京的票 ”，需标准化为 ” 订票北京明天 ”。
特征提取 ：将文本转换为机器可理解的向量。常见方法：
Bag-of-Words
TF-IDF
预训练模型（如 BERT）的 embedding
分类决策 ：通过分类模型（如 SVM、神经网络）判断意图类别。输出可能是单一意图（如 ” 订票 ”）或多意图组合（” 咨询 + 比价 ”）。

在实际调用 API 时，新手常遇到这些问题：

异步响应处理不当 ：直接轮询导致资源浪费或响应丢失
多意图冲突 ：未处理权重相近的意图（如 ” 退订 ” 和 ” 投诉 ” 同时命中）
上下文丢失 ：未携带 session_id 导致对话断层

import requests
from typing import Dict, Any

def call_intent_api(text: str, timeout: int = 5) -> Dict[str, Any]:
    """
    :param text: 用户输入文本 (建议长度 <500 字符)
    :param timeout: 超时时间 (秒)，建议 3 -10
    """url ="https://api.example.com/intent/v1/recognize"headers = {"Authorization":"Bearer YOUR_API_KEY"}
    payload = {"text": text, "lang": "zh-CN"}

    try:
        # 建议始终设置超时
        response = requests.post(url, json=payload, headers=headers, timeout=timeout)
        response.raise_for_status()  # 自动处理 4xx/5xx 错误
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"API 调用失败: {str(e)}")
        return {"error": str(e)}

典型返回结构示例：

{
  "intents": [
    {
      "name": "book_flight",
      "confidence": 0.92,
      "slots": {"city": "北京", "date": "2023-12-01"}
    }
  ],
  "session_id": "abcd1234"
}

解析代码：

result = call_intent_api("我想订明天去北京的机票")
if "intents" in result:
    primary_intent = result["intents"][0]  # 取置信度最高的意图
    print(f"识别到意图: {primary_intent['name']}")
    print(f"目的地: {primary_intent['slots'].get('city')}")

rate_limit：客户端限流（建议 200-500 请求 / 分钟）

from ratelimit import limits

@limits(calls=300, period=60)
def safe_api_call(text):
    return call_intent_api(text)

session_timeout：对话会话保持时间（建议 120-300 秒）
fallback_intent：设置默认意图（如 ”unknown”）避免空返回

开启详细日志：

import logging
logging.basicConfig(level=logging.DEBUG)  # 查看 HTTP 请求细节

检查意图置信度阈值（通常 >0.7 才视为有效）

方式	QPS 上限	资源占用	实现复杂度
同步调用	50-100	低	★★☆☆☆
异步 IO	300+	中	★★★☆☆
连接池复用	200+	低	★★☆☆☆

推荐方案：

# 使用 aiohttp 实现异步调用
import aiohttp

async def async_call(texts: List[str]):
    async with aiohttp.ClientSession() as session:
        tasks = [session.post(API_URL, json={"text": t}) for t in texts]
        return await asyncio.gather(*tasks)