Last updated
class MemoryStream:
def __init__(self):
self.memories = [] # 按时间顺序存储
def add_memory(
self,
description: str,
timestamp: datetime,
importance: float # 0-10 的重要性评分
):
memory = {
"description": description,
"timestamp": timestamp,
"importance": importance,
"embedding": embed(description),
"access_count": 0,
"last_access": timestamp
}
self.memories.append(memory)def retrieve_memories(
self,
query: str,
k: int = 10
) -> List[Memory]:
query_embedding = embed(query)
scored_memories = []
for memory in self.memories:
# 相关性分数
relevance = cosine_similarity(query_embedding, memory["embedding"])
# 近因性分数（指数衰减）
hours_passed = (now() - memory["timestamp"]).total_seconds() / 3600
recency = 0.99 ** hours_passed
# 重要性分数（归一化）
importance = memory["importance"] / 10
# 综合评分
score = relevance + recency + importance
scored_memories.append((memory, score))
# 返回 top-k
scored_memories.sort(key=lambda x: x[1], reverse=True)
return [m for m, s in scored_memories[:k]]async def reflect(self) -> List[str]:
# 检索最近的重要记忆
recent_memories = self.retrieve_memories(
query="最近发生了什么重要的事？",
k=100
)
# 提取关键问题
questions_prompt = f"""
基于以下记忆，提出 3 个值得深入思考的问题：
记忆：
{format_memories(recent_memories)}
"""
questions = await llm.generate(questions_prompt)
# 对每个问题进行反思
insights = []
for question in questions:
relevant_memories = self.retrieve_memories(question, k=20)
insight_prompt = f"""
问题：{question}
相关记忆：
{format_memories(relevant_memories)}
基于这些记忆，你能得出什么结论或洞察？
"""
insight = await llm.generate(insight_prompt)
# 将洞察作为新记忆存储（高重要性）
self.add_memory(
description=f"反思：{insight}",
timestamp=now(),
importance=8 # 反思通常很重要
)
insights.append(insight)
return insights记忆片段：
- 早上和 Maria 在咖啡馆聊天
- Maria 提到她最近很忙
- 下午看到 Maria 独自在公园散步，看起来很累
反思结论：
"Maria 最近压力很大，可能需要朋友的支持。下次见面应该主动关心她。"async def create_daily_plan(self, agent: Agent) -> List[PlanItem]:
# 获取智能体的背景信息
persona = agent.persona
plan_prompt = f"""
角色：{persona.name}
职业：{persona.occupation}
今日日期：{today()}
{persona.name} 的典型日程是什么样的？
请用简洁的方式列出今天从早到晚的活动计划。
格式：
时间 | 地点 | 活动
"""
raw_plan = await llm.generate(plan_prompt)
return parse_plan(raw_plan)高层计划（一天）：
08:00 - 起床、洗漱
09:00 - 去咖啡馆工作
12:00 - 午餐
...
中层计划（一小时）：
09:00 - 到达咖啡馆
09:10 - 点咖啡
09:20 - 打开笔记本开始写作
...
低层行动（即时）：
走向咖啡馆入口 → 开门 → 走到柜台 → 点单 → ...async def should_initiate_conversation(
agent1: Agent,
agent2: Agent,
context: str
) -> bool:
# 检索与对方相关的记忆
memories_about = agent1.retrieve_memories(
query=f"关于 {agent2.name} 的记忆",
k=5
)
prompt = f"""
{agent1.name} 刚刚看到了 {agent2.name}。
关于 {agent2.name} 的记忆：
{format_memories(memories_about)}
当前情境：{context}
{agent1.name} 应该主动打招呼吗？回答 Yes 或 No，并说明理由。
"""
response = await llm.generate(prompt)
return "yes" in response.lower()async def generate_dialogue(
speaker: Agent,
listener: Agent,
conversation_history: List[str]
) -> str:
# 获取相关记忆
relevant_memories = speaker.retrieve_memories(
query=f"与 {listener.name} 的对话",
k=10
)
prompt = f"""
说话者：{speaker.name} ({speaker.persona.traits})
听者：{listener.name}
相关记忆：
{format_memories(relevant_memories)}
对话历史：
{format_history(conversation_history)}
{speaker.name} 接下来会说什么？请保持符合角色性格。
"""
return await llm.generate(prompt)