AI Agent应用开发入门：从零构建你的第一个智能代理

1次阅读

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

传统程序像烹饪食谱——严格按步骤执行，而 AI Agent 更像是餐厅厨师：能理解模糊需求（” 少放盐 ”）、处理意外（食材缺货时替换），并主动与环境交互（询问客人忌口）。其核心能力包括：

感知决策闭环 ：通过传感器 / 输入获取信息 → 内部状态更新 → 行动输出
目标导向性 ：为实现特定目标持续优化行为
环境适应性 ：应对非预设场景（如用户突然切换话题）

智能客服 ：
自动回答高频问题（订单查询 / 退换货政策）
7×24 小时服务且响应速度 <500ms
难点：处理 ” 我的订单好像出问题了 ” 这类模糊表达
流程自动化 ：
会议纪要自动生成并分发给相关人员
需整合日历 API、语音识别、邮件系统
挑战：识别不同发言人并关联发言内容
游戏 NPC：
《荒野大镖客 2》中 NPC 会记住玩家行为并改变互动方式
技术关键：长期记忆存储与情感计算模型

pip install scikit-learn==1.3.0 flask==2.3.2

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.neighbors import NearestNeighbors

# 示例训练数据
intent_examples = {"greeting": ["你好", "早上好", "hi"],
    "weather": ["今天天气怎样", "会下雨吗"],
    "goodbye": ["再见", "下次聊"]
}

# 特征提取
all_texts = []
labels = []
for intent, texts in intent_examples.items():
    all_texts.extend(texts)
    labels.extend([intent]*len(texts))

vectorizer = TfidfVectorizer()
X = vectorizer.fit_transform(all_texts)

# 最近邻分类器
clf = NearestNeighbors(n_neighbors=1)
clf.fit(X)

def predict_intent(query):
    query_vec = vectorizer.transform([query])
    distance, index = clf.kneighbors(query_vec)
    return labels[index[0][0]]

class DialogState:
    def __init__(self):
        self.current_intent = None
        self.slots = {}  # 例如 {"city": "北京"} 
        self.context = {}  # 长期记忆

    def update(self, user_input):
        intent = predict_intent(user_input)
        self.current_intent = intent

        # 简单的槽位填充逻辑
        if intent == "weather":
            if "北京" in user_input:
                self.slots["city"] = "北京"
        return intent

import requests

def get_weather(city):
    # 模拟天气 API 调用
    mock_data = {
        "北京": "晴, 25℃",
        "上海": "多云, 28℃"
    }
    return mock_data.get(city, "暂不支持该城市")

# Flask 服务端示例
from flask import Flask, request, jsonify

app = Flask(__name__)
state = DialogState()

@app.route('/chat', methods=['POST'])
def chat():
    user_input = request.json['query']
    intent = state.update(user_input)

    if intent == "weather":
        city = state.slots.get("city", "北京")  # 默认城市
        weather = get_weather(city)
        return jsonify({"response": f"{city} 天气：{weather}"})

    return jsonify({"response": "已收到您的消息"})

短期记忆 ：
用对话 ID 关联 Redis 缓存，TTL 设置 2 小时

存储结构示例：

{
  "last_intent": "weather",
  "slots": {"city": "上海"},
  "history": ["用户：上海天气", "Agent：上海天气：多云"]
}

长期记忆 ：
重要信息（如用户偏好）存入 MySQL
配合向量数据库实现语义搜索（” 上次提到的餐厅 ”）

输入清洗 ：

def sanitize_input(text):
    # 移除特殊字符但保留中文标点
    return re.sub(r'[\\\/\|\`\^\~\=\+\<\>]', '', text)

降级策略 ：

try:
    api_response = call_external_api()
except Exception as e:
    logging.error(f"API 调用失败: {str(e)}")
    return get_cached_response()  # 返回最近成功结果

超时控制 ：

from functools import wraps
import signal

class TimeoutError(Exception):
    pass

def timeout(seconds=3):
    def decorator(func):
        def handler(signum, frame):
            raise TimeoutError()

        @wraps(func)
        def wrapper(*args, **kwargs):
            signal.signal(signal.SIGALRM, handler)
            signal.alarm(seconds)
            try:
                result = func(*args, **kwargs)
            finally:
                signal.alarm(0)
            return result
        return wrapper
    return decorator

指标名称	健康阈值	采集方式
意图识别准确率	>85%	每周人工标注测试集
平均响应延迟	<800ms	Prometheus 计时器
会话异常中断率	<5%	日志分析会话终止状态