feat(prompt): 优化系统提示词并增强节点验证逻辑
本次提交旨在提升大模型对行为树生成的理解能力和准确性，主要包含两方面的改进：丰富提示词内容和加强生成逻辑的验证。 具体文件变更如下： - **backend_service/src/prompts/system_prompt.txt**: - 在系统提示词中，增加了更多关于可用节点的详细说明。 - 补充了多个高质量的行为树生成示例（few-shot examples），以引导模型输出更符合预期的格式。 - **backend_service/src/py_tree_generator.py**: - 优化了对模型生成结果的验证规则，使其能更严格地检查节点的父子关系和参数合法性。 - 修复了当模型生成无效节点时可能出现的潜在错误。 - **backend_service/requirements.txt**: - 更新了相关依赖库版本。
This commit is contained in:
@@ -51,64 +51,124 @@ logging.basicConfig(
|
||||
|
||||
def _parse_allowed_nodes_from_prompt(prompt_text: str) -> tuple[Set[str], Set[str]]:
|
||||
"""
|
||||
从系统提示词中解析出允许的行动和条件节点。
|
||||
从系统提示词中精确解析出允许的行动和条件节点。
|
||||
"""
|
||||
try:
|
||||
# 使用正则表达式查找JSON代码块
|
||||
match = re.search(r"```json\s*({.*?})\s*```", prompt_text, re.DOTALL)
|
||||
# 使用更精确的正则表达式匹配节点定义部分
|
||||
node_section_pattern = r"#### 2\. 可用节点定义.*?```json\s*({.*?})\s*```"
|
||||
match = re.search(node_section_pattern, prompt_text, re.DOTALL | re.IGNORECASE)
|
||||
|
||||
if not match:
|
||||
logging.error("在系统提示词中未找到可用节点的JSON定义块。")
|
||||
return set(), set()
|
||||
|
||||
logging.error("在系统提示词中未找到'可用节点定义'部分的JSON代码块。")
|
||||
# 备用方案：尝试查找所有JSON块并识别节点定义
|
||||
return _fallback_parse_nodes(prompt_text)
|
||||
|
||||
json_str = match.group(1)
|
||||
logging.info("成功找到节点定义JSON代码块")
|
||||
|
||||
# 解析JSON
|
||||
allowed_nodes = json.loads(json_str)
|
||||
|
||||
# 从对象列表中提取节点名称
|
||||
actions = {action['name'] for action in allowed_nodes.get("actions", []) if 'name' in action}
|
||||
conditions = {condition['name'] for condition in allowed_nodes.get("conditions", []) if 'name' in condition}
|
||||
actions = set()
|
||||
conditions = set()
|
||||
|
||||
# 提取动作节点
|
||||
if "actions" in allowed_nodes and isinstance(allowed_nodes["actions"], list):
|
||||
for action in allowed_nodes["actions"]:
|
||||
if isinstance(action, dict) and "name" in action:
|
||||
actions.add(action["name"])
|
||||
|
||||
# 提取条件节点
|
||||
if "conditions" in allowed_nodes and isinstance(allowed_nodes["conditions"], list):
|
||||
for condition in allowed_nodes["conditions"]:
|
||||
if isinstance(condition, dict) and "name" in condition:
|
||||
conditions.add(condition["name"])
|
||||
|
||||
if not actions:
|
||||
logging.warning("关键错误：从提示词解析出的行动节点列表为空，无法生成任何有效任务。")
|
||||
logging.warning("关键错误：从提示词解析出的行动节点列表为空。")
|
||||
|
||||
logging.info(f"成功解析出动作节点: {sorted(actions)}")
|
||||
logging.info(f"成功解析出条件节点: {sorted(conditions)}")
|
||||
|
||||
return actions, conditions
|
||||
|
||||
except json.JSONDecodeError:
|
||||
logging.error("解析系统提示词中的JSON时失败。请检查格式。")
|
||||
|
||||
except json.JSONDecodeError as e:
|
||||
logging.error(f"解析节点定义JSON时失败: {e}")
|
||||
return set(), set()
|
||||
except Exception as e:
|
||||
logging.error(f"解析可用节点时发生未知错误: {e}")
|
||||
return set(), set()
|
||||
|
||||
def _fallback_parse_nodes(prompt_text: str) -> tuple[Set[str], Set[str]]:
|
||||
"""
|
||||
备用解析方案：当精确匹配失败时使用。
|
||||
"""
|
||||
logging.warning("使用备用方案解析节点定义...")
|
||||
|
||||
# 查找所有JSON代码块
|
||||
matches = re.findall(r"```json\s*({.*?})\s*```", prompt_text, re.DOTALL)
|
||||
if not matches:
|
||||
logging.error("在系统提示词中未找到任何JSON代码块。")
|
||||
return set(), set()
|
||||
|
||||
# 尝试从每个JSON块中解析节点定义
|
||||
for i, json_str in enumerate(matches):
|
||||
try:
|
||||
data = json.loads(json_str)
|
||||
|
||||
# 检查是否是节点定义的结构（包含actions、conditions、control_flow）
|
||||
if ("actions" in data and isinstance(data["actions"], list) and
|
||||
"conditions" in data and isinstance(data["conditions"], list) and
|
||||
"control_flow" in data and isinstance(data["control_flow"], list)):
|
||||
|
||||
actions = set()
|
||||
conditions = set()
|
||||
|
||||
# 提取动作节点
|
||||
for action in data["actions"]:
|
||||
if isinstance(action, dict) and "name" in action:
|
||||
actions.add(action["name"])
|
||||
|
||||
# 提取条件节点
|
||||
for condition in data["conditions"]:
|
||||
if isinstance(condition, dict) and "name" in condition:
|
||||
conditions.add(condition["name"])
|
||||
|
||||
if actions:
|
||||
logging.info(f"从第{i+1}个JSON块中成功解析出节点定义")
|
||||
logging.info(f"动作节点: {sorted(actions)}")
|
||||
logging.info(f"条件节点: {sorted(conditions)}")
|
||||
return actions, conditions
|
||||
|
||||
except json.JSONDecodeError:
|
||||
continue # 尝试下一个JSON块
|
||||
|
||||
logging.error("在所有JSON代码块中都没有找到有效的节点定义结构。")
|
||||
return set(), set()
|
||||
|
||||
def _generate_pytree_schema(allowed_actions: set, allowed_conditions: set) -> dict:
|
||||
"""
|
||||
根据允许的行动和条件节点，动态生成一个JSON Schema。
|
||||
"""
|
||||
# 所有可能的节点类型
|
||||
node_types = ["action", "condition", "Sequence", "Selector", "Parallel"]
|
||||
|
||||
# 递归节点定义
|
||||
node_definition = {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"type": {"type": "string", "enum": ["action", "condition", "Sequence", "Selector"]},
|
||||
"name": {"type": "string"},
|
||||
"type": {"type": "string", "enum": node_types},
|
||||
"name": {"type": "string"}, # 放宽对name的验证
|
||||
"params": {"type": "object"},
|
||||
"children": {
|
||||
"type": "array",
|
||||
"items": {"$ref": "#/definitions/node"}
|
||||
}
|
||||
},
|
||||
"required": ["type", "name"],
|
||||
# 使用 allOf 和 if/then 来实现基于'type'字段的条件性'name'验证
|
||||
"allOf": [
|
||||
{
|
||||
"if": {"properties": {"type": {"const": "action"}}},
|
||||
"then": {"properties": {"name": {"enum": sorted(list(allowed_actions))}}}
|
||||
},
|
||||
{
|
||||
"if": {"properties": {"type": {"const": "condition"}}},
|
||||
"then": {"properties": {"name": {"enum": sorted(list(allowed_conditions))}}}
|
||||
}
|
||||
]
|
||||
"required": ["type", "name"]
|
||||
}
|
||||
|
||||
|
||||
# 完整的Schema结构
|
||||
schema = {
|
||||
"$schema": "http://json-schema.org/draft-07/schema#",
|
||||
@@ -122,6 +182,7 @@ def _generate_pytree_schema(allowed_actions: set, allowed_conditions: set) -> di
|
||||
},
|
||||
"required": ["root"]
|
||||
}
|
||||
|
||||
return schema
|
||||
|
||||
def _validate_pytree_with_schema(pytree_instance: dict, schema: dict) -> bool:
|
||||
|
||||
Reference in New Issue
Block a user