はじめに
まずはAI処理を作る前に、ターンを処理する土台をつくる
Source
ai_agent/domain/repository/conversation.py
from datetime import datetime
from ai_agent.models import Entity, ActionTimeline, Message
class ConversationRepository:
@staticmethod
def get_all_entities():
return Entity.objects.all()
@staticmethod
def get_timelines_ordered_by_next_turn():
return ActionTimeline.objects.order_by("next_turn")
@staticmethod
def update_or_create_action_timeline(entity, defaults):
return ActionTimeline.objects.update_or_create(entity=entity, defaults=defaults)
@staticmethod
def update_next_turn(action_timeline, increment):
action_timeline.next_turn += increment
action_timeline.save()
@staticmethod
def create_message(entity, content):
return Message.objects.create(
entity=entity,
message_content=content,
created_at=datetime.now(),
)
このソースコードが実現する処理
ConversationRepository クラスは、エンティティ (Entity)、アクションタイムライン (ActionTimeline)、メッセージ (Message) の3つのモデルを操作するためのリポジトリクラスです。このクラスを利用することで、以下のような処理が可能になります。
- すべてのエンティティの取得
- ConversationRepository.get_all_entities()
- Entity モデルの全レコードを取得する
- 会話に関与するすべてのエンティティ（ユーザーやAIエージェントなど）を一覧で取得できる
- アクションタイムラインの取得
- ConversationRepository.get_timelines_ordered_by_next_turn()
- ActionTimeline モデルのレコードを next_turn の昇順で取得する
- これにより、次に発話するエンティティを決定するためのデータが得られる
- アクションタイムラインの更新または作成
- ConversationRepository.update_or_create_action_timeline(entity, defaults)
- 指定した entity に紐づく ActionTimeline を更新、または存在しない場合は作成する
- defaults には更新するフィールドの値を渡す
- アクションタイムラインの next_turn の更新
- ConversationRepository.update_next_turn(action_timeline, increment)
- action_timeline.next_turn を increment 分だけ増加させ、保存する
- これにより、エンティティの次の発話タイミングを調整できる
- メッセージの作成
- ConversationRepository.create_message(entity, content)
- 指定した entity に対して、新しいメッセージ (Message) を作成する
- メッセージの作成時刻 (created_at) は datetime.now() で設定される
この処理が活用されるケース
- エージェントの発話順管理: ActionTimeline を用いることで、複数のエージェントが会話の流れを制御する仕組みを実装できる
- 会話履歴の管理: Message を作成することで、エンティティごとの発話履歴を保存し、後から参照できるようにする
- 動的な会話システムの実装: update_or_create_action_timeline により、エージェントの状態を動的に変更しながら会話を進行できる
ai_agent/domain/service/conversation.py
from ai_agent.domain.repository.conversation import ConversationRepository
from ai_agent.domain.service.googlemaps_review import GoogleMapsReviewService
from ai_agent.domain.service.ng_word import NGWordService
from ai_agent.domain.service.rag import RagService
from ai_agent.domain.valueobject.conversation import EntityVO
from ai_agent.models import Entity, ActionHistory
class ConversationService:
@staticmethod
def calculate_next_turn_increment(speed: float) -> float:
"""
Calculate the increment for next_turn based on entity speed.
This ensures consistency in how the increment is derived, and makes future
adjustments easier.
"""
return 1 / speed
@staticmethod
def initialize_timeline():
"""
Initialize the timeline by assigning first turn values based on entity speed.
"""
entities = ConversationRepository.get_all_entities()
for entity in entities:
ConversationRepository.update_or_create_action_timeline(
entity=entity,
defaults={
"next_turn": ConversationService.calculate_next_turn_increment(
entity.speed
)
},
)
@staticmethod
def get_next_entity(input_text: str):
"""
Get the next entity to act based on the timeline, considering
its ability to act (`can_act`) determined by `think`.
Args:
input_text (str): The input text for the entity's `think` process.
Returns:
Entity: The next entity that should act.
Raises:
ValueError: If no entities are available to act.
"""
timelines = ConversationRepository.get_timelines_ordered_by_next_turn()
if not timelines.exists():
raise ValueError("No entities available in the timeline.")
candidates = []
for timeline in timelines:
timeline.can_act = ConversationService.think(timeline.entity, input_text)
timeline.save()
if timeline.can_act:
candidates.append(timeline)
# 次の行動順 (next_turn) 最小値のエンティティを選択する
if candidates:
next_action = min(candidates, key=lambda t: t.next_turn)
ConversationRepository.update_next_turn(
action_timeline=next_action,
increment=ConversationService.calculate_next_turn_increment(
next_action.entity.speed
),
)
return next_action.entity
# このターンでは発言可能なエンティティがいない場合、すべてのエンティティの next_turn を更新して次のターンへ進む
for timeline in timelines:
ConversationRepository.update_next_turn(
action_timeline=timeline,
increment=ConversationService.calculate_next_turn_increment(
timeline.entity.speed
),
)
raise ValueError("No entities are available to act in this turn.")
@staticmethod
def simulate_next_actions(max_steps=10) -> list[EntityVO]:
"""
Simulates the next sequence of entity actions up to 'max_steps'
and creates ActionHistory records for each action.
Args:
max_steps (int): How many actions to simulate.
Returns:
List[EntityVO]: A list of EntityVO objects containing the entity's name and the turn when they act.
"""
timelines = list(ConversationRepository.get_timelines_ordered_by_next_turn())
if not timelines:
raise ValueError("No entities available in the timeline.")
simulation = []
for i in range(1, max_steps + 1):
# 次の行動を決定 (next_turn が最小のタイムラインを選ぶ)
next_action = min(timelines, key=lambda t: t.next_turn)
# ActionHistory レコードを作成
ActionHistory.objects.create(
entity=next_action.entity,
acted_at_turn=i,
done=False,
)
# シミュレーションの結果を保存
simulation.append(
EntityVO(name=next_action.entity.name, next_turn=next_action.next_turn)
)
# 次の行動予定を仮で更新
next_action.next_turn += ConversationService.calculate_next_turn_increment(
next_action.entity.speed
)
return simulation
@staticmethod
def think(entity: Entity, input_text: str):
"""
Process the entity's thought logic to determine if it can respond.
Args:
entity (Entity): The entity performing the thought process.
input_text (str): The input text to evaluate.
Returns:
bool: True if the entity can respond, False otherwise.
"""
if entity.thinking_type == "google_maps_based":
return GoogleMapsReviewService.can_respond(input_text, entity)
elif entity.thinking_type == "rag_based":
return RagService.can_respond(input_text, entity)
elif entity.thinking_type == "ng_word_based":
return NGWordService.can_respond(input_text, entity)
# デフォルトで発言可能
return True
このソースコードが実現する処理
ConversationService クラスは、会話の流れを管理するサービス層のロジックを提供します。
具体的には、エンティティ (Entity) の行動順 (ActionTimeline) を管理し、次に発話するエンティティを決定する機能を持っています。
- エンティティの行動速度を考慮したターン管理
- ConversationService.calculate_next_turn_increment(speed: float) -> float
- エンティティの speed (行動速度) に基づいて next_turn の増加量を計算する
- speed が速いほど短い間隔で行動できる
- タイムラインの初期化
- ConversationService.initialize_timeline()
- すべてのエンティティの next_turn を、speed に基づいて計算し設定する
- 各エンティティの発話順序を初期化する
- 次に行動するエンティティの決定
- ConversationService.get_next_entity(input_text: str) -> Entity
- ActionTimeline を next_turn の昇順で取得し、最も早く発話できるエンティティを選択する
- think メソッドを用いて、エンティティが発話できるかどうかを判定する
- next_turn を更新し、発話順を維持する
- もしこのターンで誰も発話できなければ、すべてのエンティティの next_turn を更新し、次のターンへ進める
- 会話のシミュレーション
- ConversationService.simulate_next_actions(max_steps=10) -> list[EntityVO]
- max_steps 回分の会話をシミュレーションする
- next_turn が最も早いエンティティを選択し、ActionHistory に記録する
- 結果として、各エンティティの発話予定 (EntityVO のリスト) を返す
- エンティティの「思考」判定
- ConversationService.think(entity: Entity, input_text: str) -> bool
- エンティティが発話可能かどうかを判定する
- thinking_type に応じて、異なる判定ロジックを適用：
- "google_maps_based" → GoogleMapsReviewService による応答可否判定
- "rag_based" → RagService による応答可否判定
- "ng_word_based" → NGWordService による NG ワードチェック
- 上記以外の場合はデフォルトで True（発話可能）
この処理が活用されるケース
- エージェント同士の会話管理: ActionTimeline を用いて、複数のエージェントが適切な順番で発話できるようにする
- 会話の流れのシミュレーション: simulate_next_actions により、どのように会話が進行するか事前に試すことができる
- 発話可否の判定: think メソッドを活用することで、発話内容に応じたエージェントの行動制御が可能
ai_agent/domain/service/googlemaps_review.py
class GoogleMapsReviewService:
@staticmethod
def can_respond(input_text: str, entity) -> bool:
"""
Determines if the entity can respond based on Google Maps reviews.
TODO: Implement proper review-based logic.
Args:
input_text (str): The input text to evaluate.
entity (Entity): The entity performing the evaluation.
Returns:
bool: Always True for now (temporarily hardcoded for testing purposes).
"""
return True
このソースコードが実現する処理
GoogleMapsReviewService クラスは、Google Maps のレビューをもとにエンティティの発話可否を判定するためのサービスクラスです。
現在は仮置きの処理として、すべてのリクエストに対して True を返す仕様になっています。
- Google Maps のレビューをもとにした発話可否の判定
- GoogleMapsReviewService.can_respond(input_text: str, entity) -> bool
- input_text をもとに、エンティティが発話できるかどうかを判定する
- 現在は 仮実装 のため、常に True を返す
- 本実装では、Google Maps のレビューを分析し、適切な応答可否判定を行う予定
引数:
- input_text (str)：発話のトリガーとなるテキスト
- entity (Entity)：評価を行うエンティティ
戻り値:
- bool：現在はテスト用に常に True を返す
この処理が活用されるケース
- Google Maps のレビューを活用した対話:
- 将来的に、レビュー内容に基づいてエンティティの発話可否を判断できるようになる
- 対話エージェントのカスタマイズ:
- 例えば、特定のエリアのレビューに基づいてローカルな情報を発話するエージェントを作ることが可能
- RAG（Retrieval-Augmented Generation）との連携:
- 取得したレビューを検索・解析し、発話内容を動的に変更する仕組みも考えられる
今後の実装予定
Google Maps の API を活用してレビューを取得・解析し、発話の可否を決定する。
レビューのスコアやキーワード分析を行い、エージェントの発話ロジックを改善する。
ai_agent/domain/service/ng_word.py
class NGWordService:
@staticmethod
def can_respond(input_text, entity):
"""
Determine if the entity can respond based on forbidden keywords.
Args:
input_text (str): The input text to check.
entity (Entity): The entity being evaluated.
Returns:
bool: True if no forbidden keywords are detected, otherwise False.
"""
if entity.forbidden_keywords:
forbidden_list = entity.forbidden_keywords.split(",")
if any(keyword in input_text for keyword in forbidden_list):
return False
return True
このソースコードが実現する処理
NGWordService クラスは、エンティティごとに設定された禁止ワード（NGワード）が含まれているかどうかを判定し、発話の可否を決定するサービスクラスです。
現状では、Entity に forbidden_keywords フィールドがある前提で、カンマ区切りのリストとして禁止ワードを保持し、それに基づいて発話制御を行います。
- NGワードを含む場合は発話を禁止
- NGWordService.can_respond(input_text: str, entity) -> bool
- input_text にエンティティ固有の禁止ワード（NGワード）が含まれているかを判定する
- 禁止ワードが input_text に含まれている場合は False を返し、発話をブロックする
- 禁止ワードが設定されていない、または input_text に含まれていなければ True を返す
引数:
- input_text (str)：チェック対象の入力テキスト
- entity (Entity)：禁止ワードリストを持つエンティティ
戻り値:
- bool：NGワードを含まなければ True（発話可能）、含んでいれば False（発話禁止）
この処理が活用されるケース
- 不適切な発言のフィルタリング
- 特定の単語を含む発話をブロックすることで、不適切なコンテンツの生成を防ぐ
- カスタマイズ可能な発話制御
- エンティティごとに異なる禁止ワードを設定できるため、用途に応じたカスタマイズが可能
- 子ども向け AI チャットボットの安全対策
- 教育用途の AI などでは、不適切な単語をフィルタリングすることで安全性を高められる
今後の改善点
- forbidden_keywords の管理をより柔軟にする（データベースで禁止ワードリストを管理するなど）
- 禁止ワードが部分一致する場合の処理を強化（例：「ばか」と「ばかり」を区別する）
- 正規表現を用いた高度なフィルタリングを導入し、禁止ワードのバリエーションをカバーする
ai_agent/domain/service/rag.py
class RagService:
@staticmethod
def can_respond(input_text, entity):
"""
Determine if the entity can respond using RAG (Retrieval-Augmented Generation).
Args:
input_text (str): Input text to process.
entity (Entity): The entity being queried.
Returns:
bool: True if relevant data can be retrieved, otherwise False.
"""
# TODO: データベースまたはインデックスサーチによる情報検索を実装
# 仮実装: 特定のキーワードが含まれるかどうかで判定
return "法律" in input_text or "少子化" in input_text
このソースコードが実現する処理
RagService クラスは、RAG（Retrieval-Augmented Generation） を活用した発話可否の判定を行うサービスクラスです。
現状では、仮実装として input_text に特定のキーワード（"法律" または "少子化"）が含まれている場合に True を返し、それ以外は False を返す仕様になっています。
- RAG に基づく発話可否の判定
- RagService.can_respond(input_text: str, entity) -> bool
- 本来は、データベースやインデックス検索を用いて関連情報を取得し、その結果に基づいて発話の可否を決定する予定
- 現時点では、input_text に "法律" または "少子化" という単語が含まれているかどうかで判定する → 含まれていれば True（発話可能）、含まれていなければ False（発話不可）
引数:
- input_text (str)：検索対象の入力テキスト
- entity (Entity)：発話判定を行うエンティティ（未使用だが、今後の拡張を考慮）
戻り値:
- bool：仮実装では "法律" または "少子化" を含んでいれば True、それ以外は False
この処理が活用されるケース
- 知識ベースを活用した会話の最適化
- エンティティが持つ知識ベースを参照し、関連情報を検索して適切な応答を生成する
- RAG を用いた高度な情報検索
- 検索対象をデータベースやベクトル検索に拡張することで、より広範な情報を活用可能
- 特定のトピックに関する AI の発話制御
- 例えば、法律や少子化のようなテーマに関する質問にのみ応答する AI を作成できる
今後の改善点
- データベースまたはベクトル検索による情報検索の実装
- 例えば、Chroma や FAISS を利用し、埋め込み検索を行う
- エンティティごとに異なる知識ベースを持たせる
- Entity に関連するドキュメントを検索し、それに基づいた応答を行う仕組みを作る
- 検索結果のスコアリングとしきい値の導入
- 関連度の高いデータが見つかった場合のみ応答するようにする
ai_agent/domain/valueobject/conversation.py
from dataclasses import dataclass
@dataclass
class EntityVO:
name: str
next_turn: float
このソースコードが実現する処理
EntityVO は、会話の流れを管理するための 値オブジェクト（VO: Value Object） です。
主に、エンティティの名前 (name) と次の行動順 (next_turn) を保持する シンプルなデータ構造として機能します。
- エンティティの情報を保持する VO
- @dataclass デコレーターを使用することで、init や repr などのメソッドを自動生成
- 不変性（値オブジェクトの特性）を持つため、意図しない変更を防ぐことができる
- name（エンティティの名前）と next_turn（次の発話タイミング）を格納
フィールド:
- name (str)：エンティティの名前
- next_turn (float)：エンティティの次の行動順を示す数値
この VO が活用されるケース
- シミュレーション結果のデータ管理
- simulate_next_actions メソッド（ConversationService 内）で、各エンティティの発話順を保持するために使用される
- 発話の順序を整理するデータ構造としての利用
- EntityVO をリストに格納し、エンティティの行動予定を管理する
- エンティティのデータを簡潔に扱うための最適化
- Entity モデルを直接扱うのではなく、発話に関する最小限の情報のみを抽出し、シンプルなデータオブジェクトとして扱う
今後の改善点
- next_turn に整数型 (int) を許可するか、Decimal 型にすることで、より厳密な精度管理を行うことも可能
- 追加情報（例：speed や can_act など）を含め、エンティティの動作をより詳細に表現できるようにする
ai_agent/fixtures/entity.json
[
{
"model": "ai_agent.entity",
"pk": 1,
"fields": {
"name": "User",
"thinking_type": "google_maps_based"
}
},
{
"model": "ai_agent.entity",
"pk": 2,
"fields": {
"name": "Agent-A",
"thinking_type": "google_maps_based"
}
},
{
"model": "ai_agent.entity",
"pk": 3,
"fields": {
"name": "Agent-B",
"thinking_type": "rag_based"
}
},
{
"model": "ai_agent.entity",
"pk": 4,
"fields": {
"name": "Agent-C",
"thinking_type": "ng_word_based",
"forbidden_keywords": "暴力,万引き"
}
}
]
ai_agent/forms.py
from django import forms
class SendMessageForm(forms.Form):
user_input = forms.CharField(
required=True,
widget=forms.TextInput(
attrs={"placeholder": "Type your message", "class": "form-control"}
),
error_messages={"required": "Please enter a message."},
)
このソースコードが実現する処理
SendMessageForm クラスは、Django のフォーム (forms.Form) を使用してユーザー入力を処理する ためのフォームクラスです。
主に、テキストメッセージの送信フォームとして機能 します。
- ユーザーのメッセージ入力を受け取るフォーム
- user_input フィールドは CharField（テキスト入力）として定義されている
- 入力必須 (required=True) → ユーザーが空のメッセージを送信しようとするとエラーメッセージが表示される
- widget=forms.TextInput(...) を指定することで、HTML の を生成し、Bootstrap 風のデザインを適用
- error_messages に "required" のエラーメッセージをカスタマイズしている（デフォルトの Django メッセージを置き換え）
このフォームが活用されるケース
- Django のビューでメッセージ送信を処理する
- フォームのバリデーションを Django 側で管理し、適切なエラーハンドリングを行う
- フロントエンドの入力 UI を改善する
- widget の attrs を使って、placeholder や class を適用し、より使いやすいフォームを作成できる
- ユーザーが送信したメッセージを Django のビューで受け取り、処理する
- request.POST から user_input の値を取得し、チャットのメッセージとして処理する
今後の改善点
スタイルのカスタマイズを拡張
class="form-control" に加え、style="width:100%" などの CSS を適用することで UI を調整できる
ai_agent/models.py
from django.db import models
class GooglemapsReview(models.Model):
"""
Represents a Google Maps review for a specific location.
Attributes:
location_name (str): Name of the place being reviewed.
review_text (str): Content of the review.
rating (float): Rating given by the reviewer (1-5).
author_name (str): Name of the reviewer (optional).
review_date (datetime): Date when the review was posted.
latitude (float): Latitude of the reviewed location.
longitude (float): Longitude of the reviewed location.
vector (binary): Vector representation generated by Chroma (optional).
"""
location_name = models.CharField(max_length=255)
review_text = models.TextField()
rating = models.FloatField()
author_name = models.CharField(max_length=255, blank=True, null=True)
review_date = models.DateTimeField()
latitude = models.FloatField()
longitude = models.FloatField()
vector = models.BinaryField(null=True, blank=True)
def __str__(self):
return f"{self.location_name} - {self.rating} stars"
class Entity(models.Model):
"""
Represents an entity involved in conversations and defines its behavior.
This model is used to define conversation participants, each with specific reasoning
mechanisms, restrictions, and additional attributes for dynamic behavior in a system.
Attributes:
name (str): The name of the entity (e.g., a bot or user).
thinking_type (str): The reasoning or decision-making type associated with the entity.
Choices:
- "google_maps_based" (Google Mapsレビューに基づく)
- "rag_based" (RAGベースの推論)
- "ng_word_based" (NGワードに基づく制限)
forbidden_keywords (str, optional): A list of keywords that the entity should avoid,
typically used with "ng_word_based" reasoning.
vector (binary, optional): A binary vector representation of the entity's attributes,
commonly used for embedding-based reasoning with "rag_based".
speed (int): The decision-making speed or response speed of the entity, where
higher values may indicate slower response times.
"""
THINKING_TYPE_CHOICES = (
("google_maps_based", "Google Mapsレビューに基づく"), # Type A
("rag_based", "RAGベースの推論"), # Type B
("ng_word_based", "NGワードに基づく制限"), # Type C
)
name = models.CharField(max_length=100)
thinking_type = models.CharField(
max_length=50, choices=THINKING_TYPE_CHOICES, default="google_maps_based"
)
forbidden_keywords = models.TextField(blank=True, null=True) # Type C用
vector = models.BinaryField(null=True, blank=True) # Type B用
speed = models.IntegerField(default=10)
def __str__(self):
return f"{self.name} ({self.get_thinking_type_display()})"
class Message(models.Model):
"""
Represents a message in a conversation.
Attributes:
entity (Entity): The entity that sent the message.
message_content (str): Content of the message.
created_at (datetime): Timestamp when the message was created.
updated_at (datetime): Timestamp when the message was last updated.
"""
entity = models.ForeignKey(Entity, on_delete=models.CASCADE)
message_content = models.TextField()
created_at = models.DateTimeField(auto_now_add=True, db_index=True)
updated_at = models.DateTimeField(auto_now=True)
def __str__(self):
return f"Message from {self.entity.name} at {self.created_at}"
class ActionTimeline(models.Model):
"""
Tracks the next turn for each entity based on their speed.
"""
entity = models.OneToOneField(Entity, on_delete=models.CASCADE)
next_turn = models.FloatField(default=0)
can_act = models.BooleanField(default=True)
def __str__(self):
return f"{self.entity.name} - Next Turn: {self.next_turn}"
class ActionHistory(models.Model):
entity = models.ForeignKey(Entity, on_delete=models.CASCADE)
acted_at_turn = models.IntegerField()
done = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
def __str__(self):
return f"{self.entity.name} - Turn: {self.acted_at_turn} - Done: {self.done}"
このソースコードが実現する処理
この models.py には、AIエージェントの会話システムを支える主要なデータモデルが定義されています。各モデルは、エンティティ（AIエージェントやユーザー）、会話の流れ、メッセージの履歴、Google Maps レビューなどを管理するためのものです。
- Google Maps のレビュー情報を管理するモデル GooglemapsReview
- 会話に参加するエンティティ（AIエージェントやユーザー）を表すモデル Entity
- 会話内のメッセージを管理するモデル Message
- 会話の流れ（行動順序）を管理するモデル ActionTimeline
- 会話内でのエンティティの行動履歴を管理するモデル ActionHistory
この処理が活用されるケース
- Google Maps のレビューを活用した会話システム
- GooglemapsReview モデルに保存されたレビューを検索し、AIエージェントが発話する
- エージェントの行動順序の管理
- ActionTimeline を用いて、発話のターンを調整し、リアルな会話の流れを作る
- 過去の会話履歴を活用した対話
- Message や ActionHistory を参照することで、AIエージェントが過去の発話を踏まえた応答ができる
今後の改善点
- エンティティの行動決定をより高度にする
- speed の調整ロジックを強化し、よりリアルな会話モデルを構築する
- ベクトル検索との統合
- vector フィールドを活用し、類似度検索を行うことで、より関連性の高い情報を提供できる
- エージェントごとの発話パターンを増やす
- thinking_type を拡張し、複数の判断基準を組み合わせることで、より柔軟な AI を構築する
ai_agent/static/ai_agent/css/index.css
body {
padding-top: 48px;
}
.jumbotron li {
display: inline-block;
list-style-type: none;
padding-top: 4px;
}
ai_agent/templates/ai_agent/base.html
{% load static %}
<!DOCTYPE html>
<html lang="ja">
<head>
<!-- Global site tag (gtag.js) - Google Analytics -->
<script async src="https://www.googletagmanager.com/gtag/js?id=UA-43097095-9"></script>
<script>
window.dataLayer = window.dataLayer || [];
function gtag() {
dataLayer.push(arguments);
}
gtag('js', new Date());
gtag('config', 'UA-43097095-9');
</script>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<meta http-equiv="X-UA-Compatible" content="ie=edge">
<title>ai_agent</title>
<!-- bootstrap and css -->
<link rel="stylesheet" href="https://stackpath.bootstrapcdn.com/bootstrap/4.2.1/css/bootstrap.min.css"
integrity="sha384-GJzZqFGwb1QTTN6wy59ffF1BuGJpLSa9DkKMp0DgiMDm4iYMj70gZWKYbI706tWS" crossorigin="anonymous">
<link rel="stylesheet" href="{% static 'ai_agent/css/index.css' %}">
<!-- favicon -->
<link rel="icon" href="{% static 'ai_agent/c_a.ico' %}">
<!-- for ajax -->
<script>let myUrl = {"base": "{% url 'agt:index' %}"};</script>
</head>
<body>
<h1></h1>
<header>
<nav class="navbar fixed-top navbar-expand-lg navbar-light bg-light">
<a class="navbar-brand" href="{% url 'home:index' %}">Henojiya</a>
<button class="navbar-toggler" type="button" data-toggle="collapse" data-target="#navbarSupportedContent"
aria-controls="navbarSupportedContent" aria-expanded="false" aria-label="Toggle navigation">
<span class="navbar-toggler-icon"></span>
</button>
<div class="collapse navbar-collapse" id="navbarSupportedContent">
<ul class="navbar-nav mr-auto">
<li class="nav-item">
<select class="custom-select d-flex align-items-center" onChange="location.href=value;">
<option value="{% url 'home:index' %}">HOME</option>
<option value="{% url 'vnm:index' %}">VIETNAM</option>
<option value="{% url 'mrk:index' %}">GMARKER</option>
<option value="{% url 'shp:index' %}">SHOPPING</option>
<option value="{% url 'war:index' %}">WAREHOUSE</option>
<option value="{% url 'txo:index' %}">TAXONOMY</option>
<option value="{% url 'soil:home' %}">SOIL ANALYSIS</option>
<option value="{% url 'sec:index' %}">SECURITIES REPORT</option>
<option value="{% url 'hsp:index' %}">HOSPITAL</option>
<option value="{% url 'llm:index' %}">LLM_CHAT</option>
<option value="{% url 'agt:index' %}" selected>AI_AGENT</option>
</select>
</li>
</ul>
<form class="form-inline my-2 my-lg-0">
<input class="form-control mr-sm-2" type="search" placeholder="Search" aria-label="Search">
<button class="btn btn-outline-success my-2 my-sm-0" type="submit">Search</button>
</form>
</div>
</nav>
</header>
<div id="main">
{% block content %}{% endblock %}
</div>
<footer>
<p>© 2019 henojiya. / <a href="https://github.com/duri0214" target="_blank">github portfolio</a></p>
</footer>
<!-- Optional JavaScript -->
<!-- jQuery first, then Popper.js, then Bootstrap JS -->
<script src="https://code.jquery.com/jquery-3.7.1.min.js"
integrity="sha256-/JqT3SQfawRcv/BIHPThkBvs0OEvtFFmqPF/lYI/Cxo=" crossorigin="anonymous"></script>
<script src="https://cdnjs.cloudflare.com/ajax/libs/popper.js/1.14.6/umd/popper.min.js"
integrity="sha384-wHAiFfRlMFy6i5SRaxvfOCifBUQy1xHdJ/yoi7FRNXMRBu5WHdZYu1hA6ZOblgut"
crossorigin="anonymous"></script>
<script src="https://stackpath.bootstrapcdn.com/bootstrap/4.2.1/js/bootstrap.min.js"
integrity="sha384-B0UglyR+jN6CkvvICOB2joaf5I4l3gm9GU6Hc1og6Ls7i6U/mkkaduKaBhlAXv9k"
crossorigin="anonymous"></script>
</body>
</html>
ai_agent/templates/ai_agent/index.html
{% extends "ai_agent/base.html" %}
{% load static %}
{% block content %}
<div class="jumbotron">
<h1 class="display-4">Hello, AI Agent!</h1>
<p class="lead">It's an interesting AI Agent!</p>
<hr class="my-4">
<p>Interact with the AI Agent below.</p>
<ul>
<li>
<form method="POST" action="{% url 'agt:reset_timeline' %}" style="display: inline;">
{% csrf_token %}
<button type="submit" class="btn btn-danger">リセット</button>
</form>
</li>
<li>
<form method="POST" action="{% url 'agt:next_turn' %}" style="display: inline;">
{% csrf_token %}
<button type="submit" class="btn btn-primary">1単位時間進める</button>
</form>
</li>
</ul>
</div>
<div class="container">
<section id="content-section" class="d-flex flex-column">
<div class="d-flex flex-grow-1">
<!-- タイムラインディスプレイ (左側コンテンツ) -->
<div class="timeline-display flex-grow-1">
<div class="container mt-4">
<h2>AI Agent Simulation</h2>
{% if messages %}
{% for message in messages %}
<div class="alert alert-{{ message.tags }} alert-dismissible fade show"
role="alert">
{{ message }}
</div>
{% endfor %}
{% endif %}
{% for chat_message in chat_messages %}
<div class="alert alert-info" role="alert">
{{ chat_message.message_content }}
</div>
{% empty %}
<div class="alert alert-warning" role="alert">
メッセージ履歴がありません。
</div>
{% endfor %}
<form method="POST" action="{% url 'agt:index' %}">
{% csrf_token %}
{{ form.as_p }}
<button type="submit" class="btn btn-primary mt-2">Send</button>
</form>
</div>
</div>
<!-- タイムライン情報 (右側サイドバー) -->
<div class="timeline-info p-3 bg-light border">
<h4>現在のターン</h4>
<p>
{% with latest_completed_turn|default:0 as current_turn %}
現在のターン: {{ current_turn }}
{% endwith %}
</p>
<h4>予定される順番</h4>
<ul>
{% for future_action in future_actions %}
<li>{{ future_action.entity.name }} (次の行動予定: {{ future_action.acted_at_turn }})</li>
{% empty %}
<p>（なし）</p>
{% endfor %}
</ul>
<h4>終了したターンのログ</h4>
<ul>
{% for completed_action in completed_actions %}
<li>{{ completed_action.entity.name }} (完了済み: {{ completed_action.acted_at_turn }})</li>
{% empty %}
<p>（なし）</p>
{% endfor %}
</ul>
</div>
</div>
</section>
</div>
{% endblock %}
ai_agent/tests.py
from unittest.mock import patch
from django.test import TestCase
from ai_agent.domain.repository.conversation import ConversationRepository
from ai_agent.domain.service.conversation import ConversationService
from ai_agent.domain.valueobject.conversation import EntityVO
from ai_agent.models import Entity, ActionTimeline
class ConversationServiceTest(TestCase):
def setUp(self):
"""
Set up entities and initialize timeline for testing.
"""
# Entity1: 高速で Google Maps レビューに基づくタイプ
self.entity1 = Entity.objects.create(
name="Entity1",
speed=100, # 高速
thinking_type="google_maps_based",
)
# Entity2: 低速で NG ワードに基づくタイプ
self.entity2 = Entity.objects.create(
name="Entity2",
speed=10, # 低速
thinking_type="ng_word_based",
)
# 初期化時にタイムラインを設定
ConversationService.initialize_timeline()
# テスト用の入力テキスト
self.test_input_text = "sample input text"
def test_timeline_initialization(self):
"""
Test if the timeline is initialized correctly with all entities.
"""
# タイムラインに全エンティティが登録されているか確認
timelines = ActionTimeline.objects.all()
self.assertEqual(timelines.count(), 2)
# 各エンティティの next_turn が適切に計算されているか確認
for timeline in timelines:
self.assertEqual(timeline.next_turn, 1 / timeline.entity.speed)
def test_action_order_based_on_speed(self):
"""
Test action order based on entity speed.
"""
# 最初に行動するのは高速な Entity1 のはず
next_entity = ConversationService.get_next_entity(self.test_input_text)
self.assertEqual(next_entity, self.entity1)
# Entity1 が次回行動予定を早く更新するため、2回目も Entity1 が選ばれる
next_entity = ConversationService.get_next_entity(self.test_input_text)
self.assertEqual(next_entity, self.entity1)
# 速度の差が 10 倍であるため、Entity1 が 8 回行動した後に Entity2 のターンが来る
for _ in range(9):
next_entity = ConversationService.get_next_entity(self.test_input_text)
self.assertEqual(next_entity, self.entity2)
# Entity2 が行動した次には再び高速な Entity1 の順番となる
next_entity = ConversationService.get_next_entity(self.test_input_text)
self.assertEqual(next_entity, self.entity1)
def test_create_message_updates_timeline(self):
"""
Test if creating a message updates the timeline properly for the entity.
"""
# Entity1 でメッセージを作成
ConversationRepository.create_message(self.entity1, "Test Message")
# タイムラインを確認
timeline = ActionTimeline.objects.get(entity=self.entity1)
# タイムライン初期化時の next_turn を確認
self.assertEqual(timeline.next_turn, 1 / self.entity1.speed)
# get_next_entity を1回実行すると next_turn が更新される
ConversationService.get_next_entity(self.test_input_text)
timeline.refresh_from_db() # タイムラインを再取得
self.assertEqual(
timeline.next_turn, 1 / self.entity1.speed + 1 / self.entity1.speed
)
def test_simulate_next_actions(self):
"""
Test if the simulate_next_actions function correctly predicts the next actions.
"""
simulation = ConversationService.simulate_next_actions(max_steps=11)
# シミュレーション結果の期待値
expected_simulation = [
EntityVO(name="Entity1", next_turn=0.01),
EntityVO(name="Entity1", next_turn=0.02),
EntityVO(name="Entity1", next_turn=0.03),
EntityVO(name="Entity1", next_turn=0.04),
EntityVO(name="Entity1", next_turn=0.05),
EntityVO(name="Entity1", next_turn=0.06),
EntityVO(name="Entity1", next_turn=0.07),
EntityVO(name="Entity1", next_turn=0.08),
EntityVO(name="Entity1", next_turn=0.09),
EntityVO(name="Entity1", next_turn=0.10),
EntityVO(name="Entity2", next_turn=0.10),
]
# リスト全体を比較
for actual, expected in zip(simulation, expected_simulation):
self.assertEqual(actual.name, expected.name)
self.assertAlmostEqual(actual.next_turn, expected.next_turn, places=2)
@patch("ai_agent.domain.service.conversation.ConversationService.think")
def test_can_act_false_skips_entity(self, mock_think):
"""
Test if an entity is skipped when can_act is False.
"""
# think をモック化して、Entity1 が always False を返すように設定
def mock_think_side_effect(entity, input_text):
if entity == self.entity1:
return False # Entity1 をパスさせる
return True # 他のエンティティは True を返す
mock_think.side_effect = mock_think_side_effect
# Entity1 は skip され、Entity2 が選ばれるはず
next_entity = ConversationService.get_next_entity(self.test_input_text)
self.assertEqual(next_entity, self.entity2)
# Entity2 が選ばれた後、next_turn が次のターン（0.2）になることを確認する
timeline_entity2 = ActionTimeline.objects.get(entity=self.entity2)
self.assertEqual(timeline_entity2.next_turn, 0.2)
# モックが期待通り呼び出されたことを確認
mock_think.assert_any_call(self.entity1, self.test_input_text)
mock_think.assert_any_call(self.entity2, self.test_input_text)
このソースコードが実現する処理
この tests.py では、Django の TestCase を用いて、会話管理システムの主要なロジックをテストしています。
特に、エンティティの行動順序 (ActionTimeline) や発話可否判定 (can_act) に関する挙動を確認するためのユニットテストが含まれています。
- タイムラインが正しく初期化されるか
- 速度 (speed) に基づいて適切な順序でエンティティが行動するか
- メッセージを作成するとタイムラインが適切に更新されるか
- シミュレーション機能が正しく動作するか
- can_act=False の場合にエンティティがスキップされるか
このテストが活用されるケース
- エージェントの発話順管理の検証
- speed に基づく発話の順番が正しく制御されているかを確認できる
- タイムラインの自動更新のチェック
- next_turn の更新が適切に行われているかを保証する
- 発話可否判定 (think) の動作テスト
- NGワードや外部情報（Google Mapsレビュー、RAG）を考慮して適切に True/False が返るかをテストできる
ai_agent/urls.py
from django.urls import path
from ai_agent.views import IndexView, NextTurnView, ResetTimelineView
app_name = "agt"
urlpatterns = [
path("", IndexView.as_view(), name="index"),
path("next_turn/", NextTurnView.as_view(), name="next_turn"),
path("reset_timeline/", ResetTimelineView.as_view(), name="reset_timeline"),
]
ai_agent/views.py
from django.contrib import messages
from django.shortcuts import redirect
from django.urls import reverse_lazy
from django.views.generic import View
from django.views.generic.edit import FormView
from ai_agent.domain.repository.conversation import ConversationRepository
from ai_agent.domain.service.conversation import ConversationService
from ai_agent.forms import SendMessageForm
from ai_agent.models import Message, Entity, ActionHistory
class IndexView(FormView):
template_name = "ai_agent/index.html"
form_class = SendMessageForm
success_url = reverse_lazy("agt:index")
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
# メッセージ履歴
context["chat_messages"] = Message.objects.select_related("entity").order_by(
"created_at"
)
# タイムラインデータ
context["completed_actions"] = ActionHistory.objects.filter(done=True).order_by(
"acted_at_turn"
)
context["future_actions"] = ActionHistory.objects.filter(done=False).order_by(
"acted_at_turn"
)
context["latest_completed_turn"] = (
context["completed_actions"].last().acted_at_turn
if context["completed_actions"].exists()
else 0
)
return context
def form_valid(self, form):
entity = Entity.objects.get(name="User")
Message.objects.create(
entity=entity,
message_content=form.cleaned_data["user_input"],
)
return super().form_valid(form)
class ResetTimelineView(View):
"""
Resets the timeline and initializes future actions.
"""
@staticmethod
def post(request, *args, **kwargs):
# リセット処理を呼び出し
ResetTimelineView.reset_timeline()
return redirect("agt:index")
@staticmethod
def reset_timeline():
"""タイムラインをリセット"""
# メッセージ履歴をクリア
Message.objects.all().delete()
print("All messages have been cleared.") # デバッグログ
# ActionHistoryをクリア
ActionHistory.objects.all().delete()
print("All ActionHistory records have been cleared.") # デバッグログ
# タイムラインを初期化
ConversationService.initialize_timeline()
# 未来の10ターン分をActionHistoryに登録
ConversationService.simulate_next_actions(max_steps=10)
# ActionHistoryのすべての行動を未完了（done=False）にする
ActionHistory.objects.all().update(done=False)
class NextTurnView(View):
"""
Handles advancing to the next turn in the conversation.
"""
@staticmethod
def post(request, *args, **kwargs):
# 未完了の最初のアクションを取得。
next_action = (
ActionHistory.objects.filter(done=False).order_by("acted_at_turn").first()
)
if not next_action:
# 未完了のアクションがない場合フラッシュメッセージ設定
messages.info(
request,
"処理すべきアクションはもうありません。タイムラインがリセットされました。",
)
# リセット処理を直接呼び出し
ResetTimelineView.reset_timeline()
return redirect("agt:index")
# 選択されたアクションを完了済みにする
next_action.done = True
next_action.save()
try:
# 次のエンティティとその処理を取得
# input_text = request.POST.get("input_text") # TODO: ユーザー入力を処理する場合のメモ
next_entity = ConversationService.get_next_entity(input_text="")
# 仮の応答を生成
response = f"{next_entity.name} が行動しました: 仮の応答テキスト"
# メッセージを作成
ConversationRepository.create_message(next_entity, response)
# フラッシュメッセージを設定
messages.success(request, f"{next_entity.name} のターンが進行しました。")
except ValueError:
# 行動可能なエンティティがない場合、一旦リセット
messages.info(
request, "No more actions left to process. Timeline has been reset."
)
ResetTimelineView.reset_timeline()
return redirect("agt:index")
このソースコードが実現する処理
この views.py では、Django のビューを使って、エンティティの行動順に基づく会話の進行やタイムラインのリセット、メッセージ送信機能を提供しています。
具体的には、フォームの送信やボタンアクションに基づいて会話が進行し、ユーザーインターフェースに結果を反映させるロジックが含まれています。
- IndexView: メッセージの送信とタイムラインの表示
- ResetTimelineView: タイムラインのリセット
- NextTurnView: 次のターンへの進行
このビューが活用されるケース
- ユーザーインターフェースでの会話進行
- ユーザーが入力したメッセージを受け取り、AIエージェントが適切に反応する会話型システムの構築に使える
- 会話の流れの管理
- タイムラインやアクション履歴を使って、AIエージェントが次に発話するタイミングを管理
- タイムラインのリセットとシミュレーション
- 会話の流れをリセットして、再度シミュレーションを行い、次に進むエンティティを決定できる
今後の改善点
- エラーハンドリングの強化
- 現在のコードでは ValueError が発生した場合にリセット処理を行っていますが、エラーメッセージをもっと詳細に設定したり、ログを記録してデバッグを支援する仕組みを追加することができます
- ユーザー入力の処理
- NextTurnView でコメントアウトされている input_text を実際にユーザー入力に基づいて処理できるようにする
- 非同期処理の導入
- 長時間かかる処理（例: シミュレーション）を非同期タスクとしてバックグラウンドで処理できるようにする
config/settings.py
:
"hospital",
"home",
"llm_chat",
+ "ai_agent",
]
MIDDLEWARE = [
:
config/urls.py
path("llm_chat/", include("llm_chat.urls")),
+ path("ai_agent/", include("ai_agent.urls")),
path("admin/", admin.site.urls),
TODO: それぞれのAIのなかみを作成