wassname/Open-Assistant
1
0
Fork 0
mirror of https://github.com/wassname/Open-Assistant.git synced 2026-07-08 00:10:24 +08:00
Code Issues Packages Projects Releases Wiki Activity

Renaming post to message

Browse Source
This commit is contained in:
alexandrelefourner
2022-12-30 17:50:47 +01:00
committed by Andreas Köpf
parent 9a1a4876f0
commit 7743459220
12 changed files with 235 additions and 235 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/main.py
+48 -48
View File
@@ -67,10 +67,10 @@ if settings.DEBUG_USE_SEED_DATA:
@app.on_event("startup")
def seed_data():
class DummyPost(pydantic.BaseModel):
task_post_id: str
user_post_id: str
parent_post_id: Optional[str]
class DummyMessage(pydantic.BaseModel):
task_message_id: str
user_message_id: str
parent_message_id: Optional[str]
text: str
role: str
@@ -81,93 +81,93 @@ if settings.DEBUG_USE_SEED_DATA:
dummy_user = protocol_schema.User(id="__dummy_user__", display_name="Dummy User", auth_method="local")
pr = PromptRepository(db=db, api_client=api_client, user=dummy_user)
dummy_posts = [
DummyPost(
task_post_id="de111fa8",
user_post_id="6f1d0711",
parent_post_id=None,
dummy_messages = [
DummyMessage(
task_message_id="de111fa8",
user_message_id="6f1d0711",
parent_message_id=None,
text="Hi!",
role="user",
),
DummyPost(
task_post_id="74c381d4",
user_post_id="4a24530b",
parent_post_id="6f1d0711",
DummyMessage(
task_message_id="74c381d4",
user_message_id="4a24530b",
parent_message_id="6f1d0711",
text="Hello! How can I help you?",
role="assistant",
),
DummyPost(
task_post_id="3d5dc440",
user_post_id="a8c01c04",
parent_post_id="4a24530b",
DummyMessage(
task_message_id="3d5dc440",
user_message_id="a8c01c04",
parent_message_id="4a24530b",
text="Do you have a recipe for potato soup?",
role="user",
),
DummyPost(
task_post_id="643716c1",
user_post_id="f43a93b7",
parent_post_id="4a24530b",
DummyMessage(
task_message_id="643716c1",
user_message_id="f43a93b7",
parent_message_id="4a24530b",
text="Who were the 8 presidents before George Washington?",
role="user",
),
DummyPost(
task_post_id="2e4e1e6",
user_post_id="c886920",
parent_post_id="6f1d0711",
DummyMessage(
task_message_id="2e4e1e6",
user_message_id="c886920",
parent_message_id="6f1d0711",
text="Hey buddy! How can I serve you?",
role="assistant",
),
DummyPost(
task_post_id="970c437d",
user_post_id="cec432cf",
parent_post_id=None,
DummyMessage(
task_message_id="970c437d",
user_message_id="cec432cf",
parent_message_id=None,
text="euirdteunvglfe23908230892309832098 AAAAAAAA",
role="user",
),
DummyPost(
task_post_id="6066118e",
user_post_id="4f85f637",
parent_post_id="cec432cf",
DummyMessage(
task_message_id="6066118e",
user_message_id="4f85f637",
parent_message_id="cec432cf",
text="Sorry, I did not understand your request and it is unclear to me what you want me to do. Could you describe it in a different way?",
role="assistant",
),
DummyPost(
task_post_id="ba87780d",
user_post_id="0e276b98",
parent_post_id="cec432cf",
DummyMessage(
task_message_id="ba87780d",
user_message_id="0e276b98",
parent_message_id="cec432cf",
text="I'm unsure how to interpret this. Is it a riddle?",
role="assistant",
),
]
for p in dummy_posts:
wp = pr.fetch_workpackage_by_postid(p.task_post_id)
for p in dummy_messages:
wp = pr.fetch_workpackage_by_message_id(p.task_message_id)
if wp and not wp.ack:
logger.warning("Deleting unacknowledged seed data work package")
db.delete(wp)
wp = None
if not wp:
if p.parent_post_id is None:
if p.parent_message_id is None:
wp = pr.store_task(
protocol_schema.InitialPromptTask(hint=""), thread_id=None, parent_post_id=None
protocol_schema.InitialPromptTask(hint=""), thread_id=None, parent_message_id=None
)
else:
print("p.parent_post_id", p.parent_post_id)
parent_post = pr.fetch_post_by_frontend_post_id(p.parent_post_id, fail_if_missing=True)
print("p.parent_message_id", p.parent_message_id)
parent_message = pr.fetch_message_by_frontend_message_id(p.parent_message_id, fail_if_missing=True)
wp = pr.store_task(
protocol_schema.AssistantReplyTask(
conversation=protocol_schema.Conversation(
messages=[protocol_schema.ConversationMessage(text="dummy", is_assistant=False)]
)
),
thread_id=parent_post.thread_id,
parent_post_id=parent_post.id,
thread_id=parent_message.thread_id,
parent_message_id=parent_message.id,
)
pr.bind_frontend_post_id(wp.id, p.task_post_id)
post = pr.store_text_reply(p.text, p.task_post_id, p.user_post_id)
pr.bind_frontend_message_id(wp.id, p.task_message_id)
message = pr.store_text_reply(p.text, p.task_message_id, p.user_message_id)
logger.info(
f"Inserted: post_id: {post.id}, payload: {post.payload.payload}, parent_post_id: {post.parent_id}"
f"Inserted: message_id: {message.id}, payload: {message.payload.payload}, parent_message_id: {message.parent_id}"
)
else:
logger.debug(f"seed data work_package found: {wp.id}")
backend/oasst_backend/api/v1/tasks.py
+29 -29
View File
@@ -19,7 +19,7 @@ def generate_task(
request: protocol_schema.TaskRequest, pr: PromptRepository
) -> Tuple[protocol_schema.Task, Optional[UUID], Optional[UUID]]:
thread_id = None
parent_post_id = None
parent_message_id = None
match request.type:
case protocol_schema.TaskRequestType.random:
@@ -56,34 +56,34 @@ def generate_task(
)
case protocol_schema.TaskRequestType.user_reply:
logger.info("Generating a UserReplyTask.")
posts = pr.fetch_random_conversation("assistant")
messages = pr.fetch_random_conversation("assistant")
messages = [
protocol_schema.ConversationMessage(text=p.payload.payload.text, is_assistant=(p.role == "assistant"))
for p in posts
protocol_schema.ConversationMessage(text=m.payload.payload.text, is_assistant=(m.role == "assistant"))
for m in messages
]
task = protocol_schema.UserReplyTask(conversation=protocol_schema.Conversation(messages=messages))
thread_id = posts[-1].thread_id
parent_post_id = posts[-1].id
thread_id = messages[-1].thread_id
parent_message_id = messages[-1].id
case protocol_schema.TaskRequestType.assistant_reply:
logger.info("Generating a AssistantReplyTask.")
posts = pr.fetch_random_conversation("user")
messages = pr.fetch_random_conversation("user")
messages = [
protocol_schema.ConversationMessage(text=p.payload.payload.text, is_assistant=(p.role == "assistant"))
for p in posts
protocol_schema.ConversationMessage(text=m.payload.payload.text, is_assistant=(m.role == "assistant"))
for m in messages
]
task = protocol_schema.AssistantReplyTask(conversation=protocol_schema.Conversation(messages=messages))
thread_id = posts[-1].thread_id
parent_post_id = posts[-1].id
thread_id = messages[-1].thread_id
parent_message_id = messages[-1].id
case protocol_schema.TaskRequestType.rank_initial_prompts:
logger.info("Generating a RankInitialPromptsTask.")
posts = pr.fetch_random_initial_prompts()
task = protocol_schema.RankInitialPromptsTask(prompts=[p.payload.payload.text for p in posts])
messages = pr.fetch_random_initial_prompts()
task = protocol_schema.RankInitialPromptsTask(prompts=[m.payload.payload.text for m in messages])
case protocol_schema.TaskRequestType.rank_user_replies:
logger.info("Generating a RankUserRepliesTask.")
conversation, replies = pr.fetch_multiple_random_replies(post_role="assistant")
conversation, replies = pr.fetch_multiple_random_replies(message_role="assistant")
messages = [
protocol_schema.ConversationMessage(
@@ -102,7 +102,7 @@ def generate_task(
case protocol_schema.TaskRequestType.rank_assistant_replies:
logger.info("Generating a RankAssistantRepliesTask.")
conversation, replies = pr.fetch_multiple_random_replies(post_role="user")
conversation, replies = pr.fetch_multiple_random_replies(message_role="user")
messages = [
protocol_schema.ConversationMessage(
@@ -121,7 +121,7 @@ def generate_task(
logger.info(f"Generated {task=}.")
return task, thread_id, parent_post_id
return task, thread_id, parent_message_id
@router.post("/", response_model=protocol_schema.AnyTask) # work with Union once more types are added
@@ -138,8 +138,8 @@ def request_task(
try:
pr = PromptRepository(db, api_client, request.user)
task, thread_id, parent_post_id = generate_task(request, pr)
pr.store_task(task, thread_id, parent_post_id, request.collective)
task, thread_id, parent_message_id = generate_task(request, pr)
pr.store_task(task, thread_id, parent_message_id, request.collective)
except OasstError:
raise
@@ -166,9 +166,9 @@ def acknowledge_task(
try:
pr = PromptRepository(db, api_client, user=None)
# here we store the post id in the database for the task
# here we store the message id in the database for the task
logger.info(f"Frontend acknowledges task {task_id=}, {ack_request=}.")
pr.bind_frontend_post_id(task_id=task_id, post_id=ack_request.post_id)
pr.bind_frontend_message_id(task_id=task_id, message_id=ack_request.message_id)
except OasstError:
raise
@@ -201,7 +201,7 @@ def acknowledge_task_failure(
@router.post("/interaction")
def post_interaction(
def message_interaction(
*,
db: Session = Depends(deps.get_db),
api_key: APIKey = Depends(deps.get_api_key),
@@ -216,29 +216,29 @@ def post_interaction(
pr = PromptRepository(db, api_client, user=interaction.user)
match type(interaction):
case protocol_schema.TextReplyToPost:
case protocol_schema.TextReplyToMessage:
logger.info(
f"Frontend reports text reply to {interaction.post_id=} with {interaction.text=} by {interaction.user=}."
f"Frontend reports text reply to {interaction.message_id=} with {interaction.text=} by {interaction.user=}."
)
# here we store the text reply in the database
pr.store_text_reply(
text=interaction.text, post_id=interaction.post_id, user_post_id=interaction.user_post_id
text=interaction.text, message_id=interaction.message_id, user_message_id=interaction.user_message_id
)
return protocol_schema.TaskDone()
case protocol_schema.PostRating:
case protocol_schema.MessageRating:
logger.info(
f"Frontend reports rating of {interaction.post_id=} with {interaction.rating=} by {interaction.user=}."
f"Frontend reports rating of {interaction.message_id=} with {interaction.rating=} by {interaction.user=}."
)
# here we store the rating in the database
pr.store_rating(interaction)
return protocol_schema.TaskDone()
case protocol_schema.PostRanking:
case protocol_schema.MessageRanking:
logger.info(
f"Frontend reports ranking of {interaction.post_id=} with {interaction.ranking=} by {interaction.user=}."
f"Frontend reports ranking of {interaction.message_id=} with {interaction.ranking=} by {interaction.user=}."
)
# TODO: check if the ranking is valid
@@ -262,5 +262,5 @@ def close_collective_task(
):
api_client = deps.api_auth(api_key, db)
pr = PromptRepository(db, api_client, user=None)
pr.close_task(close_task_request.post_id)
pr.close_task(close_task_request.message_id)
return protocol_schema.TaskDone()
backend/oasst_backend/journal_writer.py
+20 -20
View File
@@ -14,36 +14,36 @@ class JournalEventType(str, enum.Enum):
"""A label for a piece of text."""
user_created = "user_created"
text_reply_to_post = "text_reply_to_post"
post_rating = "post_rating"
post_ranking = "post_ranking"
text_reply_to_message = "text_reply_to_message"
message_rating = "message_rating"
message_ranking = "message_ranking"
@payload_type
class JournalEvent(BaseModel):
type: str
user_id: Optional[UUID]
post_id: Optional[UUID]
message_id: Optional[UUID]
workpackage_id: Optional[UUID]
task_type: Optional[str]
@payload_type
class TextReplyEvent(JournalEvent):
type: Literal[JournalEventType.text_reply_to_post] = JournalEventType.text_reply_to_post
type: Literal[JournalEventType.text_reply_to_message] = JournalEventType.text_reply_to_message
length: int
role: str
@payload_type
class RatingEvent(JournalEvent):
type: Literal[JournalEventType.post_rating] = JournalEventType.post_rating
type: Literal[JournalEventType.message_rating] = JournalEventType.message_rating
rating: int
@payload_type
class RankingEvent(JournalEvent):
type: Literal[JournalEventType.post_ranking] = JournalEventType.post_ranking
type: Literal[JournalEventType.message_ranking] = JournalEventType.message_ranking
ranking: list[int]
@@ -54,31 +54,31 @@ class JournalWriter:
self.user = user
self.user_id = self.user.id if self.user else None
def log_text_reply(self, work_package: WorkPackage, post_id: UUID, role: str, length: int) -> Journal:
def log_text_reply(self, work_package: WorkPackage, message_id: UUID, role: str, length: int) -> Journal:
return self.log(
task_type=work_package.payload_type,
event_type=JournalEventType.text_reply_to_post,
event_type=JournalEventType.text_reply_to_message,
payload=TextReplyEvent(role=role, length=length),
workpackage_id=work_package.id,
post_id=post_id,
message_id=message_id,
)
def log_rating(self, work_package: WorkPackage, post_id: UUID, rating: int) -> Journal:
def log_rating(self, work_package: WorkPackage, message_id: UUID, rating: int) -> Journal:
return self.log(
task_type=work_package.payload_type,
event_type=JournalEventType.post_rating,
event_type=JournalEventType.message_rating,
payload=RatingEvent(rating=rating),
workpackage_id=work_package.id,
post_id=post_id,
message_id=message_id,
)
def log_ranking(self, work_package: WorkPackage, post_id: UUID, ranking: list[int]) -> Journal:
def log_ranking(self, work_package: WorkPackage, message_id: UUID, ranking: list[int]) -> Journal:
return self.log(
task_type=work_package.payload_type,
event_type=JournalEventType.post_ranking,
event_type=JournalEventType.message_ranking,
payload=RankingEvent(ranking=ranking),
workpackage_id=work_package.id,
post_id=post_id,
message_id=message_id,
)
def log(
@@ -88,7 +88,7 @@ class JournalWriter:
task_type: str,
event_type: str = None,
workpackage_id: Optional[UUID] = None,
post_id: Optional[UUID] = None,
message_id: Optional[UUID] = None,
commit: bool = True,
) -> Journal:
if event_type is None:
@@ -99,8 +99,8 @@ class JournalWriter:
if payload.user_id is None:
payload.user_id = self.user_id
if payload.post_id is None:
payload.post_id = post_id
if payload.message_id is None:
payload.message_id = message_id
if payload.workpackage_id is None:
payload.workpackage_id = workpackage_id
if payload.task_type is None:
@@ -112,7 +112,7 @@ class JournalWriter:
created_date=utcnow(),
event_type=event_type,
event_payload=PayloadContainer(payload=payload),
post_id=post_id,
message_id=message_id,
)
self.db.add(entry)
backend/oasst_backend/models/__init__.py
+4 -4
View File
@@ -3,8 +3,8 @@ from .api_client import ApiClient
from .journal import Journal, JournalIntegration
from .user import User
from .user_stats import UserStats
from .post import Post
from .post_reaction import PostReaction
from .message import Message
from .message_reaction import MessageReaction
from .text_labels import TextLabels
from .work_package import WorkPackage
@@ -12,8 +12,8 @@ __all__ = [
"ApiClient",
"User",
"UserStats",
"Post",
"PostReaction",
"Message",
"MessageReaction",
"WorkPackage",
"TextLabels",
"Journal",
backend/oasst_backend/models/db_payload.py
+2 -2
View File
@@ -56,13 +56,13 @@ class ReactionPayload(BaseModel):
@payload_type
class RatingReactionPayload(ReactionPayload):
type: Literal["post_rating"] = "post_rating"
type: Literal["message_rating"] = "message_rating"
rating: str
@payload_type
class RankingReactionPayload(ReactionPayload):
type: Literal["post_ranking"] = "post_ranking"
type: Literal["message_ranking"] = "message_ranking"
ranking: list[int]
backend/oasst_backend/models/post.py → backend/oasst_backend/models/message.py
+4 -4
View File
@@ -10,9 +10,9 @@ from sqlmodel import Field, Index, SQLModel
from .payload_column_type import PayloadContainer, payload_column_type
class Post(SQLModel, table=True):
__tablename__ = "post"
__table_args__ = (Index("ix_post_frontend_post_id", "api_client_id", "frontend_post_id", unique=True),)
class Message(SQLModel, table=True):
__tablename__ = "message"
__table_args__ = (Index("ix_message_frontend_message_id", "api_client_id", "frontend_message_id", unique=True),)
id: Optional[UUID] = Field(
sa_column=sa.Column(
@@ -25,7 +25,7 @@ class Post(SQLModel, table=True):
user_id: UUID = Field(nullable=True, foreign_key="user.id", index=True)
role: str = Field(nullable=False, max_length=128)
api_client_id: UUID = Field(nullable=False, foreign_key="api_client.id")
frontend_post_id: str = Field(max_length=200, nullable=False)
frontend_message_id: str = Field(max_length=200, nullable=False)
created_date: Optional[datetime] = Field(
sa_column=sa.Column(sa.DateTime(), nullable=False, server_default=sa.func.current_timestamp())
)
backend/oasst_backend/models/post_reaction.py → backend/oasst_backend/models/message_reaction.py
+2 -2
View File
@@ -10,8 +10,8 @@ from sqlmodel import Field, SQLModel
from .payload_column_type import PayloadContainer, payload_column_type
class PostReaction(SQLModel, table=True):
__tablename__ = "post_reaction"
class MessageReaction(SQLModel, table=True):
__tablename__ = "message_reaction"
work_package_id: Optional[UUID] = Field(
sa_column=sa.Column(pg.UUID(as_uuid=True), sa.ForeignKey("work_package.id"), nullable=False, primary_key=True)
backend/oasst_backend/models/text_labels.py
+1 -1
View File
@@ -21,5 +21,5 @@ class TextLabels(SQLModel, table=True):
)
api_client_id: UUID = Field(nullable=False, foreign_key="api_client.id")
text: str = Field(nullable=False, max_length=2**16)
post_id: Optional[UUID] = Field(sa_column=sa.Column(pg.UUID(as_uuid=True), sa.ForeignKey("post.id"), nullable=True))
message_id: Optional[UUID] = Field(sa_column=sa.Column(pg.UUID(as_uuid=True), sa.ForeignKey("message.id"), nullable=True))
labels: dict[str, float] = Field(default={}, sa_column=sa.Column(pg.JSONB), nullable=False)
backend/oasst_backend/models/user_stats.py
+3 -3
View File
@@ -20,9 +20,9 @@ class UserStats(SQLModel, table=True):
)
reactions: int = 0 # reactions sent by user
posts: int = 0 # posts sent by user
messages: int = 0 # messages sent by user
upvotes: int = 0 # received upvotes (form other users)
downvotes: int = 0 # received downvotes (from other users)
work_reward: int = 0 # reward for workpackage completions
compare_wins: int = 0 # num times user's post won compare tasks
compare_losses: int = 0 # num times users's post lost compare tasks
compare_wins: int = 0 # num times user's message won compare tasks
compare_losses: int = 0 # num times users's message lost compare tasks
backend/oasst_backend/models/work_package.py
+2 -2
View File
@@ -29,9 +29,9 @@ class WorkPackage(SQLModel, table=True):
api_client_id: UUID = Field(nullable=False, foreign_key="api_client.id")
ack: Optional[bool] = None
done: bool = Field(sa_column=sa.Column(sa.Boolean, nullable=False, server_default=false()))
frontend_ref_post_id: Optional[str] = None
frontend_ref_message_id: Optional[str] = None
thread_id: Optional[UUID] = None
parent_post_id: Optional[UUID] = None
parent_message_id: Optional[UUID] = None
collective: bool = Field(sa_column=sa.Column(sa.Boolean, nullable=False, server_default=false()))
@property
backend/oasst_backend/prompt_repository.py
+116 -116
View File
@@ -7,7 +7,7 @@ import oasst_backend.models.db_payload as db_payload
from loguru import logger
from oasst_backend.exceptions import OasstError, OasstErrorCode
from oasst_backend.journal_writer import JournalWriter
from oasst_backend.models import ApiClient, User, Post, PostReaction, TextLabels, WorkPackage
from oasst_backend.models import ApiClient, User, Message, MessageReaction, TextLabels, WorkPackage
from oasst_backend.models.payload_column_type import PayloadContainer
from oasst_shared.schemas import protocol as protocol_schema
from sqlmodel import Session, func
@@ -51,14 +51,14 @@ class PromptRepository:
self.db.commit()
return user
def validate_post_id(self, post_id: str) -> None:
if not isinstance(post_id, str):
raise OasstError(f"post_id must be string, not {type(post_id)}", OasstErrorCode.INVALID_POST_ID)
if not post_id:
raise OasstError("post_id must not be empty", OasstErrorCode.INVALID_POST_ID)
def validate_message_id(self, message_id: str) -> None:
if not isinstance(message_id, str):
raise OasstError(f"message_id must be string, not {type(message_id)}", OasstErrorCode.INVALID_POST_ID)
if not message_id:
raise OasstError("message_id must not be empty", OasstErrorCode.INVALID_POST_ID)
def bind_frontend_post_id(self, task_id: UUID, post_id: str):
self.validate_post_id(post_id)
def bind_frontend_message_id(self, task_id: UUID, message_id: str):
self.validate_message_id(message_id)
# find work package
work_pack: WorkPackage = (
@@ -73,7 +73,7 @@ class PromptRepository:
if work_pack.done or work_pack.ack is not None:
raise OasstError("WorkPackage already updated.", OasstErrorCode.WORK_PACKAGE_ALREADY_UPDATED)
work_pack.frontend_ref_post_id = post_id
work_pack.frontend_ref_message_id = message_id
work_pack.ack = True
# ToDo: check race-condition, transaction
self.db.add(work_pack)
@@ -98,34 +98,34 @@ class PromptRepository:
self.db.add(work_pack)
self.db.commit()
def fetch_post_by_frontend_post_id(self, frontend_post_id: str, fail_if_missing: bool = True) -> Post:
self.validate_post_id(frontend_post_id)
post: Post = (
self.db.query(Post)
.filter(Post.api_client_id == self.api_client.id, Post.frontend_post_id == frontend_post_id)
def fetch_message_by_frontend_message_id(self, frontend_message_id: str, fail_if_missing: bool = True) -> Message:
self.validate_message_id(frontend_message_id)
message: Message = (
self.db.query(Message)
.filter(Message.api_client_id == self.api_client.id, Message.frontend_message_id == frontend_message_id)
.one_or_none()
)
if fail_if_missing and post is None:
raise OasstError(f"Post with post_id {frontend_post_id} not found.", OasstErrorCode.POST_NOT_FOUND)
return post
if fail_if_missing and message is None:
raise OasstError(f"Message with message_id {frontend_message_id} not found.", OasstErrorCode.POST_NOT_FOUND)
return message
def fetch_workpackage_by_postid(self, post_id: str) -> WorkPackage:
self.validate_post_id(post_id)
def fetch_workpackage_by_message_id(self, message_id: str) -> WorkPackage:
self.validate_message_id(message_id)
work_pack = (
self.db.query(WorkPackage)
.filter(WorkPackage.api_client_id == self.api_client.id, WorkPackage.frontend_ref_post_id == post_id)
.filter(WorkPackage.api_client_id == self.api_client.id, WorkPackage.frontend_ref_message_id == message_id)
.one_or_none()
)
return work_pack
def store_text_reply(self, text: str, post_id: str, user_post_id: str, role: str = None) -> Post:
self.validate_post_id(post_id)
self.validate_post_id(user_post_id)
def store_text_reply(self, text: str, message_id: str, user_message_id: str, role: str = None) -> Message:
self.validate_message_id(message_id)
self.validate_message_id(user_message_id)
wp = self.fetch_workpackage_by_postid(post_id)
wp = self.fetch_workpackage_by_message_id(message_id)
if wp is None:
raise OasstError(f"WorkPackage for {post_id=} not found", OasstErrorCode.WORK_PACKAGE_NOT_FOUND)
raise OasstError(f"WorkPackage for {message_id=} not found", OasstErrorCode.WORK_PACKAGE_NOT_FOUND)
if wp.expired:
raise OasstError("WorkPackage already expired.", OasstErrorCode.WORK_PACKAGE_EXPIRED)
if not wp.ack:
@@ -133,44 +133,44 @@ class PromptRepository:
if wp.done:
raise OasstError("WorkPackage already done.", OasstErrorCode.WORK_PACKAGE_ALREADY_DONE)
# If there's no parent post assume user started new conversation
# If there's no parent message assume user started new conversation
role = "user"
depth = 0
if wp.parent_post_id:
parent_post = self.fetch_post(wp.parent_post_id)
parent_post.children_count += 1
self.db.add(parent_post)
if wp.parent_message_id:
parent_message = self.fetch_message(wp.parent_message_id)
parent_message.children_count += 1
self.db.add(parent_message)
depth = parent_post.depth + 1
if parent_post.role == "assistant":
depth = parent_message.depth + 1
if parent_message.role == "assistant":
role = "user"
else:
role = "assistant"
# create reply post
new_post_id = uuid4()
user_post = self.insert_post(
post_id=new_post_id,
frontend_post_id=user_post_id,
parent_id=wp.parent_post_id,
thread_id=wp.thread_id or new_post_id,
# create reply message
new_message_id = uuid4()
user_message = self.insert_message(
message_id=new_message_id,
frontend_message_id=user_message_id,
parent_id=wp.parent_message_id,
thread_id=wp.thread_id or new_message_id,
workpackage_id=wp.id,
role=role,
payload=db_payload.PostPayload(text=text),
payload=db_payload.MessagePayload(text=text),
depth=depth,
)
if not wp.collective:
wp.done = True
self.db.add(wp)
self.db.commit()
self.journal.log_text_reply(work_package=wp, post_id=new_post_id, role=role, length=len(text))
return user_post
self.journal.log_text_reply(work_package=wp, message_id=new_message_id, role=role, length=len(text))
return user_message
def store_rating(self, rating: protocol_schema.PostRating) -> PostReaction:
post = self.fetch_post_by_frontend_post_id(rating.post_id, fail_if_missing=True)
def store_rating(self, rating: protocol_schema.MessageRating) -> MessageReaction:
message = self.fetch_message_by_frontend_message_id(rating.message_id, fail_if_missing=True)
work_package = self.fetch_workpackage_by_postid(rating.post_id)
work_package = self.fetch_workpackage_by_message_id(rating.message_id)
work_payload: db_payload.RateSummaryPayload = work_package.payload.payload
if type(work_payload) != db_payload.RateSummaryPayload:
raise OasstError(
@@ -184,20 +184,20 @@ class PromptRepository:
OasstErrorCode.RATING_OUT_OF_RANGE,
)
# store reaction to post
# store reaction to message
reaction_payload = db_payload.RatingReactionPayload(rating=rating.rating)
reaction = self.insert_reaction(post.id, reaction_payload)
reaction = self.insert_reaction(message.id, reaction_payload)
if not work_package.collective:
work_package.done = True
self.db.add(work_package)
self.journal.log_rating(work_package, post_id=post.id, rating=rating.rating)
self.journal.log_rating(work_package, message_id=message.id, rating=rating.rating)
logger.info(f"Ranking {rating.rating} stored for work_package {work_package.id}.")
return reaction
def store_ranking(self, ranking: protocol_schema.PostRanking) -> PostReaction:
def store_ranking(self, ranking: protocol_schema.MessageRanking) -> MessageReaction:
# fetch work_package
work_package = self.fetch_workpackage_by_postid(ranking.post_id)
work_package = self.fetch_workpackage_by_message_id(ranking.message_id)
if not work_package.collective:
work_package.done = True
self.db.add(work_package)
@@ -217,11 +217,11 @@ class PromptRepository:
OasstErrorCode.INVALID_RANKING_VALUE,
)
# store reaction to post
# store reaction to message
reaction_payload = db_payload.RankingReactionPayload(ranking=ranking.ranking)
reaction = self.insert_reaction(work_package.id, reaction_payload)
# TODO: resolve post_id
self.journal.log_ranking(work_package, post_id=None, ranking=ranking.ranking)
# TODO: resolve message_id
self.journal.log_ranking(work_package, message_id=None, ranking=ranking.ranking)
logger.info(f"Ranking {ranking.ranking} stored for work_package {work_package.id}.")
@@ -235,11 +235,11 @@ class PromptRepository:
OasstErrorCode.INVALID_RANKING_VALUE,
)
# store reaction to post
# store reaction to message
reaction_payload = db_payload.RankingReactionPayload(ranking=ranking.ranking)
reaction = self.insert_reaction(work_package.id, reaction_payload)
# TODO: resolve post_id
self.journal.log_ranking(work_package, post_id=None, ranking=ranking.ranking)
# TODO: resolve message_id
self.journal.log_ranking(work_package, message_id=None, ranking=ranking.ranking)
logger.info(f"Ranking {ranking.ranking} stored for work_package {work_package.id}.")
@@ -255,7 +255,7 @@ class PromptRepository:
self,
task: protocol_schema.Task,
thread_id: UUID = None,
parent_post_id: UUID = None,
parent_message_id: UUID = None,
collective: bool = False,
) -> WorkPackage:
payload: db_payload.TaskPayload
@@ -294,7 +294,7 @@ class PromptRepository:
raise OasstError(f"Invalid task type: {type(task)=}", OasstErrorCode.INVALID_TASK_TYPE)
wp = self.insert_work_package(
payload=payload, id=task.id, thread_id=thread_id, parent_post_id=parent_post_id, collective=collective
payload=payload, id=task.id, thread_id=thread_id, parent_message_id=parent_message_id, collective=collective
)
assert wp.id == task.id
return wp
@@ -304,7 +304,7 @@ class PromptRepository:
payload: db_payload.TaskPayload,
id: UUID = None,
thread_id: UUID = None,
parent_post_id: UUID = None,
parent_message_id: UUID = None,
collective: bool = False,
) -> WorkPackage:
c = PayloadContainer(payload=payload)
@@ -315,7 +315,7 @@ class PromptRepository:
payload=c,
api_client_id=self.api_client.id,
thread_id=thread_id,
parent_post_id=parent_post_id,
parent_message_id=parent_message_id,
collective=collective,
)
self.db.add(wp)
@@ -323,49 +323,49 @@ class PromptRepository:
self.db.refresh(wp)
return wp
def insert_post(
def insert_message(
self,
*,
post_id: UUID,
frontend_post_id: str,
message_id: UUID,
frontend_message_id: str,
parent_id: UUID,
thread_id: UUID,
workpackage_id: UUID,
role: str,
payload: db_payload.PostPayload,
payload: db_payload.MessagePayload,
payload_type: str = None,
depth: int = 0,
) -> Post:
) -> Message:
if payload_type is None:
if payload is None:
payload_type = "null"
else:
payload_type = type(payload).__name__
post = Post(
id=post_id,
message = Message(
id=message_id,
parent_id=parent_id,
thread_id=thread_id,
workpackage_id=workpackage_id,
user_id=self.user_id,
role=role,
frontend_post_id=frontend_post_id,
frontend_message_id=frontend_message_id,
api_client_id=self.api_client.id,
payload_type=payload_type,
payload=PayloadContainer(payload=payload),
depth=depth,
)
self.db.add(post)
self.db.add(message)
self.db.commit()
self.db.refresh(post)
return post
self.db.refresh(message)
return message
def insert_reaction(self, work_package_id: UUID, payload: db_payload.ReactionPayload) -> PostReaction:
def insert_reaction(self, work_package_id: UUID, payload: db_payload.ReactionPayload) -> MessageReaction:
if self.user_id is None:
raise OasstError("User required", OasstErrorCode.USER_NOT_SPECIFIED)
container = PayloadContainer(payload=payload)
reaction = PostReaction(
reaction = MessageReaction(
work_package_id=work_package_id,
user_id=self.user_id,
payload=container,
@@ -383,98 +383,98 @@ class PromptRepository:
text=text_labels.text,
labels=text_labels.labels,
)
if text_labels.has_post_id:
self.fetch_post_by_frontend_post_id(text_labels.post_id, fail_if_missing=True)
model.post_id = text_labels.post_id
if text_labels.has_message_id:
self.fetch_message_by_frontend_message_id(text_labels.message_id, fail_if_missing=True)
model.message_id = text_labels.message_id
self.db.add(model)
self.db.commit()
self.db.refresh(model)
return model
def fetch_random_thread(self, require_role: str = None) -> list[Post]:
def fetch_random_thread(self, require_role: str = None) -> list[Message]:
"""
Loads all posts of a random thread.
Loads all messages of a random thread.
:param require_role: If set loads only thread which has
at least one post with given role.
at least one message with given role.
"""
distinct_threads = self.db.query(Post.thread_id).distinct(Post.thread_id)
distinct_threads = self.db.query(Message.thread_id).distinct(Message.thread_id)
if require_role:
distinct_threads = distinct_threads.filter(Post.role == require_role)
distinct_threads = distinct_threads.filter(Message.role == require_role)
distinct_threads = distinct_threads.subquery()
random_thread = self.db.query(distinct_threads).order_by(func.random()).limit(1)
thread_posts = self.db.query(Post).filter(Post.thread_id.in_(random_thread)).all()
return thread_posts
thread_messages = self.db.query(Message).filter(Message.thread_id.in_(random_thread)).all()
return thread_messages
def fetch_random_conversation(self, last_post_role: str = None) -> list[Post]:
def fetch_random_conversation(self, last_message_role: str = None) -> list[Message]:
"""
Picks a random linear conversation starting from any root post
Picks a random linear conversation starting from any root message
and ending somewhere in the thread, possibly at the root itself.
:param last_post_role: If set will form a conversation ending with a post
:param last_message_role: If set will form a conversation ending with a message
created by this role. Necessary for the tasks like "user_reply" where
the user should reply as a human and hence the last message of the conversation
needs to have "assistant" role.
"""
thread_posts = self.fetch_random_thread(last_post_role)
if not thread_posts:
thread_messages = self.fetch_random_thread(last_message_role)
if not thread_messages:
raise OasstError("No threads found", OasstErrorCode.NO_THREADS_FOUND)
if last_post_role:
conv_posts = [p for p in thread_posts if p.role == last_post_role]
conv_posts = [random.choice(conv_posts)]
if last_message_role:
conv_messages = [p for p in thread_messages if p.role == last_message_role]
conv_messages = [random.choice(conv_messages)]
else:
conv_posts = [random.choice(thread_posts)]
thread_posts = {p.id: p for p in thread_posts}
conv_messages = [random.choice(thread_messages)]
thread_messages = {p.id: p for p in thread_messages}
while True:
if not conv_posts[-1].parent_id:
if not conv_messages[-1].parent_id:
# reached the start of the conversation
break
parent_post = thread_posts[conv_posts[-1].parent_id]
conv_posts.append(parent_post)
parent_message = thread_messages[conv_messages[-1].parent_id]
conv_messages.append(parent_message)
return list(reversed(conv_posts))
return list(reversed(conv_messages))
def fetch_random_initial_prompts(self, size: int = 5):
posts = self.db.query(Post).filter(Post.parent_id.is_(None)).order_by(func.random()).limit(size).all()
return posts
messages = self.db.query(Message).filter(Message.parent_id.is_(None)).order_by(func.random()).limit(size).all()
return messages
def fetch_thread(self, thread_id: UUID):
return self.db.query(Post).filter(Post.thread_id == thread_id).all()
return self.db.query(Message).filter(Message.thread_id == thread_id).all()
def fetch_multiple_random_replies(self, max_size: int = 5, post_role: str = None):
parent = self.db.query(Post.id).filter(Post.children_count > 1)
if post_role:
parent = parent.filter(Post.role == post_role)
def fetch_multiple_random_replies(self, max_size: int = 5, message_role: str = None):
parent = self.db.query(Message.id).filter(Message.children_count > 1)
if message_role:
parent = parent.filter(Message.role == message_role)
parent = parent.order_by(func.random()).limit(1)
replies = self.db.query(Post).filter(Post.parent_id.in_(parent)).order_by(func.random()).limit(max_size).all()
replies = self.db.query(Message).filter(Message.parent_id.in_(parent)).order_by(func.random()).limit(max_size).all()
if not replies:
raise OasstError("No replies found", OasstErrorCode.NO_REPLIES_FOUND)
thread = self.fetch_thread(replies[0].thread_id)
thread = {p.id: p for p in thread}
thread_posts = [thread[replies[0].parent_id]]
thread_messages = [thread[replies[0].parent_id]]
while True:
if not thread_posts[-1].parent_id:
if not thread_messages[-1].parent_id:
# reached start of the conversation
break
parent_post = thread[thread_posts[-1].parent_id]
thread_posts.append(parent_post)
parent_message = thread[thread_messages[-1].parent_id]
thread_messages.append(parent_message)
thread_posts = reversed(thread_posts)
thread_messages = reversed(thread_messages)
return thread_posts, replies
return thread_messages, replies
def fetch_post(self, post_id: UUID) -> Optional[Post]:
return self.db.query(Post).filter(Post.id == post_id).one()
def fetch_message(self, message_id: UUID) -> Optional[Message]:
return self.db.query(Message).filter(Message.id == message_id).one()
def close_task(self, post_id: str, allow_personal_tasks: bool = False):
self.validate_post_id(post_id)
wp = self.fetch_workpackage_by_postid(post_id)
def close_task(self, message_id: str, allow_personal_tasks: bool = False):
self.validate_message_id(message_id)
wp = self.fetch_workpackage_by_message_id(message_id)
if not wp:
raise OasstError("Work package not found", OasstErrorCode.WORK_PACKAGE_NOT_FOUND)
oasst-shared/oasst_shared/schemas/protocol.py
+4 -4
View File
@@ -190,18 +190,18 @@ class TextReplyToPost(Interaction):
text: str
class PostRating(Interaction):
class MessageRating(Interaction):
"""A user has rated a post."""
type: Literal["post_rating"] = "post_rating"
type: Literal["post_rating"] = "message_rating"
post_id: str
rating: int
class PostRanking(Interaction):
class MessageRanking(Interaction):
"""A user has given a ranking for a post."""
type: Literal["post_ranking"] = "post_ranking"
type: Literal["post_ranking"] = "message_ranking"
post_id: str
ranking: list[int]