MemGPT Python SDK详解：客户端开发完整教程
·
MemGPT Python SDK详解：客户端开发完整教程
项目地址: https://gitcode.com/GitHub_Trending/me/MemGPT 概述
MemGPT（Memory-GPT）是一个革命性的AI代理框架，通过为大型语言模型（LLM）提供内存管理能力，实现了无限上下文处理。本文将深入解析MemGPT Python SDK的完整使用方法，帮助开发者快速掌握客户端开发技巧。
环境准备
安装MemGPT
# 通过Git安装最新版本
pip install git+https://gitcode.com/GitHub_Trending/me/MemGPT.git
# 或者安装稳定版本
pip install memgpt
依赖检查
import sys
print(f"Python版本: {sys.version}")
# 确保Python版本在3.8以上
核心客户端架构
MemGPT Python SDK采用模块化设计，主要包含以下核心组件：
客户端初始化
基础连接配置
from letta import create_client
# 本地服务器连接
client = create_client(
base_url="http://localhost:8283",
token="your-auth-token"
)
# 远程服务器连接
client = create_client(
base_url="https://your-letta-server.com",
token="sk-your-secret-key"
)
认证机制
MemGPT支持多种认证方式：
| 认证类型 | 配置方式 | 适用场景 |
|---|---|---|
| API Key | token参数 | 生产环境 |
| OAuth2 | OAuth流程 | 第三方集成 |
| 本地认证 | 无token | 开发测试 |
代理管理
创建基础代理
# 创建简单代理
basic_agent = client.create_agent(
name="basic_assistant",
model="gpt-4", # 指定LLM模型
embedding="text-embedding-ada-002" # 指定嵌入模型
)
print(f"代理创建成功: {basic_agent.name} (ID: {basic_agent.id})")
个性化代理配置
from letta_client import CreateBlock
# 定义个性化配置
persona = """
你是一个专业的AI助手，擅长技术咨询和问题解答。
你始终保持友好和专业的态度。
"""
user_profile = """
我是一名全栈开发者，专注于Python和AI技术。
我喜欢学习新技术和解决复杂问题。
"""
# 创建个性化代理
custom_agent = client.agents.create(
name="tech_assistant",
memory_blocks=[
CreateBlock(
label="persona",
value=persona
),
CreateBlock(
label="user",
value=user_profile
)
],
model="gpt-4-turbo",
embedding="text-embedding-3-small"
)
内存块类型说明
MemGPT支持多种内存块类型：
| 内存块类型 | 描述 | 示例 |
|---|---|---|
| persona | 代理个性定义 | 专业助手、幽默伴侣 |
| user | 用户配置信息 | 职业、偏好 |
| system | 系统级配置 | 工具使用规则 |
| archival | 长期记忆存储 | 历史对话摘要 |
消息交互
发送消息
from letta.client.utils import pprint
from letta_client import MessageCreate
# 发送单条消息
response = client.agents.messages.create(
agent_id=custom_agent.id,
messages=[
MessageCreate(
role="user",
content="请介绍一下Python的异步编程"
)
]
)
# 美化输出消息
pprint(response.messages)
流式消息处理
from letta.client.streaming import LettaStreamingResponse
# 流式消息处理示例
def handle_streaming_response():
streaming_client = Letta(base_url="http://localhost:8283")
response_stream = streaming_client.agents.messages.create(
agent_id=agent_id,
messages=[MessageCreate(role="user", content="讲一个技术故事")],
stream=True
)
for chunk in response_stream:
if isinstance(chunk, LettaStreamingResponse):
print(chunk.content, end="", flush=True)
批量消息处理
# 批量发送消息
batch_messages = [
MessageCreate(role="user", content="第一个问题"),
MessageCreate(role="user", content="第二个问题"),
MessageCreate(role="user", content="第三个问题")
]
responses = []
for msg in batch_messages:
response = client.agents.messages.create(
agent_id=agent_id,
messages=[msg]
)
responses.append(response)
工具集成
查看可用工具
# 列出所有可用工具
tools_response = client.list_tools()
available_tools = tools_response.tools
print("可用工具列表:")
for tool in available_tools:
print(f"- {tool.name}: {tool.json_schema.get('description', '无描述')}")
print(f" 参数: {list(tool.json_schema.get('parameters', {}).get('properties', {}).keys())}")
工具启用配置
# 创建带工具的代理
tool_agent = client.create_agent(
name="tool_enabled_agent",
model="gpt-4",
tools=["web_search", "calculator", "file_reader"] # 启用特定工具
)
# 工具使用示例对话
response = client.user_message(
agent_id=tool_agent.id,
message="请搜索最新的AI技术动态"
)
内存管理
核心内存操作
# 查看代理内存状态
agent_state = client.agents.get(agent_id=custom_agent.id)
print("核心内存内容:")
print(f"Persona: {agent_state.core_memory.persona}")
print(f"User: {agent_state.core_memory.user}")
# 更新内存内容
update_response = client.agents.update(
agent_id=custom_agent.id,
memory_blocks=[
CreateBlock(
label="user",
value="我现在是一名机器学习工程师"
)
]
)
归档内存管理
# 查看归档记忆
archival_memory = client.agents.memory.archival.list(
agent_id=agent_id,
limit=10 # 获取最近10条归档记忆
)
print("归档记忆:")
for memory in archival_memory:
print(f"- {memory.content} (时间: {memory.created_at})")
高级功能
多代理协作
# 创建多个协作代理
analyst_agent = client.create_agent(
name="data_analyst",
persona="你是一个数据分析专家，擅长数据解读和可视化"
)
writer_agent = client.create_agent(
name="content_writer",
persona="你是一个技术作家，擅长将复杂概念转化为易懂的内容"
)
# 代理间消息传递（通过用户中转）
analysis = client.user_message(
agent_id=analyst_agent.id,
message="分析这份销售数据趋势"
)
summary = client.user_message(
agent_id=writer_agent.id,
message=f"请将这份分析结果写成报告: {analysis.messages[-1].content}"
)
自定义工具开发
from typing import Dict, Any
from letta_client import ToolCreate
# 定义自定义工具
def custom_calculator(expression: str) -> Dict[str, Any]:
"""自定义计算器工具"""
try:
result = eval(expression)
return {"result": result, "status": "success"}
except Exception as e:
return {"error": str(e), "status": "error"}
# 注册工具（需要管理员权限）
tool_schema = {
"name": "custom_calculator",
"description": "执行数学表达式计算",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式"
}
},
"required": ["expression"]
}
}
# 注意：工具注册通常需要服务器端配置
错误处理与调试
异常处理机制
from letta.errors import LettaAPIError, AuthenticationError
try:
response = client.user_message(agent_id=agent_id, message="测试消息")
except AuthenticationError as e:
print(f"认证失败: {e}")
# 重新获取token或提示用户登录
except LettaAPIError as e:
print(f"API错误: {e.status_code} - {e.message}")
# 根据状态码进行相应处理
except Exception as e:
print(f"未知错误: {e}")
# 通用错误处理
调试与日志
import logging
# 配置详细日志
logging.basicConfig(level=logging.DEBUG)
# 请求调试信息
def debug_request():
import httpx
from letta.client import Letta
client = Letta(base_url="http://localhost:8283")
# 启用请求日志
with httpx.Client() as http_client:
client._client = http_client
response = client.agents.list()
print(f"请求详情: {response._request}")
print(f"响应详情: {response._response}")
性能优化
连接池配置
import httpx
from letta import create_client
# 使用连接池提高性能
http_client = httpx.Client(
limits=httpx.Limits(
max_keepalive_connections=10,
max_connections=20
),
timeout=30.0
)
client = create_client(
base_url="http://localhost:8283",
token="your-token",
http_client=http_client # 传入自定义HTTP客户端
)
批量操作优化
from concurrent.futures import ThreadPoolExecutor
def batch_process_agents(agent_ids, messages):
"""批量处理多个代理的消息"""
results = []
with ThreadPoolExecutor(max_workers=5) as executor:
future_to_agent = {
executor.submit(
client.user_message,
agent_id=agent_id,
message=message
): agent_id
for agent_id, message in zip(agent_ids, messages)
}
for future in concurrent.futures.as_completed(future_to_agent):
agent_id = future_to_agent[future]
try:
result = future.result()
results.append((agent_id, result))
except Exception as e:
print(f"代理 {agent_id} 处理失败: {e}")
return results
实战案例：智能客服系统
class CustomerServiceAgent:
def __init__(self, client, base_persona):
self.client = client
self.base_persona = base_persona
self.agents = {} # 存储用户对应的代理
def get_or_create_agent(self, user_id):
"""获取或创建用户专属代理"""
if user_id in self.agents:
return self.agents[user_id]
# 创建个性化代理
agent = self.client.create_agent(
name=f"cs_agent_{user_id}",
persona=self.base_persona,
user=f"用户ID: {user_id}"
)
self.agents[user_id] = agent
return agent
def handle_message(self, user_id, message):
"""处理用户消息"""
agent = self.get_or_create_agent(user_id)
response = self.client.user_message(
agent_id=agent.id,
message=message
)
# 记录交互历史
self._log_interaction(user_id, message, response)
return response.messages[-1].content
def _log_interaction(self, user_id, message, response):
"""记录交互日志"""
# 实现日志记录逻辑
pass
# 使用示例
cs_persona = """
你是一个专业的客服助手，擅长解决技术问题。
你始终保持耐心和友好，能够理解用户的需求。
"""
cs_system = CustomerServiceAgent(client, cs_persona)
response = cs_system.handle_message("user123", "我的账户无法登录")
print(response)
最佳实践总结
开发规范
- 资源管理: 及时清理不再使用的代理，避免资源泄露
- 错误处理: 对所有API调用进行适当的异常处理
- 性能监控: 监控请求延迟和错误率，优化连接参数
- 安全实践: 妥善管理认证令牌，使用环境变量存储敏感信息
配置建议
# 推荐的生产环境配置
production_config = {
"base_url": "https://api.your-letta-instance.com",
"timeout": 30, # 请求超时时间
"max_retries": 3, # 最大重试次数
"retry_delay": 1, # 重试延迟(秒)
}
常见问题解答
Q: 如何处理代理的内存溢出？
A: 定期清理归档记忆，优化核心内存内容，使用适当的模型配置。
Q: 如何提高消息处理速度？
A: 使用流式响应、批量处理消息、优化网络连接。
Q: 如何保证对话的连续性？
A: 合理使用内存块，确保重要信息被正确存储和检索。
通过本教程，您应该已经掌握了MemGPT Python SDK的核心功能和高级用法。MemGPT的强大内存管理能力为构建智能对话系统提供了全新的可能性，期待看到您基于此开发的创新应用！
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
5
0- 0
已为社区贡献28条内容
所有评论(0)