源码级别解析 · 源码深度解析 · 多智能体行为模拟框架
2026-04-11 | 每日技术深度解读
基于 arXiv:2304.03442 研究论文
模块化设计，支持扩展与定制
清晰的分层架构设计
class ReverieServer:
def __init__(self, fork_sim_code, sim_code):
# 叉式模拟：从已有模拟分支
self.fork_sim_code = fork_sim_code
fork_folder = f"{fs_storage}/{self.fork_sim_code}"
# 当前模拟代码
self.sim_code = sim_code
sim_folder = f"{fs_storage}/{self.sim_code}"
copyanything(fork_folder, sim_folder)支持模拟的分支与继承机制
真实时间与游戏时间的映射关系
# 每步推进游戏时间
self.curr_time = self.curr_time + datetime.timedelta(
seconds=self.sec_per_step)
# 更新步数
self.step += 1
# 保存模拟状态
self.save_simulation()基于时间步长的推进逻辑
每个智能体具有完整的认知能力
短期与长期记忆的分层管理
class Persona:
def __init__(self, name, folder_mem_saved=False):
# 身份标识
self.name = name
# 空间记忆 - 树形结构
f_s_mem_saved = f"{folder_mem_saved}/bootstrap_memory/spatial_memory.json"
self.s_mem = MemoryTree(f_s_mem_saved)
# 关联记忆 - 语义网络
f_a_mem_saved = f"{folder_mem_saved}/bootstrap_memory/associative_memory"
self.a_mem = AssociativeMemory(f_a_mem_saved)加载预训练的记忆数据
智能体对物理世界的理解
"""
空间记忆结构示例:
{
"double studio": {
"double studio": {
"bedroom 2": ["painting", "easel", "closet", "bed"]
}
}
}
"""
class MemoryTree:
def __init__(self, saved_file):
# 从JSON文件加载空间记忆
self.tree = json.load(open(saved_file))层次化的空间表示
基于语义关联的记忆检索
class ConceptNode:
def __init__(self, node_id, type, created, expiration,
s, p, o, description, embedding_key,
poignancy, keywords, filling):
self.node_id = node_id
self.type = type # event/thought/chat
self.created = created
self.expiration = expiration
self.subject = s
self.predicate = p
self.object = o
self.description = description
self.embedding_key = embedding_key
self.poignancy = poignancy # 情绪权重
self.keywords = keywords
self.filling = filling完整的语义表示单元
结合语义与上下文的检索策略
快速访问的工作内存
"""
工作记忆JSON结构:
{
"curr_time": "2023-06-25 10:30:00",
"curr_location": "bedroom 2",
"last_action": "move_to_bathroom",
"plan": "brush_teeth",
"dialogue_context": []
}
"""
class Scratch:
def __init__(self, saved_file):
self.data = json.load(open(saved_file))轻量级的短期记忆管理
完整的人类认知模拟
智能化的环境感知
def perceive(self, maze):
"""
感知周围事件，受两个关键参数控制:
1. att_bandwidth: 注意力带宽，感知事件数量限制
2. retention: 保留时间，避免重复感知
"""
# 获取视野内的事件
nearby_events = maze.get_nearby_events(self, vision_radius)
# 根据注意力带宽筛选
if len(nearby_events) > self.att_bandwidth:
nearby_events = sorted(nearby_events,
key=lambda e: e.distance_to_agent)[0:self.att_bandwidth]
# 过滤已感知的事件
new_events = [e for e in nearby_events
if e not in self.recently_perceived]
return new_events基于注意力机制的智能感知
智能的记忆关联检索
def retrieve(self, perceived):
"""
对感知到的相关事件和思想进行检索
返回字典: {event: {"curr_event": ..., "events": ..., "thoughts": ...}}
"""
retrieved = {}
for event in perceived:
# 基于嵌入向量相似度检索
similar_events = self.a_mem.find_similar_events(
event.embedding, top_k=10)
# 基于关键词检索
keyword_events = self.a_mem.find_by_keywords(
event.keywords, top_k=5)
# 基于时间范围检索
time_events = self.a_mem.find_in_time_range(
event.time - timedelta(hours=1),
event.time + timedelta(hours=1))
retrieved[event] = {
"curr_event": event,
"events": similar_events + keyword_events + time_events,
"thoughts": self.a_mem.find_related_thoughts(event)
}
return retrieved多维度的记忆关联机制
基于上下文的动态规划
复杂的决策制定过程
def plan(self, maze, personas, new_day, retrieved):
"""
主要认知函数，制定短期和长期计划
"""
if new_day:
# 新的一天：制定长期目标
self.create_daily_goals(personas)
# 基于感知和检索的记忆制定短期计划
short_term_plan = []
for event, context in retrieved.items():
# 分析当前情境
situation_analysis = self.analyze_situation(event, context)
# 制定响应计划
response_plan = self.generate_response_plan(
situation_analysis, context)
short_term_plan.extend(response_plan)
return short_term_plan上下文感知的智能规划
持续学习的反思机制
将计划转化为具体行动
def execute(self, action):
"""执行预定义的行动"""
if action["type"] == "move":
# 移动到指定位置
target_location = action["target"]
self.move_to_location(target_location)
# 更新空间记忆
self.update_spatial_memory(target_location)
elif action["type"] == "interact":
# 与物体交互
target_object = action["target"]
interaction_result = self.interact_with_object(target_object)
# 记录交互事件
self.record_interaction(interaction_result)
elif action["type"] == "converse":
# 与其他智能体对话
target_persona = action["target"]
conversation = self.generate_conversation(target_persona)
self.converse(target_persona, conversation)完整的行动执行链
自然的智能体间对话
完整的虚拟世界模拟
class Maze:
def __init__(self, maze_name):
self.name = maze_name
self.matrix = self.load_matrix(maze_name)
self.objects = self.load_objects(maze_name)
self.visuals = self.load_visuals(maze_name)
def load_matrix(self, maze_name):
"""加载迷宫矩阵数据"""
matrix_path = f"{maze_assets_loc}/{maze_name}/matrix"
return json.load(open(matrix_path))
def get_nearby_events(self, persona, radius):
"""获取角色周围的事件"""
events = []
for obj in self.objects:
if self.distance(persona.location, obj.location) <= radius:
events.append(obj)
return events环境感知与管理
复杂的智能体社会模拟
可靠的状态管理
def save(self, save_folder):
"""保存智能体当前状态"""
# 保存空间记忆（JSON格式）
f_s_mem = f"{save_folder}/spatial_memory.json"
self.s_mem.save(f_s_mem)
# 保存关联记忆（CSV格式）
f_a_mem = f"{save_folder}/associative_memory"
self.a_mem.save(f_a_mem)
# 保存工作记忆（JSON格式）
f_scratch = f"{save_folder}/scratch.json"
self.scratch.save(f_scratch)分层的数据保存策略
直观的用户体验
灵活的接口设计
灵活的配置管理
高效的运行性能
支持灵活扩展
全面的测试体系
企业级部署方案
广泛的应用前景
需要持续改进的方面
持续的技术创新
欢迎社区参与
丰富的学习资源
深远的技术与社会影响
感谢阅读！
访问 https://atcfu.com/ai-articles/generative-agents/ 回顾本文