Files
Odoo-Tutorial/quickboard/wizard/ai/quickboard_ai_generator.py
2025-08-19 15:05:55 +07:00

629 lines
27 KiB
Python

# -*- coding: utf-8 -*-
import json
import copy
from textwrap import dedent
from prettytable import PrettyTable
from autogen import Agent, AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
from autogen.coding import MarkdownCodeExtractor
from .consts import DEFAULT_AUTOGEN_LLM_CONFIG, QUICKBOARD_FG_COLORS, QUICKBOARD_BG_COLORS, QUICKBOARD_DATA_ONLY_JSON_SCHEMA, QUICKBOARD_LAYOUT_JSON_SCHEMA
from .json_validator_agent import UserProxyAgentForJsonValidation
class QuickboardAiGenerator:
_QUICKBOARD_GENERATOR_SYSTEM_MESSAGE = f"""
You are an AI assistant specializing in data analysis.
Your task is to assist user to build an intuitive and informative dashboard.
You help by creating a list dashboard items base on the fields from the given tables.
Ignore all your previous knowledge about models, here models are the same with relational database tables.
Only use the models as mentioned here with common sense, e.g. when the model name is 'sale.order'
then the model is about sales orders, etc.
## DASHBOARD ITEM TYPES
There are two kind of dashboard items, 'basic', 'list' and 'chart'.
A 'basic' item is used for single value KPI, 'list' items are usualy used to show top (n) data in a tabular list
while a 'chart' item is used to show data in a chart.
Both shared the following common parameters:
1. name (string): The title of the item. Required.
2. icon (string): Font awesome version 4.7 icon in 'fa-*' format. Required.
3. type (string): Dashboard item type. Required.
4. model (string): model name, e.g. "sale.order", "product.product", etc. Required.
5. value_field (string): Field of the chosen model for the value to be shown, only accept one field. Required.
6. aggregate_function (string): Function to be applied on the value field. Required.
Basic item has the following parameters in addition of the common parameters above:
1. text_color (string): Text color for the item. Required.
2. background_color (string): Background color for the item. Required.
Chart item has the following parameters in addition of the common parameters above:
1. dimension_field (string): Field of the chosen model for the grouping of the data, only accept one field. Required.
2. chart_type (string): The chart type. Required.
List item has the following parameters in addition of the common parameters above:
1. dimension_field (string): Field of the chosen model for the grouping of the data, only accept one field. Required.
2. list_row_limit (integer): Number of rows in the list. Required.
## COLORS
Use these color palette for text color parameters: {QUICKBOARD_FG_COLORS}
Use these color palette for background color parameters: {QUICKBOARD_BG_COLORS}
#EXAMPLE
Given a model with name 'sale.order' which have the following fields:
+-------------+------------------------+-----------+-------------------------+-------------------+
| Model | Name | Type | Description | Usage |
+-------------+------------------------+-----------+-------------------------+-------------------+
| ... | ... | ... | ... | ... |
| sale.order | id | integer | ID | value, dimension |
| sale.order | date_order | datetime | Order Date | dimension |
| sale.order | medium_id | integer | Medium | dimension |
| sale.order | sale_order_option_ids | list | Optional Products Lines | |
| sale.order | amount_total | monetary | Total | value |
| sale.order | amount_to_invoice | monetary | Amount to invoice | value |
| ... | ... | ... | ... | ... |
+-------------+------------------------+-----------+-------------------------+-------------------+
And a model with name 'sale.order.line' which have the following fields:
+------------------+---------------------------+-----------+--------------------------------+-------------------+
| Model | Name | Type | Description | Usage |
+------------------+---------------------------+-----------+--------------------------------+-------------------+
| ... | ... | ... | ... | ... |
| sale.order.line | product_id | integer | Product | value, dimension |
| sale.order.line | product_uom_qty | float | Quantity | value |
| sale.order.line | qty_delivered_method | char | Method to update delivered qty | value, dimension |
| sale.order.line | qty_delivered | float | Delivery Quantity | value |
| ... | ... | ... | ... | ... |
+------------------+---------------------------+-----------+--------------------------------+-------------------+
You may choose to answer as follow:
```json
[
{{
\"name\": \"Sales Order Count\",
\"icon\": \"fa-shopping-bag\",
\"type\": \"basic\",
\"model\": \"sale.order\",
\"value_field\": \"id\",
\"aggregate_function\": \"count\",
\"text_color\": \"#000000\",
\"background_color\": \"#FFFFFF\"
}},
{{
\"name\": \"Sales Order Total\",
\"icon\": \"fa-shopping-cart\",
\"type\": \"basic\",
\"model\": \"sale.order\",
\"value_field\": \"amount_total\",
\"aggregate_function\": \"sum\",
\"text_color\": \"#000000\",
\"background_color\": \"#3b3b3b\"
}},
{{
\"name\": \"Total Amount By Date\",
\"icon\": \"fa-usd\",
\"type\": \"chart\",
\"model\": \"sale.order\",
\"value_field\": \"amount_total\",
\"aggregate_function\": \"sum\",
\"dimension_field\": \"date_order\",
\"datetime_granularity\": \"day\",
\"chart_type\": \"line\"
}},
{{
\"name\": \"Top 10 Products\",
\"icon\": \"fa-shopping-bag\",
\"type\": \"list\",
\"model\": \"sale.order.line\",
\"value_field\": \"product_uom_qty\",
\"aggregate_function\": \"sum\",
\"dimension_field\": \"product_id\",
\"list_row_limit\": 10
}}
]
```
## RULES
1. Answer only with the definitions of the items in a json list fenced in json code block.
2. Do not comment. Do not explain your answer.
3. Do not use models other than the user specifies.
4. Pay attention to which field belong to which model. Do not to use fields from other models.
5. Treat each model independently, do not mix fields from one model to another.
6. All parameters are required, never set any parameter to null.
7. Set parameter 'type' to 'basic' for basic items, set it to 'list' for list items and set it to 'chart' for chart items.
8. Parameter 'chart_type' must be one of ['bar', 'horizontal-bar', 'doughnut', 'line', 'pie', 'polar'].
9. Use one of ['avg', 'count', 'max', 'min', 'sum'] for 'aggregate_function' if the 'value_field' is one of ['integer', 'float', 'monetary'].
10. For other types of 'value_field' such as 'char', 'many2one', etc., the parameter 'aggregate_function' must only be 'count'.
11. Never use field with type 'date' or 'datetime' for 'value_field'.
12. Parameter 'value_field' and 'dimension_field', requires exact field name, use the field as is without prefixes nor suffixes.
13. If the type of 'dimension_field' is date or datetime, you may add 'datetime_granularity' parameter to specify the precision.
'datetime_granularity' must be one of ["year", "month", "day"].
14. Do not assume a field has relation to other model, so again, 'value_field' and 'dimension_field' must use the exact name as mentioned here.
15. Use only single color for color related parameters, not an array of colors, choose one of the colors mentioned above.
16. Fence your anwser with markdown json code block (```json your_answer ```).
17. Your answer will be validated by a bot, if your answer is not valid then you must fix it.
18. When fixing answer re-evaluate everything and do not give comment on the json code block as it will create another error.
19. When fixing answer always reply with the the fixed json code block with every items.
"""
def __init__(self, env):
self.env = env
self._admin: UserProxyAgent = None
self._quickboard_ai: AssistantAgent = None
self._json_validator: UserProxyAgentForJsonValidation = None
self._groupchat: GroupChat = None
self._manager: GroupChatManager = None
def _create_agents(self, data_json_schema):
admin = UserProxyAgent(
"admin",
description="The user who give tasks and questions.",
human_input_mode="NEVER",
is_termination_msg=lambda message: True, # Always True
code_execution_config=False,
)
quickboard_generator = AssistantAgent(
name="quickboard_generator",
description=f"AI that generate dashboards.",
system_message=dedent(self._QUICKBOARD_GENERATOR_SYSTEM_MESSAGE),
human_input_mode="NEVER",
llm_config=DEFAULT_AUTOGEN_LLM_CONFIG,
)
data_json_validator = UserProxyAgentForJsonValidation(
"data_json_validator",
json_schema=data_json_schema,
description="An bot that performs no other action than validating json (provided to it's quoted in json blocks).",
human_input_mode="NEVER",
)
def _speaker_selection_func(last_speaker: Agent, groupchat: GroupChat):
last_messages = groupchat.messages[-1]
next_speaker = admin
if last_speaker is admin:
next_speaker = quickboard_generator
elif last_speaker is quickboard_generator:
next_speaker = data_json_validator
elif last_speaker is data_json_validator:
# if last agent reply with invalid json then let it try again
if last_messages["content"].find("exitcode: -1") > -1:
next_speaker = quickboard_generator
elif last_messages["content"].strip() == "":
next_speaker = quickboard_generator
return next_speaker
groupchat = GroupChat(
agents=[admin, quickboard_generator, data_json_validator],
messages=[],
max_round=5,
speaker_selection_method= _speaker_selection_func,
# send_introductions=True,
)
manager = GroupChatManager(
groupchat=groupchat,
name="chat_manager",
llm_config=DEFAULT_AUTOGEN_LLM_CONFIG
)
return (admin, quickboard_generator, data_json_validator, manager, groupchat)
def _build_agent_parameters(self, models):
model_names = []
model_infos = []
data_json_schema = copy.deepcopy(QUICKBOARD_DATA_ONLY_JSON_SCHEMA)
for model in models:
field_defs = PrettyTable()
field_defs.align = "l"
field_defs.field_names = ["Model", "Name", "Type", "Description", "Usage"]
fields = self.env[model.model].fields_get()
valid_value_fields = []
valid_dimension_fields = []
for k, v in fields.items():
if v["store"]:
if v["type"] in ["many2many", "one2many"]:
field_type = "list"
elif v["type"] == "selection":
field_type = "char"
elif v["type"] == "many2one":
field_type = "integer"
else:
field_type = v["type"]
usage = []
if v["type"] not in ["many2many", "one2many", "float", "monetary"]:
valid_dimension_fields.append(f"{k}")
usage.append("dimension")
if v["type"] in ["integer", "float", "monetary", "many2one", "selection"]:
valid_value_fields.append(f"{k}")
usage.append("value")
field_defs.add_row([model.model, f"{k}", field_type, v['string'], ",".join(usage)])
valid_value_fields_str = ", ".join([f"'{o}'" for o in valid_value_fields])
valid_dimension_fields_str = ", ".join([f"'{o}'" for o in valid_dimension_fields])
model_names.append(model.model)
model_infos.append({
"name": model.model,
"field_defs": field_defs.get_string(),
"value_fields": valid_value_fields_str,
"dimension_fields": valid_dimension_fields_str
})
model_name_schema = {
"const": model.model
}
value_field_schema = {
"if": {
"properties": {
"model": model_name_schema
}
},
"then": {
"properties": {
"value_field": {
"enum": valid_value_fields
}
}
}
}
dimension_field_schema = {
"if": {
"properties": {
"model": model_name_schema
}
},
"then": {
"properties": {
"dimension_field": {
"enum": valid_dimension_fields
}
}
}
}
data_json_schema["items"]["model"] = model_name_schema
data_json_schema["items"]["allOf"].append(value_field_schema)
data_json_schema["items"]["allOf"].append(dimension_field_schema)
return model_names, model_infos, data_json_schema
def _create_message(self, model_names, model_infos):
message = dedent(f"""
Create dashboard items from these models {[o for o in model_names]}.
""")
for mi in model_infos:
mi_string = dedent(f"""
### MODEL '{mi["name"]}'
The model '{mi["name"]}' has the following fields:
{mi["field_defs"]}
""")
message = message + "\n" + mi_string
message += dedent("""
Pay attention to the field usage. Valid values for 'value_field' and 'dimension_field' are based on it.
Also pay attention to which model a field belongs to, do not use other models field in a dashboard item.
Always use the exact field name as mentioned above on the models field.
Create at least 3 'basic' items, 3 'list' items and 3 'chart' items.
Make the 'basic' items background colorful with matching but still readable text color.
Answer only in a json list fenced in a json code block.
""")
return message
def _arrange_items(self, quickboard_items):
items = copy.deepcopy(quickboard_items)
basic_items = [o for o in items if o["type"] == "basic"]
list_chart_items = [o for o in items if o["type"] != "basic"]
fin = []
row = 0
while len(basic_items) > 0:
count = 1
width = 6
if len(basic_items) >= 6:
count = 6
width = 2
elif len(basic_items) >= 4:
count = 4
width = 3
elif len(basic_items) >= 3:
count = 3
width = 4
elif len(basic_items) >= 2:
count = 2
width = 6
row_items = basic_items[:count]
for k, o in enumerate(row_items):
o["y_pos"] = row
o["x_pos"] = k * width
o["height"] = 1
o["width"] = width
fin.extend(row_items)
del basic_items[:count]
row += 1
while len(list_chart_items) > 0:
width = 4
count = 3
if len(list_chart_items) >= 4:
count = 4
width = 3
elif len(list_chart_items) >= 3:
count = 3
width = 4
elif len(list_chart_items) >= 2:
count = 2
width = 6
row_items = list_chart_items[:count]
for k, o in enumerate(row_items):
o["y_pos"] = row
o["x_pos"] = k * width
o["height"] = 2
o["width"] = width
fin.extend(row_items)
del list_chart_items[:count]
row += 2
return fin
def generate_quickboard(self, models, layout_by_ai, screen_w, screen_h, cell_w, cell_h):
model_names, model_infos, json_schema = self._build_agent_parameters(models)
self._admin,\
self._quickboard_ai,\
self._json_validator,\
self._manager,\
self._groupchat = self._create_agents(json_schema)
message = self._create_message(model_names, model_infos)
answer = self._admin.initiate_chat(self._manager, message=message) #, silent=True)
res = ""
# The json is not on answer.summary / last message since the last agent is json validator when successful
if answer.summary.find("exitcode: 0") > -1:
extractor = MarkdownCodeExtractor()
code_blocks = extractor.extract_code_blocks(answer.chat_history[-2]["content"])
if len(code_blocks) > 0:
quickboard = code_blocks[0].code
quickboard_items = json.loads(quickboard)
arranged_items = []
if layout_by_ai:
aiDesigner = QuickboardAIDesigner()
arranged_items = aiDesigner.arrange_items_with_ai(quickboard_items, screen_w, screen_h, cell_w, cell_h)
else:
arranged_items = self._arrange_items(quickboard_items)
res = json.dumps(arranged_items)
return res
class QuickboardAIDesigner:
_QUICKBOARD_DESIGNER_SYSTEM_MESSAGE = f"""
Your task is to design the layout of a dashboard from the given dashboard items.
## DASHBOARD ITEM TYPES
There are three kind of dashboard items, basic, list and chart.
A basic item is used for single value KPI, a list is used to display data in a tabular list
while a chart item is used to show data in a graphical chart.
All have the following attribute:
1. id: The id of the dashboard item.
2. type: Dashboard item type, 'basic' for basic items, 'list' for list items and 'chart' for chart items.
3. x_pos: The horizontal position on the grid (0 to 11, representing the block's position).
4. y_pos: The vertical position on the grid (measured in blocks, can be unlimited).
5. width: The width of the item.
6. height: The height of the item.
## LAYOUT
The dashboard layout is a grid system measured in square blocks.
It has 12 blocks for column width and unlimited cell rows.
The origin (0, 0) position is on the top left. The maximum position of the top row is (12, 0).
## REQUIREMENTS
1. Prioritize the arrangement of blocks to optimize space, i,e. no gaps between items.
2. Ensure that no blocks overlap and that all blocks are positioned within the defined grid limits.
3. Arrange the 'basic' items to fill the top rows.
4. Arrange the rest of the items to fill the rows after the 'basic' items.
## RULES
1. Do not change the 'id' and the 'type' of the dashboard items,
i.e, the 'id' and 'type' is a fixed pair. If you want to re-arrange the items, always use the same 'id' and 'type' pair.
2. Only edit these attribute 'x_pos', 'y_pos', 'width', 'height'.
3. Do not add nor remove dashboard items!
4. Fence your anwser with markdown json block.
5. Do not comment. Do not explain your answer.
6. Your answer will be validated by a bot, if your answer is not valid then you must fix it.
7. When fixing answer re-evaluate everything and do not give comment as it will create another error.
8. When fixing answer always reply with the the fixed json code block with every items.
"""
def __init__(self):
self._admin: UserProxyAgent = None
self._quickboard_ai: AssistantAgent = None
self._json_validator: UserProxyAgentForJsonValidation = None
self._groupchat: GroupChat = None
self._manager: GroupChatManager = None
def _create_agents(self, ui_json_schema):
admin = UserProxyAgent(
"admin",
description="The user who give tasks and questions.",
human_input_mode="NEVER",
is_termination_msg=lambda message: True, # Always True
code_execution_config=False,
)
quickboard_designer = AssistantAgent(
name="quickboard_designer",
description=f"AI that design the layout of dashboards.",
system_message=dedent(self._QUICKBOARD_DESIGNER_SYSTEM_MESSAGE),
human_input_mode="NEVER",
llm_config=DEFAULT_AUTOGEN_LLM_CONFIG,
)
ui_json_validator = UserProxyAgentForJsonValidation(
"ui_json_validator",
json_schema=ui_json_schema,
description="An bot that performs no other action than validating json (provided to it's quoted in json blocks).",
human_input_mode="NEVER",
)
def _speaker_selection_func(last_speaker: Agent, groupchat: GroupChat):
last_messages = groupchat.messages[-1]
next_speaker = admin
if last_speaker is admin:
next_speaker = quickboard_designer
elif last_speaker is quickboard_designer:
next_speaker = ui_json_validator
elif last_speaker is ui_json_validator:
# if last agent reply with invalid json then let it try again
if last_messages["content"].find("exitcode: -1") > -1:
next_speaker = quickboard_designer
elif last_messages["content"].strip() == "":
next_speaker = quickboard_designer
return next_speaker
groupchat = GroupChat(
agents=[admin, quickboard_designer, ui_json_validator],
messages=[],
max_round=5,
speaker_selection_method= _speaker_selection_func,
# send_introductions=True,
)
manager = GroupChatManager(
groupchat=groupchat,
name="chat_manager",
llm_config=DEFAULT_AUTOGEN_LLM_CONFIG
)
return (admin, quickboard_designer, ui_json_validator, manager, groupchat)
def _build_agent_parameters(self, quickboard_items):
layout_json_schema = copy.deepcopy(QUICKBOARD_LAYOUT_JSON_SCHEMA)
item_count = len(quickboard_items)
item_count_schema = {
"minItems": item_count,
"maxItems": item_count
}
layout_json_schema.update(item_count_schema)
item_id_type_schema = []
for item in quickboard_items:
id_type_schema = {
"if": {
"properties": {
"id": { "const": item["id"] }
}
},
"then": {
"properties": {
"type": { "const": item["type"] }
}
}
}
item_id_type_schema.append(id_type_schema)
layout_json_schema["items"]["allOf"] = item_id_type_schema
return layout_json_schema
def _create_message(self, quickboard_items, screen_w, screen_h, cell_w, cell_h):
layout_items = [[{
"id": o["id"],
"type": o["type"],
"x_pos": 0,
"y_pos": 0,
"width": 0,
"height": 0
}] for o in quickboard_items]
layout_items_count = len(layout_items)
layout_items_str = ", ".join([ f"{json.dumps(o)}\n" for o in layout_items])
message = f"""
Design dashboard layout for these {layout_items_count} items: {layout_items_str}.
The cell size for my screen is {cell_w} x {cell_h} (width x height).
My screen size is {screen_w} x {screen_h} (width x height).
Make it compact, for 'basic' items 1 block for its height is enough.
For 'list' and 'chart' items 3 blocks for its height are enough.
Put basic items first then list and chart items.
Do not change the 'id' and the 'type' of the items.
Remember there are {layout_items_count}, use them all, do not add any item nor remove any of them.
"""
return message
def arrange_items_with_ai(self, quickboard_items, screen_w, screen_h, cell_w, cell_h):
res = quickboard_items
for i, item in enumerate(quickboard_items, start=1):
item["id"] = i
json_schema = self._build_agent_parameters(quickboard_items)
self._admin,\
self._quickboard_ai,\
self._json_validator,\
self._manager,\
self._groupchat = self._create_agents(json_schema)
message = self._create_message(quickboard_items, screen_w, screen_h, cell_w, cell_h)
answer = self._admin.initiate_chat(self._manager, message=message) #, silent=True)
if answer.summary.find("exitcode: 0") > -1:
extractor = MarkdownCodeExtractor()
code_blocks = extractor.extract_code_blocks(answer.chat_history[-2]["content"])
if len(code_blocks) > 0:
layout = code_blocks[0].code
layout_items = json.loads(layout)
# merge item and layout
fin_items = []
for item in quickboard_items:
for layout in layout_items:
if layout["id"] == item["id"] and layout["type"] == item["type"]:
item.update(layout)
break
fin_items.append(item)
return fin_items
return res