Python AI Agent 开发实战：从零构建智能代理的避坑指南

1次阅读

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

开发 AI Agent 时，新手常会遇到几个典型问题：

对话状态丢失 ：在无状态服务中（如 AWS Lambda），若不刻意保存，用户对话上下文会在请求结束后消失
长上下文处理低效 ：直接将全部历史对话丢给大模型，会导致 token 爆炸和响应延迟
任务调度混乱 ：同步阻塞式调用 API 会导致并发性能急剧下降
API 稳定性问题 ：未处理限流和超时就直接上线，生产环境必定翻车

方案	优点	缺点	适用场景
原生 OpenAI API	直接控制请求细节，性能上限高	需自行实现对话管理等组件	需要极致定制的轻量级场景
LangChain	提供现成记忆、工具链等模块	抽象层较多，调试复杂度高	快速验证业务逻辑
Semantic Kernel	微软系生态集成友好	社区资源相对较少	已有 Azure 技术栈的项目

使用 asyncio 避免阻塞主线程，关键实现要点：

import asyncio
from typing import List, Coroutine

class AsyncScheduler:
    """线程安全的异步任务调度器（时间复杂度 O(n)）"""
    def __init__(self, max_concurrent: int = 10):
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def run_tasks(self, tasks: List[Coroutine]) -> list:
        async def _wrap_task(task):
            async with self.semaphore:  # 控制并发量
                try:
                    return await task
                except Exception as e:
                    print(f"Task failed: {e}")
                    return None

        return await asyncio.gather(*[_wrap_task(t) for t in tasks])

基于 Redis 的上下文缓存方案：

import redis
import pickle
from datetime import timedelta

class DialogueMemory:
    """使用 Redis 存储对话历史（时间复杂度 O(1)）"""
    def __init__(self, host: str, ttl_hours: int = 24):
        self.client = redis.Redis(host=host)
        self.ttl = timedelta(hours=ttl_hours)

    def save_context(self, session_id: str, messages: list) -> bool:
        """序列化存储对话上下文"""
        try:
            return self.client.setex(name=f"agent:{session_id}",
                time=self.ttl,
                value=pickle.dumps(messages)
            )
        except Exception as e:
            print(f"Save failed: {e}")
            return False

    def load_context(self, session_id: str) -> list:
        """反序列化读取上下文"""
        try:
            data = self.client.get(f"agent:{session_id}")
            return pickle.loads(data) if data else []
        except Exception as e:
            print(f"Load failed: {e}")
            return []

from functools import wraps
import time

class CircuitBreaker:
    """当错误率超过阈值时自动熔断（时间复杂度 O(1)）"""
    def __init__(self, max_failures=3, reset_timeout=60):
        self.failures = 0
        self.last_failure = 0
        self.max_failures = max_failures
        self.reset_timeout = reset_timeout

    def __call__(self, func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            if self._is_circuit_open():
                raise Exception("Service unavailable (circuit open)")

            try:
                result = await func(*args, **kwargs)
                self._reset()
                return result
            except Exception as e:
                self._record_failure()
                raise
        return wrapper

    def _is_circuit_open(self) -> bool:
        return (self.failures >= self.max_failures and 
                time.time() - self.last_failure < self.reset_timeout)

    def _record_failure(self):
        self.failures += 1
        self.last_failure = time.time()

    def _reset(self):
        self.failures = 0

关键策略：

对话日志加密存储（使用 AES-256）
实现自动擦除接口
日志中去除 PII（Personally Identifiable Information）信息

错误示范：

# 错误！Lambda 冷启动时会被重置
counter = 0  

def handler(event, context):
    global counter
    counter += 1
    return counter

正确做法：

使用外部存储（如 DynamoDB）
通过环境变量传递配置

处理 API 限流的黄金标准：

import random
import time

async def call_api_with_retry(func, max_retries=5):
    """时间复杂度 O(2^n) 的指数退避"""
    for attempt in range(max_retries):
        try:
            return await func()
        except RateLimitError:
            wait = min((2 ** attempt) + random.uniform(0, 1), 60)
            await asyncio.sleep(wait)
    raise Exception("Max retries exceeded")

当 Agent 需要调用外部 API 时，如何设计重试机制保证幂等性？以下是我的思考方向：