Agent人工智能新手入门：从零搭建你的第一个智能代理系统

1次阅读

没有评论

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

Agent（智能代理）是一种能够自主感知环境、做出决策并执行动作的软件实体。与普通 Chatbot 相比，它具有三个核心特性：

自主性：能在没有直接干预的情况下控制自身行为和内部状态
反应性：能够感知环境变化并做出及时响应
目标导向：具备明确的目标并主动采取行动实现它

举个简单例子：一个天气查询 Chatbot 只会被动回答用户提问，而一个天气 Agent 可能会主动在检测到用户所在地将下雨时，自动建议携带雨具并调整行程安排。

刚开始接触 Agent 开发时，容易陷入以下几个陷阱：

过度关注自然语言理解(NLU)：很多初学者把 90% 精力花在文本解析上，却忽视了 Agent 的核心——决策逻辑
状态管理混乱：用全局变量随意存储对话状态，导致多用户访问时数据互相覆盖
缺乏异常处理：没有考虑 API 调用失败、用户输入异常等情况，系统脆弱易崩溃
忽略超时控制：长时间等待用户响应导致资源占用，影响系统吞吐量

下面我们用 Python 构建一个基于规则 + 有限状态机的问答 Agent 框架。这个示例包含完整的对话上下文管理，并集成了天气查询功能。

class SimpleAgent:
    def __init__(self):
        # 使用字典存储不同用户的对话状态
        self.user_sessions = {}  
        # 定义有限状态机的状态转移规则
        self.state_transitions = {'GREETING': {'next_states': ['QUERY_WEATHER', 'END']},
            'QUERY_WEATHER': {'next_states': ['END']}
        }

    def handle_message(self, user_id, message):
        """处理用户消息的核心方法"""
        # 初始化新用户的会话状态
        if user_id not in self.user_sessions:
            self.user_sessions[user_id] = {
                'current_state': 'GREETING',
                'context': {},
                'last_active': time.time()}

        session = self.user_sessions[user_id]
        session['last_active'] = time.time()  # 更新最后活跃时间

        try:
            # 根据当前状态处理消息
            if session['current_state'] == 'GREETING':
                return "你好！我是天气助手，你可以问我某个城市的天气情况"

            elif session['current_state'] == 'QUERY_WEATHER':
                city = self._extract_city(message)
                if not city:
                    return "请告诉我您想查询哪个城市的天气？"

                # 调用外部天气 API（模拟）weather = self._get_weather(city)
                session['current_state'] = 'END'
                return f"{city}的天气是{weather}"

        except Exception as e:
            # 记录错误并优雅降级
            print(f"处理消息出错: {str(e)}")
            session['current_state'] = 'END'
            return "抱歉，处理您的请求时出现问题"

    def _extract_city(self, text):
        """简单的城市名提取逻辑（实际项目应使用 NER）"""
        # 这里简化为直接返回第一个中文词汇
        import re
        match = re.search(r'[\u4e00-\u9fa5]+', text)
        return match.group(0) if match else None

    def _get_weather(self, city):
        """模拟天气 API 调用"""
        # 实际项目中这里应该是真实 API 调用
        weather_data = {
            '北京': '晴天 25℃',
            '上海': '多云 28℃',
            '广州': '阵雨 30℃'
        }
        return weather_data.get(city, '未知天气')

在上面的代码中，我们使用 user_sessions 字典来存储不同用户的对话状态。这里有两个重要细节：

使用用户 ID 作为 key，确保不同用户的数据隔离
在实际 web 应用中，应考虑使用线程安全的数据结构或数据库存储

我们记录了每个会话的 last_active 时间戳，可以定期清理长时间不活动的会话：

# 在 Agent 类中添加这个方法
def cleanup_inactive_sessions(self, timeout=300):
    """清理超过 5 分钟不活动的会话"""
    current_time = time.time()
    inactive_users = [user_id for user_id, session in self.user_sessions.items()
        if current_time - session['last_active'] > timeout
    ]
    for user_id in inactive_users:
        del self.user_sessions[user_id]

任何面向用户的系统都必须包含内容安全机制：

class SafetyChecker:
    def __init__(self):
        # 实际项目中应从文件或数据库加载
        self.banned_words = ['敏感词 1', '敏感词 2']  

    def check_input(self, text):
        """检查用户输入是否包含敏感词"""
        for word in self.banned_words:
            if word in text:
                return False
        return True

根据实现方式不同，Agent 系统的性能特征差异很大：