Agent实践入门指南：从零构建你的第一个智能代理

1次阅读

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

Agent 通过模块化设计实现复杂对话流程的拆解与管理
上下文感知能力使对话系统具备多轮交互的智能性
外部服务集成扩展了传统聊天机器人的能力边界

Rasa
优势：完整的 NLU 管道、丰富的企业级部署方案
适用场景：需要精细控制对话流程的客服系统
学习曲线：需掌握 YAML 配置和自定义 action
LangChain
优势：大语言模型快速集成、丰富的工具链支持
适用场景：需要结合 LLM 能力的知识型应用
注意事项：依赖外部 API 可能产生延迟

# 正则表达式方案
def regex_intent(text):
    import re
    weather_pattern = r'(天气 | 气温 | 下雨 | 下雪 | 温度)'
    return bool(re.search(weather_pattern, text))

# 机器学习方案（需准备训练数据）from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.svm import LinearSVC

class MLIntentClassifier:
    def __init__(self):
        self.vectorizer = TfidfVectorizer()
        self.clf = LinearSVC()

    def train(self, texts, labels):
        X = self.vectorizer.fit_transform(texts)
        self.clf.fit(X, labels)  # 时间复杂度 O(n_samples * n_features)

    def predict(self, text):
        return self.clf.predict(self.vectorizer.transform([text]))[0]

stateDiagram
    [*] --> 初始状态
    初始状态 --> 获取城市: 识别天气查询意图
    获取城市 --> 查询天气: 收到有效城市名称
    查询天气 --> 展示结果: API 返回成功
    展示结果 --> [*]: 超时或用户确认

class WeatherStateMachine:
    def __init__(self):
        self.state = 'INIT'
        self.city = None

    def transition(self, user_input):
        if self.state == 'INIT' and '天气' in user_input:
            self.state = 'GET_CITY'
            return "请输入城市名称"

        elif self.state == 'GET_CITY':
            self.city = extract_city(user_input)  # 城市提取函数
            if self.city:
                self.state = 'FETCH_WEATHER'
                return self._fetch_weather()
            return "城市识别失败，请重新输入"

    async def _fetch_weather(self):
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(f'https://api.weather.com/{self.city}') as resp:
                    if resp.status == 200:
                        self.state = 'SHOW_RESULT'
                        return await resp.json()
                    raise Exception(f'API error: {resp.status}')
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            self.state = 'ERROR'
            return f"服务暂时不可用: {type(e).__name__}"

日志脱敏 ：

def sanitize_log(text):
    import re
    return re.sub(r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b', '[IP]', text)

会话隔离 ：

from threading import Lock

class SessionManager:
    def __init__(self):
        self.sessions = {}
        self.lock = Lock()

    def get_session(self, session_id):
        with self.lock:
            if session_id not in self.sessions:
                self.sessions[session_id] = WeatherStateMachine()
            return self.sessions[session_id]

压测指标 （使用 Locust 测试）：

from locust import HttpUser, task

class WeatherUser(HttpUser):
    @task
    def query_weather(self):
        self.client.post("/chat", json={"text":"北京天气"})