feat：接入mcp，为模型提供图表信息
This commit is contained in:
@@ -1,50 +1,206 @@
|
||||
"""从物料目录读取组件的可编辑字段，供动作生成理解字段含义，并供校验约束属性结构"""
|
||||
"""按物料展开字段定义：解析共享分组引用、取字段默认值、派生前端属性面板 setters"""
|
||||
|
||||
from app.resources.component import ComponentProvider
|
||||
from app.schemas.component_resource import ComponentDefinition, ComponentQuery
|
||||
from app.schemas.screen import ScreenNode
|
||||
from pydantic import JsonValue
|
||||
|
||||
# 属性面板的 key 以区域开头，只有这两个区域可以被 update 修改
|
||||
EDITABLE_SECTIONS = ("props", "style")
|
||||
from app.schemas.component_resource import ComponentCatalog, ComponentDefinition
|
||||
from app.schemas.material_field import (
|
||||
MISSING,
|
||||
PANEL_CONTROLS,
|
||||
FieldGroup,
|
||||
MaterialField,
|
||||
lookup,
|
||||
)
|
||||
|
||||
|
||||
def _materials(node_type: str) -> list[ComponentDefinition]:
|
||||
# 多个物料预设可以共用同一前端 type，按 type 取全部预设
|
||||
result = ComponentProvider().query(ComponentQuery(type=node_type))
|
||||
return result.components if result.status == "resolved" else []
|
||||
def resolve_field_groups(
|
||||
catalog: ComponentCatalog, component: ComponentDefinition
|
||||
) -> list[FieldGroup]:
|
||||
"""按物料声明的顺序展开字段分组，共享分组取目录中的定义；返回副本，不污染目录"""
|
||||
shared = {group.key: group for group in catalog.field_groups}
|
||||
return [
|
||||
(shared[group] if isinstance(group, str) else group).model_copy(deep=True)
|
||||
for group in component.field_groups
|
||||
]
|
||||
|
||||
|
||||
def _split_key(key: str) -> tuple[str, list[str | int]] | None:
|
||||
# 前端点路径中的纯数字段是数组下标，转换为与 UpdateAction.path 一致的整数
|
||||
section, *path = key.split(".")
|
||||
if section not in EDITABLE_SECTIONS or not path:
|
||||
return None
|
||||
return section, [int(part) if part.isdigit() else part for part in path]
|
||||
def field_default(component: ComponentDefinition, field: MaterialField) -> JsonValue:
|
||||
"""物料默认配置中有该路径时取默认配置的值，否则取字段声明的 default"""
|
||||
value = lookup(component.template(), field.segments)
|
||||
return field.default if value is MISSING else value
|
||||
|
||||
|
||||
def editable_fields(node_type: str) -> list[dict]:
|
||||
"""返回组件属性面板中的字段：用户可见名称、区域、路径及可选值，按出现顺序去重"""
|
||||
fields: dict[str, dict] = {}
|
||||
for material in _materials(node_type):
|
||||
for setter in material.setters:
|
||||
parsed = _split_key(setter.key)
|
||||
if parsed is None or setter.key in fields:
|
||||
def panel_setters(groups: list[FieldGroup]) -> list[dict]:
|
||||
"""派生前端属性面板的 setters，格式与前端 SetterSchema 一致"""
|
||||
setters = []
|
||||
for group in groups:
|
||||
for field in group.fields:
|
||||
if not field.panel:
|
||||
continue
|
||||
section, path = parsed
|
||||
field = {"label": setter.label, "section": section, "path": path}
|
||||
if options := setter.props.get("options"):
|
||||
field["options"] = options
|
||||
fields[setter.key] = field
|
||||
return list(fields.values())
|
||||
props = {}
|
||||
if field.options:
|
||||
props["options"] = [
|
||||
{"label": option.label, "value": option.value} for option in field.options
|
||||
]
|
||||
setter = {
|
||||
"type": PANEL_CONTROLS[field.value_type],
|
||||
"label": field.label,
|
||||
"key": field.path,
|
||||
"props": props,
|
||||
}
|
||||
if field.span is not None:
|
||||
setter["span"] = field.span
|
||||
setters.append(setter)
|
||||
return setters
|
||||
|
||||
|
||||
def known_prop_keys(node: ScreenNode) -> set[str] | None:
|
||||
"""props 第一层是组件与渲染器的约定，只能是节点已有或物料默认配置中的字段；
|
||||
目录中找不到该类型时约定未知，返回 None 表示不做限制"""
|
||||
materials = _materials(node.type)
|
||||
if not materials:
|
||||
return None
|
||||
keys = set(node.props)
|
||||
for material in materials:
|
||||
keys.update(material.schema_.props)
|
||||
return keys
|
||||
def frontend_catalog(catalog: ComponentCatalog) -> dict:
|
||||
"""前端物料接口的数据：在保存的定义上附加由字段定义派生的 setters"""
|
||||
data = catalog.model_dump(mode="json", by_alias=True)
|
||||
for item, component in zip(data["components"], catalog.components, strict=True):
|
||||
item["setters"] = panel_setters(resolve_field_groups(catalog, component))
|
||||
return data
|
||||
|
||||
|
||||
# ---- 字段查询：MCP 工具、动作提示词和动作校验共用同一套逻辑 ----
|
||||
|
||||
|
||||
def find_material(catalog: ComponentCatalog, component_type: str) -> ComponentDefinition:
|
||||
"""按节点 type 找物料；找不到或对应多个物料时报错，不猜测"""
|
||||
matches = [item for item in catalog.components if item.type == component_type]
|
||||
if len(matches) != 1:
|
||||
known = "、".join(item.type for item in catalog.components)
|
||||
reason = "没有对应的物料" if not matches else "对应多个物料"
|
||||
raise ValueError(f"组件类型 {component_type} {reason}，可用类型：{known}")
|
||||
return matches[0]
|
||||
|
||||
|
||||
def describe_field(
|
||||
component: ComponentDefinition, group: FieldGroup, field: MaterialField
|
||||
) -> dict:
|
||||
"""给模型看的字段说明：省略空值和属性面板专用的信息"""
|
||||
data = field.model_dump(
|
||||
mode="json",
|
||||
exclude={"default", "panel", "span", "editable"},
|
||||
exclude_none=True,
|
||||
exclude_defaults=True,
|
||||
)
|
||||
data["default"] = field_default(component, field)
|
||||
if not field.editable:
|
||||
data["editable"] = False
|
||||
data["group"] = group.key
|
||||
return data
|
||||
|
||||
|
||||
def list_field_groups(catalog: ComponentCatalog, component_type: str) -> dict:
|
||||
"""物料的字段分组目录：每组的说明和字段名称，模型据此决定查看哪几组"""
|
||||
component = find_material(catalog, component_type)
|
||||
return {
|
||||
"component_type": component.type,
|
||||
"name": component.name,
|
||||
"catalog_version": catalog.version,
|
||||
"groups": [
|
||||
{
|
||||
"key": group.key,
|
||||
"label": group.label,
|
||||
"description": group.description,
|
||||
"fields": [field.label for field in group.fields],
|
||||
}
|
||||
for group in resolve_field_groups(catalog, component)
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def get_fields(
|
||||
catalog: ComponentCatalog, component_type: str, group_keys: list[str]
|
||||
) -> dict:
|
||||
"""指定分组的完整字段定义"""
|
||||
component = find_material(catalog, component_type)
|
||||
groups = {group.key: group for group in resolve_field_groups(catalog, component)}
|
||||
if unknown := [key for key in group_keys if key not in groups]:
|
||||
raise ValueError(
|
||||
f"{component.type} 没有字段分组 {unknown}，可用分组：{list(groups)}"
|
||||
)
|
||||
return {
|
||||
"component_type": component.type,
|
||||
"catalog_version": catalog.version,
|
||||
"fields": [
|
||||
describe_field(component, groups[key], field)
|
||||
for key in dict.fromkeys(group_keys)
|
||||
for field in groups[key].fields
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _match_score(field: MaterialField, query: str) -> int:
|
||||
"""名称、别名完全一致最相关，其次是互相包含、路径末段一致，最后是说明中出现"""
|
||||
names = [name.casefold() for name in (field.label, *field.aliases)]
|
||||
if query in names:
|
||||
return 4
|
||||
if any(query in name or name in query for name in names):
|
||||
return 3
|
||||
if query == field.path.rsplit(".", 1)[-1].casefold():
|
||||
return 2
|
||||
if query in field.description.casefold() or query in field.path.casefold():
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
def search_fields(
|
||||
catalog: ComponentCatalog, component_type: str, query: str, limit: int = 8
|
||||
) -> dict:
|
||||
"""按名称、别名、路径和说明搜索字段，按相关度返回"""
|
||||
component = find_material(catalog, component_type)
|
||||
keyword = query.strip().casefold()
|
||||
if not keyword:
|
||||
raise ValueError("搜索内容不能为空")
|
||||
scored = [
|
||||
(score, index, group, field)
|
||||
for index, (group, field) in enumerate(
|
||||
(group, field)
|
||||
for group in resolve_field_groups(catalog, component)
|
||||
for field in group.fields
|
||||
)
|
||||
if (score := _match_score(field, keyword))
|
||||
]
|
||||
scored.sort(key=lambda item: (-item[0], item[1]))
|
||||
return {
|
||||
"component_type": component.type,
|
||||
"catalog_version": catalog.version,
|
||||
"query": query,
|
||||
"fields": [
|
||||
describe_field(component, group, field)
|
||||
for _, _, group, field in scored[:limit]
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def check_update_field(catalog: ComponentCatalog | None, node_type: str, action) -> None:
|
||||
"""update 只能修改字段定义中的可编辑字段，值须符合字段类型和取值约束
|
||||
|
||||
不合法时抛 ValueError，路径不存在时附上可能对应的字段供修正参考
|
||||
"""
|
||||
if catalog is None:
|
||||
raise ValueError("物料目录未载入，无法确认组件有哪些字段")
|
||||
component = find_material(catalog, node_type)
|
||||
fields = {
|
||||
field.path: field
|
||||
for group in resolve_field_groups(catalog, component)
|
||||
for field in group.fields
|
||||
}
|
||||
path = action.field_path
|
||||
field = fields.get(path)
|
||||
if field is None:
|
||||
message = f"{node_type} 没有字段 {path}，只能修改字段定义中的路径"
|
||||
keyword = next(
|
||||
(str(item) for item in reversed(action.path) if isinstance(item, str)),
|
||||
action.section,
|
||||
)
|
||||
if similar := search_fields(catalog, node_type, keyword, limit=3)["fields"]:
|
||||
message += "；可能是：" + "、".join(
|
||||
f"{item['label']}（{item['path']}）" for item in similar
|
||||
)
|
||||
raise ValueError(message)
|
||||
if not field.editable:
|
||||
raise ValueError(f"字段 {path}（{field.label}）不允许修改")
|
||||
if error := field.value_error(action.value):
|
||||
raise ValueError(f"字段 {path}（{field.label}）{error}")
|
||||
|
||||
Reference in New Issue
Block a user