はじめに
まずは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(),
)
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
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
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
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
ai_agent/domain/valueobject/conversation.py
from dataclasses import dataclass
@dataclass
class EntityVO:
name: str
next_turn: float
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."},
)
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}"
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)
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")
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のなかみを作成