File size: 2,613 Bytes
3d3d712 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | from __future__ import annotations
import secrets
from dataclasses import dataclass, field
from typing import List
from taskweaver.memory.plugin import PluginEntry
from taskweaver.memory.round import Round
from taskweaver.utils import create_id
from ..utils import read_yaml, validate_yaml
@dataclass
class Conversation:
"""A conversation denotes the interaction with the user, which is a collection of rounds.
The conversation is also used to construct the Examples.
Args:
id: the unique id of the conversation.
rounds: a list of rounds.
plugins: a list of plugins that are used in the conversation.
enabled: whether the conversation is enabled, used for Example only.
"""
id: str = ""
rounds: List[Round] = field(default_factory=list)
plugins: List[PluginEntry] = field(default_factory=list)
enabled: bool = True
plugin_only: bool = False
@staticmethod
def init():
"""init a conversation with empty rounds and plugins."""
return Conversation(
id="conv-" + create_id(),
rounds=[],
plugins=[],
enabled=True,
)
def add_round(self, round: Round):
self.rounds.append(round)
def to_dict(self):
"""Convert the conversation to a dict."""
return {
"id": self.id,
"plugins": [plugin.to_dict() for plugin in self.plugins],
"enabled": self.enabled,
"rounds": [_round.to_dict() for _round in self.rounds],
}
@staticmethod
def from_yaml(path: str) -> Conversation: # It is the same as from_dict
content = read_yaml(path)
do_validate = False
valid_state = False
if do_validate:
valid_state = validate_yaml(content, schema="example_schema")
if not do_validate or valid_state:
enabled = content["enabled"]
if "plugins" in content.keys():
plugins = [PluginEntry.from_yaml_content(plugin) for plugin in content["plugins"]]
else:
plugins = []
rounds = [Round.from_dict(r) for r in content["rounds"]]
if "plugin_only" in content.keys():
plugin_only = content["plugin_only"]
else:
plugin_only = False
return Conversation(
id="conv-" + secrets.token_hex(6),
rounds=rounds,
plugins=plugins,
enabled=enabled,
plugin_only=plugin_only,
)
raise ValueError("Yaml validation failed.")
|