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

1次阅读

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

Agent Prompt（智能代理提示）是传统 Prompt 工程的进阶形态。简单来说，传统 Prompt 像一次性指令（比如让 ChatGPT 写首诗），而 Agent Prompt 则是给 AI 装上了 ” 大脑 ” 和 ” 工具箱 ”：

状态保持（Memory）：能记住对话历史（比如知道用户昨天问了天气）
工具调用（Tool Usage）：可以调用外部 API 或函数（比如查询实时股价）
自主决策（Decision Making）：根据上下文决定下一步动作

举个生活化的例子：传统 Prompt 像是问路人 ” 最近的咖啡店在哪 ”，而 Agent Prompt 更像雇佣了一个私人助理，他会记住你的口味偏好，主动推荐新品，还能帮你下单支付。

我们先实现一个混合记忆系统，用列表存储短期记忆，用文件模拟长期记忆：

import json
from datetime import datetime

class AgentMemory:
    def __init__(self):
        self.short_term = []  # 短期记忆（最近 5 条）self.long_term_file = 'memory.json'  # 长期记忆存储文件

    def add_memory(self, event_type, content):
        """添加记忆"""
        memory = {'timestamp': str(datetime.now()),
            'type': event_type,
            'content': content
        }

        # 短期记忆维护
        self.short_term.append(memory)
        if len(self.short_term) > 5:
            self.short_term.pop(0)

        # 长期记忆存储
        try:
            with open(self.long_term_file, 'a') as f:
                f.write(json.dumps(memory) + '\n')
        except Exception as e:
            print(f"记忆存储失败: {e}")

    def recall(self, keyword):
        """回忆相关事件"""
        results = []
        # 先查短期记忆
        for mem in self.short_term:
            if keyword in str(mem):
                results.append(mem)

        # 再查长期记忆
        try:
            with open(self.long_term_file, 'r') as f:
                for line in f:
                    mem = json.loads(line.strip())
                    if keyword in str(mem):
                        results.append(mem)
        except FileNotFoundError:
            pass

        return results[:3]  # 返回最近 3 条相关记忆

用装饰器模式实现工具注册，支持本地函数和 HTTP 请求两种类型：

import requests
from functools import wraps

class ToolBox:
    def __init__(self):
        self.tools = {}

    def register(self, name, desc=''):""" 工具注册装饰器 """
        def decorator(func):
            self.tools[name] = {
                'function': func,
                'description': desc
            }
            @wraps(func)
            def wrapper(*args, **kwargs):
                return func(*args, **kwargs)
            return wrapper
        return decorator

    async def call_http(self, url, method='GET', params=None):
        """HTTP 请求工具（Python3.7+ 需 async）"""
        try:
            resp = requests.request(
                method, 
                url,
                params=params,
                timeout=3
            )
            return resp.json()
        except Exception as e:
            return {'error': str(e)}

# 示例工具注册
if __name__ == '__main__':
    toolbox = ToolBox()

    @toolbox.register('get_weather', '查询城市天气')
    def get_weather(city):
        """模拟天气查询"""
        return {
            'city': city,
            'weather': 'sunny',
            'temp': 25
        }

    # 使用 HTTP 工具
    async def search_news(keyword):
        return await toolbox.call_http(
            'https://news-api.com/search',
            params={'q': keyword}
        )

class SimpleAgent:
    def __init__(self):
        self.memory = AgentMemory()
        self.tools = ToolBox()

    def process_input(self, user_input):
        """完整的决策处理流程"""
        # 1. 记忆更新
        self.memory.add_memory('user_input', user_input)

        # 2. 意图识别（简化版）if '天气' in user_input:
            city = user_input.replace('天气', '').strip()
            weather = self.tools.tools['get_weather']['function'](city)
            response = f"{city}天气是{weather['weather']}，温度{weather['temp']}℃"
        else:
            response = "我还在学习中，暂时不明白您的需求"

        # 3. 记录系统响应
        self.memory.add_memory('system_response', response)
        return response

内存方式：仅适合开发测试

# 不推荐生产环境使用
agent = SimpleAgent()  # 每次重启服务记忆清零

Redis 方案：推荐生产使用

import redis

class RedisMemory:
    def __init__(self):
        self.r = redis.Redis(
            host='localhost',
            port=6379,
            db=0,
            decode_responses=True
        )

    def add_memory(self, user_id, memory):
        self.r.lpush(f"user:{user_id}:memories", json.dumps(memory))
        self.r.ltrim(f"user:{user_id}:memories", 0, 99)  # 保留最近 100 条

超时机制：所有 HTTP 请求必须设置超时

# 好例子
requests.get(url, timeout=3)  # 3 秒超时

# 反面教材（可能永远挂起）requests.get(url)

敏感词过滤：基础防御措施

BANNED_WORDS = ['密码', '银行卡']

def sanitize_input(text):
    for word in BANNED_WORDS:
        if word in text:
            raise ValueError("包含敏感词汇")
    return text

当你掌握基础 Agent 构建后，可以尝试这些进阶方向：

优先级调度：
给工具添加 priority 属性
在决策循环中根据紧急程度排序任务

多 Agent 协作：

class MultiAgentSystem:
    def __init__(self):
        self.agents = {'research': ResearchAgent(),
            'customer': CustomerServiceAgent()}

    def route(self, message):
        if '投诉' in message:
            return self.agents['customer']
        else:
            return self.agents['research']