修改指令json生成相关提示词和代码逻辑

lyg

2025-05-22 c099e6662b8a6e320ac314d31eda9b40455e5aa7

修改指令json生成相关提示词和代码逻辑

 db_struct_flow.py

@@ -12,7 +12,7 @@
import data_templates
from knowledgebase import utils
from knowledgebase.db.db_helper import create_project, create_device, create_data_stream, \
    update_rule_enc, create_extend_info, create_ref_ds_rule_stream, create_ins_format
    update_rule_enc, create_extend_info, create_ref_ds_rule_stream, create_ins_format, make_attr
from knowledgebase.db.data_creator import create_prop_enc, create_enc_pkt, get_data_ty, create_any_pkt

from knowledgebase.db.models import TProject
@@ -111,7 +111,16 @@
- 不输出任何注释等描述性信息。
"""

tc_system_msg = """
# 角色
你是一个资深软件工程师。
# 约束
- 输出内容必须根据文档和问题回答，不要创造其他内容；
- 输出内容必须是，JSON格式，不要输出其他文本。
"""

g_completion = None


def read_from_file(cache_file):
    with open(cache_file, 'r', encoding='utf-8') as f:
@@ -416,6 +425,9 @@
        return self.generate_text(msg, cache_file, msgs, doc_text, validation, try_cnt, True)

    def generate_tc_text(self, msg, cache_file, msgs=None, doc_text=None, validation=None, try_cnt=5):
        msgs = [
            {'role': 'system', 'content': tc_system_msg},
            {'role': 'user', 'content': "以下是文档内容：\n" + doc_text}]
        return self.generate_text(msg, cache_file, msgs, doc_text, validation, try_cnt, True)

    def gen_project(self):
@@ -905,6 +917,7 @@
]
        """
        print('遥测源包所属虚拟信道：')

        def validation(gen_text):
            pkts = json.loads(gen_text)
            assert len(pkts), 'VC源包列表不能为空'
@@ -1015,8 +1028,6 @@
        def validation(gen_text):
            json.loads(gen_text)

        

        doc_text = self.get_text_with_entity(['RT地址分配','分系统源包'])
        text = self.generate_text_json(_msg, 'out/总线.json', doc_text=doc_text,
                                       validation=validation)
@@ -1118,18 +1129,18 @@
        # 遥控包格式
        pkt_format = self.gen_tc_pkt_format()
        # 遥控包列表
        pkts = self.gen_tc_transfer_pkts()
        for pkt in pkts:
            # 遥控包数据区内容
            self.gen_tc_pkt_details(pkt)
            pkt['type'] = 'insUnit'
        instructions = self.gen_tc_transfer_pkts()
        for inst in instructions:
            # 遥控指令数据区内容
            self.gen_tc_pkt_details(inst)
            inst['type'] = 'insUnit'
            format_text = json.dumps(pkt_format, ensure_ascii=False)
            format_text = utils.replace_tpl_paras(format_text, pkt)
            format_text = utils.replace_tpl_paras(format_text, inst)
            pf = json.loads(format_text)
            pf['name'] = pkt['name']
            pf['code'] = pkt['code']
            pf['name'] = inst['name']
            pf['code'] = inst['code']
            data_area = next(filter(lambda x: x['name'] == '应用数据区', pf['children']))
            data_area['children'].append(pkt)
            data_area['children'].append(inst)
            frame['subPkts'].append(pf)
        self.order = 0

@@ -1143,7 +1154,7 @@
            elif item['type'] == 'length':
                return None
            elif item['type'] == 'checkSum':
                return json.dumps({"ChecksumType": "CRC-CCITT"})
                return json.dumps({"ChecksumType": item['value']['type']})
            elif item['type'] == 'subPkt':
                return json.dumps({"CanInput": False})
            elif item['type'] in ['combPkt', 'insUnitList', 'input']:
@@ -1152,26 +1163,10 @@
                return '{"MinLength":null,"MaxLength":null,"IsSubPackage":false,"InputParams":[],"OutPutParams":[],"MatchItems":[]}'
            elif item['type'] == 'pkt':
                return '''{"MaxLength":1024,"IsSplit8":false,"Split8Start":null,"Split8End":null,"PadCode":null,"Alignment":null,"InputParams":[],"OutPutParams":[],"MatchItems":[]}'''
            elif item['type'] == 'pktSeqCnt':
                return json.dumps({"FirstPackValue":"PackCount","MiddlePackValue":"PackIndex","LastPackValue":"PackIndex","IndependPackValue":"InsUnitCount"})
            elif 'value' in item:
                return item['value']

        def make_attr(ty: str):
            """
            获取字段定义的ATTR。

            位掩码，用于标识节点类型。
            类型：0~2 BinaryType;
                3~5 DataType;
                6~8: InputFormat;
                9 : IsSubPackage;
                10: IsSendFlag；
                11~13: ProcessMethod；
                14~16: ExpressionType；
                17~19: EnumType

            :param ty:
            :return:
            """

        def create_tc_format(parent_pk, field, parent_parent_pk=None):
            """
@@ -1199,11 +1194,20 @@
            if 'length' in field:
                field['bitWidth'] = field['length']
            field['bitOrder'] = None
            field['attr'] = 0
            field['attr'] = make_attr(field)
            if field['type'] == 'length' and 'value' in field and field['value']:
                val = field['value']
                field['range'] = val['start'] + "~" + val['end']
                field['formula'] = val['formula']
            # 即时输入长度为null则是变长字段，需要把类型改为variableLength
            if field['type'] == 'input' and field['length'] is None:
                field['type'] = 'variableLength'
            # 枚举值默认值设置
            if field['type'] == 'enum' and len(field['enums']) and not next(filter(lambda x: 'default' in x and x['default'], field['enums']), None):
                field['enums'][0]['default'] = True
            # 校验和
            if field['type'] == 'checkSum':
                field['range'] = f'{field["value"]["start"]}~{field["value"]["end"]}'
            ins_format = create_ins_format(self.proj.C_PROJECT_PK, parent_pk, field)
            ins_format_pk = ins_format.C_INS_FORMAT_PK
            if 'children' in field:
@@ -1242,11 +1246,12 @@
- 通过标志：const，二进制，以B结尾；
- 控制命令标志：const，二进制，以B结尾；
- 空闲位：const，二进制，以B结尾；
- 航天器标识：const，十六进制，以0x开头；
- 航天器标识：const，十六进制，以0x开头，如果是二进制或十进制需要转换为十六进制；
- 虚拟信道标识：sendFlag，发送标记，默认为“任务注入帧”，所有的值都要列举出来；
# 数据类型
- const：固定码字，数值，二进制以B结尾，十进制，十六进制以0x开头；
- sendFlag：发送标记，类似枚举，定义样例：[{"n":"name","v":"value","c":"code","default":true}]，n表示名称，v表示值，c表示code（没有空着），default表示是默认值；
- checkSum：校验和，如果是校验和类型还需要分析校验和的算法，并保存在value中，校验和算法包括：字节异或（ByteXOR）、累加和取反（SumNot）、累加和（AddSum）、应用循环冗余（CRC-CCITT）、CRC8（CRC8）、ISO和校验（ISOSum）、奇校验（Odd）、偶校验（Even）、其他（Other）
# 约束
- 以JSON格式输出；
- 仅输出JSON文本，不要输出任何其他文本。
@@ -1288,7 +1293,19 @@
- 命令正确应答: const，二进制；
- 源地址: const，十六进制。
# 数据类型
- const：固定码字，数值，二进制以B结尾，十进制，十六进制以0x开头；
- 固定码字：const，数值，二进制以B结尾，十进制，十六进制以0x开头；
- 长度：length，如果字段描述内容为数据区域的长度则表示是长度，长度的value为数值、null或范围定义，
- 枚举值：enum，
- 校验和：checkSum，如果是校验和类型还需要分析校验和的算法，并保存在value中，校验和算法包括：字节异或（ByteXOR）、累加和取反（SumNot）、累加和（AddSum）、应用循环冗余（CRC-CCITT）、CRC8（CRC8）、ISO和校验（ISOSum）、奇校验（Odd）、偶校验（Even）、其他（Other）
- 即时输入：input。
# 长度类型的范围定义描述
{"start": "起始字段code", "end": "结束字段code", "formula": "计算公式"}
- start：起始字段code，长度包括起始字段，字段描述中说明了起始字段，
- end：结束字段code，长度包括结束字段，字段描述中说明了结束字段，
- formula：计算公式，如果没有计算相关描述则表示不需要计算公式。
计算公式定义：
- BYTES：按字节计算；
- N-x：总字节数减x，例如总字节数减1的公式为N-1。
# 约束
- 以JSON格式输出；
- 仅输出JSON文本，不要输出任何其他文本。
@@ -1320,15 +1337,17 @@
# 指令
分析文档列出所有的遥控指令。
#  约束
- 应用过程标识：应用过程标识就是APID，一般会在名称后的括号中列出来
- code是指令代号，没有就空着
- 应用过程标识：应用过程标识就是APID，一般会在名称后的括号中列出来；
- code：指令代号，没有就空着；
- 应用数据区：提取表格中的应用数据区内容。
# 输出例子：
[{
"name": "xxx",
"code":"pkt",
"应用过程标识符":"0xAA",
"服务类型":"0x1",
"服务子类型":"0x2"
"服务子类型":"0x2",
"应用数据区": ""
}]
'''

@@ -1352,28 +1371,43 @@

# 指令
分析文档，从文档中提取遥控指令名称为“{tc_name}”代号为“{tc_code}”的指令应用数据区定义。

有些文档内容非常简单仅仅包含特定字节的内容描述，如果是这种文档，则每个特定字节的内容描述定义为一个字段，字段类型根据字节内容确定。
""" + """
# 约束
- code 如果没有明确定义则使用名称的英文翻译，尽量简短；
- length 自动转换为bit长度，必须是数值或null，不能为0；
- value 根据字段描述提取；
- enums 有些字段是枚举值，根据字段描述提取，枚举元素的数据结构为{"n":"","v":"","c":""}；
- 输出内容必须为严格的json，不能输出除json以外的任何内容。

# 字段类型
- 固定码字：const，
- 长度：length，
- 固定码字：const，数值，二进制以B结尾，十进制，十六进制以0x开头；
- 长度：length，如果字段描述内容为数据区域的长度则表示是长度，长度的value为数值、null或范围定义，
- 枚举值：enum，
- 校验和：checkSum，
- 即时输入：input。
- 校验和：checkSum，如果是校验和类型还需要分析校验和的算法是什么，并保存在value中，
- 即时输入：input，如果是即时输入value的值为空字符串。

# 长度类型的范围定义描述
{"start": "起始字段code", "end": "结束字段code", "formula": "计算公式"}
- start：起始字段code，长度包括起始字段，字段描述中说明了起始字段，
- end：结束字段code，长度包括结束字段，字段描述中说明了结束字段，
- formula：计算公式，如果没有计算相关描述则表示不需要计算公式。
计算公式定义：
- BYTES：按字节计算；
- N-x：总字节数减x，例如总字节数减1的公式为N-1。

# 字段类型分析方法
- 根据字段描述分析字段的类型；
- 字段描述中明确指定了字段值的，类型为const；
- 字段中没有明确指定字段值，但是罗列了取值范围的，类型为enum；
- 字段描述中没有明确指定字段值，但是罗列了取值范围的，类型为enum；
- 字段描述中如果没有明确指定字段值也没有罗列取值范围的，类型为input；
- 字段如果是和“长度”有关，类型为length；
- 如果和数据域有关，类型为const；
- 字段如果和校验和有关，类型为checkSum。
- 字段如果和校验和有关，类型为checkSum，分析校验和的算法，并保存在value中，校验和算法包括：字节异或（ByteXOR）、累加和取反（SumNot）、累加和（AddSum）、应用循环冗余（CRC-CCITT）、CRC8（CRC8）、ISO和校验（ISOSum）、奇校验（Odd）、偶校验（Even）、其他（Other）。

# 约束
- code 如果没有明确定义则使用名称的英文翻译，尽量简短；
- length 自动转换为bit长度，必须是数值、null或范围定义，不能为0；
- value 根据字段描述提取字段值，字段值一般为数值类型，需要根据字段类型来分析，如果是length类型value的值为范围定义；
- enums 枚举类型的字段必须要有enums，根据字段描述提取，枚举元素的数据结构为{"n":"","v":"","c":""}；
- length类型的范围定义中的start和end必须是生成结果中的字段code；
- 输出数据结构为数组，数组元素为字段信息；
- 输出内容必须为严格的json，不能输出除json以外的任何内容。

# 输出例子：
[
@@ -1383,18 +1417,52 @@
        "length": 8,
        "type": "const",
        "value": "0xAA"
    },
    {
        "name": "para2",
        "code": "para2",
        "length": 8,
        "type": "length",
        "value": {"start": "para1", "end": "data", "formula": "BYTES"}
    },
    {
        "name": "数据",
        "code": "data",
        "length": null,
        "type": "input",
        "value": ""
    }
    ...
]
"""

        def validation(gen_text):
            json.loads(gen_text)
            fields = json.loads(gen_text)
            for field in fields:
                if field['type'] == 'length':
                    if field['value'] is None:
                        raise Exception('length类型的value不能为空')
                    if 'start' not in field['value'] or 'end' not in field['value']:
                        raise Exception('length类型的value必须包含start和end')
                    if field['value']['start'] not in [f['code'] for f in fields]:
                        raise Exception('length类型的value的start字段必须在fields中')
                    if field['value']['end'] not in [f['code'] for f in fields]:
                        raise Exception('length类型的value的end字段必须在fields中')
                elif field['type'] == 'enum':
                    if 'enums' not in field:
                        raise Exception('enum类型的field必须包含enums')
                    if len(field['enums']) == 0:
                        raise Exception('enum类型的field的enums不能为空')
                    for enum in field['enums']:
                        if 'n' not in enum or 'v' not in enum:
                            raise Exception('enum类型的field的enums的元素必须包含n、v、c')
                        if enum['n'] == '' or enum['v'] == '':
                            raise Exception('enum类型的field的enums的元素不能为空')

        doc_text = self.get_text_with_entity([tc_name])
        if doc_text == '':
            return None
        text = self.generate_tc_text(_msg, f'out/遥控指令数据域-{tc_code}-{utils.to_file_name(tc_name)}.json',
            doc_text = pkt['应用数据区']
        text = self.generate_tc_text(_msg,
                                     f'out/遥控指令数据域-{tc_code}-{utils.to_file_name(tc_name)}.json',
                                     doc_text=doc_text,
                                     validation=validation)
        result = json.loads(text)

 knowledgebase/db/db_helper.py

@@ -1,4 +1,5 @@
import uuid
from enum import Enum

from sqlalchemy.orm import sessionmaker, scoped_session

@@ -384,10 +385,82 @@
    "checkSum": 20,
    "insUnit": 4,
    "insUnitList": 11,
    "input": 19
    "input": 19,
    "pktSeqCnt": 25,
    "variableLength": 14
}


class BinaryType(Enum):
    Integer = 0 # 整型
    Float = 1 # 浮点型
    Ascii = 2 # ASCII


class NumberDataType(Enum):
    Unsigned = 0 # 无符号整型
    SignInteger = 1 # 有符号整型
    Phy = 2 # 物理量
    Bytes = 3 # 多字节码


class InputFormat(Enum):
    Binary = 0 # 二进制
    Decimal = 1 # 十进制
    Hex = 2 # 十六进制


class ProcessMethod(Enum):
    Dirct = 0 # 直读
    Equivalent = 1 # 当量
    Formula = 2 # 公式
    Script = 3 # 脚本


class ExpressionType(Enum):
    Count = 0 # 个数计算
    Formula = 1 # 数值计算


class EnumType(Enum):
    NameValue = 0 # 名值对枚举
    Range = 1 # 范围枚举
    Classify = 2 # 多级分类
    InsCategory = 3 # 指令类别枚举
    InsUnit = 4 # 指令单元枚举
    InsFormat = 5 # 指令格式枚举


def make_attr(field: dict):
    """
    获取字段定义的ATTR。

    位掩码，用于标识节点类型。
    类型：0~2 BinaryType;
        3~5 DataType;
        6~8: InputFormat;
        9 : IsSubPackage;
        10: IsSendFlag；
        11~13: ProcessMethod；
        14~16: ExpressionType；
        17~19: EnumType

    :param field: 字段信息
    :return:
    """
    attr = 0
    # 即时输入，无符号整数，十进制，直读
    if field['type'] == ins_ty['input']:
        attr |= (NumberDataType.Unsigned.value << 3) | (InputFormat.Decimal.value << 6) | (ProcessMethod.Dirct.value << 11)
    # 是否是子包
    attr |= (1 << 9) if field['type']==ins_ty['subPkt'] else 0
    # 是否是发送标记
    attr |= (1 << 10) if field['type']==ins_ty['sendFlag'] else 0
    # 计算类型
    # 枚举类型

    return attr

def create_ins_format(proj_pk: str, parent_pk: str, info: dict) -> TInsFormat:
    ins_format = TInsFormat(
        C_INS_FORMAT_PK=get_pk(),

 knowledgebase/doc/doc_processor.py

@@ -50,7 +50,33 @@
        resp = llm.invoke([msg])
        Log.info(f'识别结果：{resp.content}')
        return resp.content

    def get_tc_info(self,  paragraph: ParagraphInfo):
        if self.doc_type not in [DocType.tc_format]:
            return ''
        prompt = HumanMessagePromptTemplate.from_template('''
# 指令
请从下面的文本中识别指令信息，如果识别失败不要输出任何字符。
指令信息包括：指令名称。
# 识别规则
- 文本内容为遥控指令数据域或遥控指令应用数据的定义描述。
# 约束
- 如果文本内容是目录则不要输出任何字符；
- 指令名称在章节标题中，提取指令名称要和文本中的严格一致；
- 如果没有识别到指令信息不要输出任何字符；
- 识别失败，不要输出任何内容，包括解释性文本；
- 输出json格式。
# 示例 - 识别到指令
{{
    "name": "xxx"
}}
# 示例 - 未识别到数据包
""
# 文本内容：
{text}
''')
        chain = prompt.prompt | llm | JsonOutputParser()
        resp = chain.invoke({"text": paragraph.full_text})
        return resp
    def get_tm_pkt_info(self, paragraph: ParagraphInfo):
        if self.doc_type not in [DocType.tm_outline, DocType.tm_pkt_design]:
            return ''
@@ -99,7 +125,18 @@
            e = doc_dbh.get_entity(entity)
            if e:
                entity.id = e.id
                return e
            else:
                doc_dbh.add_entity(entity)
                Log.info(f"新增Entity：{entity.name}，id：{entity.id}")
            paragraph.entities.append(entity)
        # 获取指令信息
        cmd = self.get_tc_info(paragraph)
        if cmd:
            entity = TEntity(name=cmd['name'], type='指令格式配置', prompts='', doc_type='')
            e = doc_dbh.get_entity(entity)
            if e:
                entity.id = e.id
            else:
            doc_dbh.add_entity(entity)
            Log.info(f"新增Entity：{entity.name}，id：{entity.id}")
            paragraph.entities.append(entity)

 testcases/test_doc_processor.py

@@ -22,7 +22,7 @@
        doc_processor = DocProcessor(file)
        doc_processor.process()
def test_get_text_by_entity():
    text = doc_dbh.get_text_with_entities(['分系统源包'])
    text = doc_dbh.get_text_with_entities(['指令组内容下传'])
    print(text)
if __name__ == '__main__':
    # test_process()

 tpl/tc_pkt_format.json

@@ -61,7 +61,7 @@
              "name": "包序列计数",
              "code": "packetSequenceCount",
              "length": 14,
              "type": "const",
              "type": "pktSeqCnt",
              "value": "0"
            }
          ]
@@ -73,7 +73,7 @@
          "type": "length",
          "value": {
            "start": "secondaryHeader",
            "end": "packetDataEnd",
            "end": "data",
            "formula": "N-1"
          }
        }
@@ -140,7 +140,8 @@
      "name": "包差错控制域",
      "code": "pktCheckSum",
      "length": 16,
      "type": "checkSum"
      "type": "checkSum",
      "value": {"start": "START", "end": "PREV", "type": "ISOSum"}
    }
  ],
  "subPkts": []

 tpl/tc_transfer_frame.json

@@ -47,7 +47,7 @@
                    "name": "虚拟信道标识",
                    "code": "vcid",
                    "length": 6,
                    "type": "enum",
                    "type": "sendFlag",
                    "enums": "{{虚拟信道标识}}"
                },
                {
@@ -76,7 +76,8 @@
            "name": "帧差错控制域",
            "code": "frameCRC",
            "length": 16,
            "type": "checkSum"
            "type": "checkSum",
            "value": {"start": "START", "end": "PREV", "type": "CRC-CCITT"}
        }
    ],
    "subPkts": []