Photo by
ovestdesign
LangChainによるGenerative Agents: 生成エージェントの実装 その１
LangChain0.0.42で実装された、TimeWeightedVectorStoreRetrieverのチュートリアルの前半部分をgoogl colabで試してみました。行動を性格に反映することで、より生き生きとした会話が楽しめるようになっています。設定やストーリーを色々かえて試してみると楽しそうです。お試しあれ
# ライブラリーのインストール
!pip install openai > /dev/null
!pip install langchain > /dev/null
!pip install faiss-cpu > /dev/null
!pip install termcolor > /dev/null
!pip install tiktoken > /dev/null
import os
os.environ["OPENAI_API_KEY"] = "YOUR OPENAI_API_KEY"USER_NAME = "hamachi" # エージェントにインタビューするときに使う名前
LLM = ChatOpenAI(max_tokens=1500, model="gpt-4") # Can be any LLM you want.Generative Agentのメモリ コンポーネント
生成エージェントの記憶とその動作への影響について説明します。メモリは、次の 2 つの点で標準の LangChain チャット メモリとは異なります。
１．メモリ形成
観察 - 仮想世界との対話、相互作用から、自己または他者について観察
振り返り - 核となる記憶を再浮上させ、要約する
２．メモリの取得
メモリは、重要度、新しさ、重要度の加重合計を使用して取得されます。
メモリのライフサイクル
エージェントによる観察時：
言語モデルは、記憶の重要性をスコア付けします (平凡な場合は 1、心に訴えるものは 10)
観察結果と重要性は、TimeWeightedVectorStoreRetriever によってlast_accessed_timeを使ってドキュメント内に保存される
エージェントが観察に応答する場合：
顕著性、新しさ、および重要性に基づいてドキュメントをフェッチするレトリバー用のクエリを生成
取得した情報を要約
使用したドキュメントのlast_accessed_timeを更新。
class GenerativeAgent(BaseModel):
"""記憶力と生来の性格を持つキャラクター"""
name: str
age: int
traits: str
"""変えたくないキャラクターの特徴"""
status: str
"""Current activities of the character."""
llm: BaseLanguageModel
memory_retriever: TimeWeightedVectorStoreRetriever
"""キャラクターの現在の活動状況"""
verbose: bool = False
reflection_threshold: Optional[float] = None
"""記憶の「importance:重要度」の合計が上記の閾値を超えたら、立ち止まって振り返る。"""
current_plan: List[str] = []
"""エージェントの現在の方針"""
# メタ情報
summary: str = "" #: :meta private:
summary_refresh_seconds: int= 3600 #: :meta private:
last_refreshed: datetime =Field(default_factory=datetime.now) #: :meta private:
daily_summaries: List[str] #: :meta private:
memory_importance: float = 0.0 #: :meta private:
max_tokens_limit: int = 1200 #: :meta private:
class Config:
"""Configuration for this pydantic object."""
arbitrary_types_allowed = True
@staticmethod
def _parse_list(text: str) -> List[str]:
"""改行で区切られた文字列をリストに分割"""
lines = re.split(r'\n', text.strip())
return [re.sub(r'^\s*\d+\.\s*', '', line).strip() for line in lines]
def _compute_agent_summary(self):
""""""
prompt = PromptTemplate.from_template(
"How would you summarize {name}'s core characteristics given the"
+" following statements:\n"
+"{related_memories}"
+ "Do not embellish."
+"\n\nSummary: "
)
# 登場人物の核となる特徴を考察
relevant_memories = self.fetch_memories(f"{self.name}'s core characteristics")
relevant_memories_str = "\n".join([f"{mem.page_content}" for mem in relevant_memories])
chain = LLMChain(llm=self.llm, prompt=prompt, verbose=self.verbose)
return chain.run(name=self.name, related_memories=relevant_memories_str).strip()
def _get_topics_of_reflection(self, last_k: int = 50) -> Tuple[str, str, str]:
"""Return the 3 most salient high-level questions about recent observations."""
prompt = PromptTemplate.from_template(
"{observations}\n\n"
+ "Given only the information above, what are the 3 most salient in Japanese"
+ " high-level questions we can answer about the subjects in the statements?"
+ " Provide each question on a new line.\n\n"
)
reflection_chain = LLMChain(llm=self.llm, prompt=prompt, verbose=self.verbose)
observations = self.memory_retriever.memory_stream[-last_k:]
observation_str = "\n".join([o.page_content for o in observations])
result = reflection_chain.run(observations=observation_str)
return self._parse_list(result)
def _get_insights_on_topic(self, topic: str) -> List[str]:
"""Generate 'insights' on a topic of reflection, based on pertinent memories."""
prompt = PromptTemplate.from_template(
"Statements about {topic}\n"
+"{related_statements}\n\n"
+ "What 5 high-level insights can you infer from the above statements?"
+ " (example format: insight in Japanese(because of 1, 5, 3))"
)
related_memories = self.fetch_memories(topic)
related_statements = "\n".join([f"{i+1}. {memory.page_content}"
for i, memory in
enumerate(related_memories)])
reflection_chain = LLMChain(llm=self.llm, prompt=prompt, verbose=self.verbose)
result = reflection_chain.run(topic=topic, related_statements=related_statements)
# TODO: Parse the connections between memories and insights
return self._parse_list(result)
def pause_to_reflect(self) -> List[str]:
"""Reflect on recent observations and generate 'insights'."""
print(colored(f"Character {self.name} is reflecting", "blue"))
new_insights = []
topics = self._get_topics_of_reflection()
for topic in topics:
insights = self._get_insights_on_topic( topic)
for insight in insights:
self.add_memory(insight)
new_insights.extend(insights)
return new_insights
def _score_memory_importance(self, memory_content: str, weight: float = 0.15) -> float:
"""Score the absolute importance of the given memory."""
# A weight of 0.25 makes this less important than it
# would be otherwise, relative to salience and time
prompt = PromptTemplate.from_template(
"On the scale of 1 to 10, where 1 is purely mundane"
+" (e.g., 歯磨き, ベッドメーキング) and 10 is"
+ " extremely poignant (e.g., 別れ話, 大学入学試験の合否"
+ " ), rate the likely poignancy of the"
+ " following piece of memory. Respond with a single integer."
+ "\nMemory: {memory_content}"
+ "\nRating: "
)
chain = LLMChain(llm=self.llm, prompt=prompt, verbose=self.verbose)
score = chain.run(memory_content=memory_content).strip()
match = re.search(r"^\D*(\d+)", score)
if match:
return (float(score[0]) / 10) * weight
else:
return 0.0
def add_memory(self, memory_content: str) -> List[str]:
"""Add an observation or memory to the agent's memory."""
importance_score = self._score_memory_importance(memory_content)
self.memory_importance += importance_score
document = Document(page_content=memory_content, metadata={"importance": importance_score})
result = self.memory_retriever.add_documents([document])
# After an agent has processed a certain amount of memories (as measured by
# aggregate importance), it is time to reflect on recent events to add
# more synthesized memories to the agent's memory stream.
if (self.reflection_threshold is not None
and self.memory_importance > self.reflection_threshold
and self.status != "Reflecting"):
old_status = self.status
self.status = "Reflecting"
self.pause_to_reflect()
# Hack to clear the importance from reflection
self.memory_importance = 0.0
self.status = old_status
return result
def fetch_memories(self, observation: str) -> List[Document]:
"""Fetch related memories."""
return self.memory_retriever.get_relevant_documents(observation)
def get_summary(self, force_refresh: bool = False) -> str:
"""Return a descriptive summary of the agent."""
current_time = datetime.now()
since_refresh = (current_time - self.last_refreshed).seconds
if not self.summary or since_refresh >= self.summary_refresh_seconds or force_refresh:
self.summary = self._compute_agent_summary()
self.last_refreshed = current_time
return (
f"Name: {self.name} (age: {self.age})"
+f"\nInnate traits: {self.traits}"
+f"\n{self.summary}"
)
def get_full_header(self, force_refresh: bool = False) -> str:
"""Return a full header of the agent's status, summary, and current time."""
summary = self.get_summary(force_refresh=force_refresh)
current_time_str = datetime.now().strftime("%B %d, %Y, %I:%M %p")
return f"{summary}\nIt is {current_time_str}.\n{self.name}'s status: {self.status}"
def _get_entity_from_observation(self, observation: str) -> str:
prompt = PromptTemplate.from_template(
"What is the observed entity in the following observation? {observation}"
+"\nEntity="
)
chain = LLMChain(llm=self.llm, prompt=prompt, verbose=self.verbose)
return chain.run(observation=observation).strip()
def _get_entity_action(self, observation: str, entity_name: str) -> str:
prompt = PromptTemplate.from_template(
"What is the {entity} doing in the following observation? {observation}"
+"\nThe {entity} is"
)
chain = LLMChain(llm=self.llm, prompt=prompt, verbose=self.verbose)
return chain.run(entity=entity_name, observation=observation).strip()
def _format_memories_to_summarize(self, relevant_memories: List[Document]) -> str:
content_strs = set()
content = []
for mem in relevant_memories:
if mem.page_content in content_strs:
continue
content_strs.add(mem.page_content)
created_time = mem.metadata["created_at"].strftime("%B %d, %Y, %I:%M %p")
content.append(f"- {created_time}: {mem.page_content.strip()}")
return "\n".join([f"{mem}" for mem in content])
def summarize_related_memories(self, observation: str) -> str:
"""Summarize memories that are most relevant to an observation."""
entity_name = self._get_entity_from_observation(observation)
entity_action = self._get_entity_action(observation, entity_name)
q1 = f"What is the relationship between {self.name} and {entity_name} in Japanese"
relevant_memories = self.fetch_memories(q1) # Fetch memories related to the agent's relationship with the entity
q2 = f"{entity_name} is {entity_action}"
relevant_memories += self.fetch_memories(q2) # Fetch things related to the entity-action pair
context_str = self._format_memories_to_summarize(relevant_memories)
prompt = PromptTemplate.from_template(
"{q1}?\nContext from memory:\n{context_str}\nRelevant context: "
)
chain = LLMChain(llm=self.llm, prompt=prompt, verbose=self.verbose)
return chain.run(q1=q1, context_str=context_str.strip()).strip()
def _get_memories_until_limit(self, consumed_tokens: int) -> str:
"""Reduce the number of tokens in the documents."""
result = []
for doc in self.memory_retriever.memory_stream[::-1]:
if consumed_tokens >= self.max_tokens_limit:
break
consumed_tokens += self.llm.get_num_tokens(doc.page_content)
if consumed_tokens < self.max_tokens_limit:
result.append(doc.page_content)
return "; ".join(result[::-1])
def _generate_reaction(
self,
observation: str,
suffix: str
) -> str:
"""React to a given observation."""
prompt = PromptTemplate.from_template(
"{agent_summary_description}"
+"\nIt is {current_time}."
+"\n{agent_name}'s status: {agent_status}"
+ "\nSummary of relevant context from {agent_name}'s memory:"
+"\n{relevant_memories}"
+"\nMost recent observations: {recent_observations}"
+ "\nObservation: {observation}"
+ "\n\n" + suffix
)
agent_summary_description = self.get_summary()
relevant_memories_str = self.summarize_related_memories(observation)
current_time_str = datetime.now().strftime("%B %d, %Y, %I:%M %p")
kwargs = dict(agent_summary_description=agent_summary_description,
current_time=current_time_str,
relevant_memories=relevant_memories_str,
agent_name=self.name,
observation=observation,
agent_status=self.status)
consumed_tokens = self.llm.get_num_tokens(prompt.format(recent_observations="", **kwargs))
kwargs["recent_observations"] = self._get_memories_until_limit(consumed_tokens)
action_prediction_chain = LLMChain(llm=self.llm, prompt=prompt)
result = action_prediction_chain.run(**kwargs)
return result.strip()
def generate_reaction(self, observation: str) -> Tuple[bool, str]:
"""React to a given observation."""
call_to_action_template = (
"Should {agent_name} react to the observation, and if so,"
+" what would be an appropriate reaction? Respond in one line."
+' If the action is to engage in dialogue, write:\nSAY: "what to say"'
+"\notherwise, write:\nREACT: {agent_name}'s reaction (if anything)."
+ "\nEither do nothing, react, or say something but not both.\n\n"
)
full_result = self._generate_reaction(observation, call_to_action_template)
result = full_result.strip().split('\n')[0]
self.add_memory(f"{self.name} observed {observation} and reacted by {result}")
if "REACT:" in result:
reaction = result.split("REACT:")[-1].strip()
return False, f"{self.name} {reaction}"
if "SAY:" in result:
said_value = result.split("SAY:")[-1].strip()
return True, f"{self.name} said {said_value}"
else:
return False, result
def generate_dialogue_response(self, observation: str) -> Tuple[bool, str]:
"""React to a given observation."""
# 会話を終わらせる場合は GOODBYE：会話を続ける場合は： SAY: "what to say next"（次に何を言うか）
call_to_action_template = (
'What would {agent_name} say? To end the conversation, write: GOODBYE: "what to say". Otherwise to continue the conversation, write: SAY: "what to say next"\n\n'
)
full_result = self._generate_reaction(observation, call_to_action_template)
result = full_result.strip().split('\n')[0]
if "GOODBYE:" in result:
farewell = result.split("GOODBYE:")[-1].strip()
self.add_memory(f"{self.name} observed {observation} and said {farewell}")
return False, f"{self.name} said {farewell}"
if "SAY:" in result:
response_text = result.split("SAY:")[-1].strip()
self.add_memory(f"{self.name} observed {observation} and said {response_text}")
return True, f"{self.name} said {response_text}"
else:
return False, result生成エージェントを作成
import math
import faiss
def relevance_score_fn(score: float) -> float:
"""Return a similarity score on a scale [0, 1]."""
# This will differ depending on a few things:
# - the distance / similarity metric used by the VectorStore
# - the scale of your embeddings (OpenAI's are unit norm. Many others are not!)
# This function converts the euclidean norm of normalized embeddings
# (0 is most similar, sqrt(2) most dissimilar)
# to a similarity function (0 to 1)
return 1.0 - score / math.sqrt(2)
def create_new_memory_retriever():
"""Create a new vector store retriever unique to the agent."""
# Define your embedding model
embeddings_model = OpenAIEmbeddings()
# Initialize the vectorstore as empty
embedding_size = 1536
index = faiss.IndexFlatL2(embedding_size)
vectorstore = FAISS(embeddings_model.embed_query, index, InMemoryDocstore({}), {}, relevance_score_fn=relevance_score_fn)
return TimeWeightedVectorStoreRetriever(vectorstore=vectorstore, other_score_keys=["importance"], k=15) taroh = GenerativeAgent(name="Taroh",
age=25,
traits="心配性, デザインが好き", # You can add more persistent traits here
status="就職活動中", # When connected to a virtual world, we can have the characters update their status
memory_retriever=create_new_memory_retriever(),
llm=LLM,
daily_summaries = [
"新しい街に県をまたいで車でやってきたが、まだ仕事がない。."
],
reflection_threshold = 8, # we will give this a relatively low number to show how reflection works
)print(taroh.get_summary())Name: Taroh (age: 25)
Innate traits: 心配性, デザインが好き
Taroh is a straightforward individual who does not engage in exaggeration or embellishment.
太郎のメモリーに直接に書き込む
# We can give the character memories directly
taroh_memories = [
"Tarohは、子供の頃に飼っていた犬のハチを覚えています",
"Tarohは、長距離運転で疲れている",
"Tarohは、新居を見つけた",
"隣人は、猫を飼っている",
"夜は、道がうるさい",
"Tarohは、空腹だ",
"Tarohは、休息をとろうとしている",
]
for memory in taroh_memories:
taroh.add_memory(memory)print(taroh.get_summary(force_refresh=True))Name: Taroh (age: 25)
Innate traits: 心配性, デザインが好き
Taroh is a person who recently found a new home, remembers his childhood dog Hachi, is trying to take a rest, is hungry, is tired from long-distance driving, finds the noise of the road bothersome at night, and has a neighbor who owns a cat.
太郎は、最近新しい家を見つけた人、幼い頃の愛犬ハチを覚えている人、休もうとしている人、お腹が空いている人、長距離運転で疲れている人、夜中に道路の音が気になる人、隣人に猫を飼っている人などです。
事前インタビュー
def interview_agent(agent: GenerativeAgent, message: str) -> str:
"""Help the notebook user interact with the agent."""
new_message = f"{USER_NAME} says {message}"
return agent.generate_dialogue_response(new_message)[1]interview_agent(taroh, "'好きなこと'を教えてください")Taroh said 僕はデザインが好きです。特にグラフィックデザインやインテリアデザインに興味があります。あなたは何が好きですか？
interview_agent(taroh, "いま、一番気になることは何ですか")Taroh said 僕はデザインが好きです。特にグラフィックデザインやインテリアデザインに興味があります。あなたは何が好きですか？
interview_agent(taroh, "いま、一番気になっていることは何ですか")Taroh said いま一番気になることは、就職活動です。いい仕事を見つけることができるか不安ですが、がんばっています。あなたは何か気になることがありますか？
それでは、太郎さんに一日の生活を始めてもらいましょう。
observations = [
"Tarohは、窓の外の騒がしい工事現場の音で目を覚ました。",
"Tarohは、ベッドから起き上がり,コーヒーを淹れに台所へ向かいます",
"Tarohは、コーヒーのフィルターを買い忘れたことに気づき、引っ越しの段ボール箱をあさって探し始める。",
"Tarohは、やっとフィルターを見つけ、コーヒーを淹れる。",
"コーヒーの味は苦く、Tarohは、もっと良い銘柄を買わなかったことを後悔する。",
"Tarohは、メールをチェックし,まだ仕事の依頼がないことを確認する。",
"Tarohは、履歴書とカバーレターの更新に時間を費やす。",
"Tarohは、街に出て、求人情報を探すことにする。",
"Tarohは、就職説明会の看板を見て,参加することにした。",
"Tarohは、就職説明会に参加することにした。",
"就職説明会場に入場するための列は長く,Tarohは、1時間待つことになる",
"Tarohは、就職説明会で何人かの雇用主候補に会うが,内定はもらえない",
"Tarohは、がっかりした気持ちで就職説明会場を後にする。",
"Tarohは、昼食を取るために地元の食堂に立ち寄ります。",
"食堂のサービスが遅く,Tarohは食事にありつくのに30分も待たされた。",
"Tarohは、隣のテーブルで求人の会話をしているのを耳にする。",
"Tarohは、隣のテーブルの人に求人について尋ね、その会社についての情報を得る。",
"Tarohは、その仕事に応募することを決め、履歴書とカバーレターを送る。",
"Tarohは、求人情報の検索を続け、いくつかの地元の企業に履歴書を郵送する",
"Tarohは、仕事探しの合間を縫って,近くの公園に散歩に行く",
"犬が近づいてきてTarohの足を舐めるので、数分間撫でてやる",
"Tarohは、フリスビーで遊んでいるグループを見つけて、一緒に遊ぶことにした。",
"Tarohは、フリスビーで遊んでいる",
"Tarohは、フリスビーで楽しく遊ぶが,フリスビーに顔をぶつけて,鼻をけがしてしまう。",
"Tarohは、フリスビーで遊んでいる人たちと友達になる。",
"Tarohは、自分のアパートに戻って少し休む。",
"カラスがアパートの外のゴミ袋を破ってしまい,ゴミが床一面に散らばっている。",
"Tarohは、就職活動に焦りを感じ始める",
"Tarohは、親友に苦悩を吐き出すために電話をかける。",
"Tarohの友人は、励ましの言葉をかけ、努力を続けるように言う。",
"Tarohは、友人と話した後、少し気分が良くなった。",
]# それでは、タローを送り出して、その変化を観察します。
for i, observation in enumerate(observations):
_, reaction = taroh.generate_reaction(observation)
print(colored(observation, "green"), reaction)
if ((i+1) % 20) == 0:
print('*'*40)
print(colored(f"After {i+1} observations, Taroh's summary is:\n{taroh.get_summary(force_refresh=True)}", "blue"))
print('*'*40)Tarohは、窓の外の騒がしい工事現場の音で目を覚ました。
Taroh Taroh sighs and rubs his eyes, feeling frustrated by the noisy construction site.
Tarohは、ベッドから起き上がり,コーヒーを淹れに台所へ向かいます
Taroh Taroh stretches and heads to the kitchen to make coffee, hoping it will help wake him up.
（中略）
****************************************
After 20 observations, Taroh's summary is:
Name: Taroh (age: 25)
Innate traits: 心配性, デザインが好き
タローは決意と希望に満ちた集中力のある人物で、積極的に就職活動をしています。就職活動中は、興奮、失望、心配、モチベーションなど、さまざまな感情を経験する。就職フェアに参加し、履歴書やカバーレターを更新し、さまざまな方法で就職先を探すなど、率先して行動しています。また、公園を散歩するなど、就職活動中は休憩を取り、リラックスする時間を大切にしています。
****************************************
（以下略）
一日を終えてのインタビュー
interview_agent(taroh, "あれ、鼻を怪我していませんか？")Taroh said "はい、実はフリスビーで遊んでいて、うっかり顔にぶつけて鼻をけがしてしまいました。ちょっと痛いですが、大丈夫です。ありがとうございます！"
interview_agent(taroh, "今日はどんな一日だったか教えてください")Taroh said "今日は結構忙しい一日でした。求人情報を検索して、いくつかの企業に履歴書を送りました。そして、ちょっと休憩して公園で散歩したり、フリスビーで遊んだりしました。でも、フリスビーで顔をけがしてしまってちょっと恥ずかしかったです。その後、アパートに戻って休んで、親友と電話で話しました。彼から励ましの言葉をもらって、気分が良くなりました。あなたはどんな一日でしたか？"
interview_agent(taroh, "就職活動の進捗はどうですか?")Taroh said "就職活動の進捗はまだまだですが、前向きに取り組んでいます。最近は、求人情報を調べたり、履歴書やカバーレターを修正したりしています。また、ジョブフェアに参加して、様々な企業と出会い、自分の適性を見極めることも大切だと感じています。焦らず、自分に合った仕事が見つかるよう努力しています。あなたはどのような仕事をしていますか？"
次回の予定
生成エージェント１、インタビュー１の簡単な構成で例題通り動作を見てみました。プロンプトだけでロールプレイをしてもらうよりも、より生き生きとした感じの会話になっているのではないでしょうか。
次回は、生成エージェントを複数人にして、設定も工夫しながら楽しんでみたいと思います。
この記事が参加している募集
この記事が気に入ったらサポートをしてみませんか？