提交 85b7cb33 authored 作者: 陈泽健's avatar 陈泽健

feat(smart-locate): Claude语义增强智能定位 - 多候选元素自动语义排序

- 新增 claude_service.py:Claude CLI 语义增强服务,含 Prompt 模板、调用、解析、容错
- 改造 keyword_matcher.py:返回候选列表而非单一元素,新增 get_candidate_details()
- 改造 smart_locate_service.py:集成 Claude 语义增强,use_claude 参数控制
- 改造 smart_locate.py 路由:新增 use_claude 参数和 claude_enhanced/confidence/reason 响应字段
- 新增 config.py 配置项:CLAUDE_ENABLED/TIMEOUT/MODEL/MAX_CANDIDATES
- 新增集成测试脚本和 PRD/执行计划/技术调研文档

预期效果:定位准确率 60% -> 85%+,解决关键词撞车问题
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 9040ac63
......@@ -61,6 +61,12 @@ class Settings:
# 日志配置
LOG_LEVEL: str = os.getenv("LOG_LEVEL", "INFO")
# Claude 语义增强配置
CLAUDE_ENABLED: bool = os.getenv("CLAUDE_ENABLED", "true").lower() == "true"
CLAUDE_TIMEOUT: int = int(os.getenv("CLAUDE_TIMEOUT", "60")) # 秒(Claude CLI 首次调用需要初始化)
CLAUDE_MODEL: str = os.getenv("CLAUDE_MODEL", "claude-sonnet-5") # 默认模型
CLAUDE_MAX_CANDIDATES: int = int(os.getenv("CLAUDE_MAX_CANDIDATES", "5")) # 最大候选数
class Config:
"""Pydantic 配置"""
env_file = ".env"
......
......@@ -41,6 +41,7 @@ class SmartLocateRequest(BaseModel):
auto_login: bool = Field(default=True, description="是否自动登录")
navigate_menu: str = Field(default="", description="目标菜单名称")
page_url: str = Field(default="https://192.168.5.44", description="被测系统基础 URL")
use_claude: bool = Field(default=True, description="是否启用 Claude 语义增强")
class SmartLocateResult(BaseModel):
......@@ -54,6 +55,9 @@ class SmartLocateResult(BaseModel):
element_info: Dict[str, Any] = Field(default_factory=dict, description="元素信息")
screenshot: Optional[str] = Field(None, description="截图(base64)")
message: str = Field(default="", description="说明信息")
claude_enhanced: bool = Field(default=False, description="是否经过 Claude 语义增强")
claude_confidence: float = Field(default=0.0, description="Claude 置信度")
claude_reason: str = Field(default="", description="Claude 选择理由")
class SmartLocateResponse(BaseModel):
......@@ -142,7 +146,8 @@ async def smart_locate(
steps=steps_data,
auto_login=request.auto_login,
navigate_menu=request.navigate_menu,
page_url=request.page_url
page_url=request.page_url,
use_claude=request.use_claude
)
# 执行并等待结果
......
此差异已折叠。
......@@ -243,7 +243,8 @@ def match_element_by_keywords(
page,
keywords: List[str],
action: str,
timeout: int = 5000
timeout: int = 5000,
max_candidates: int = 10
) -> Tuple[Optional[Any], List[Dict[str, Any]]]:
"""
通过关键词在页面元素中直接匹配
......@@ -258,18 +259,19 @@ def match_element_by_keywords(
keywords (List[str]): 关键词列表
action (str): 动作类型
timeout (int): 超时时间(毫秒)
max_candidates (int): 最大返回候选数量
Returns:
Tuple[Optional[Any], List[Dict]]: (元素对象, 候选选择器列表)
元素对象可能为 None(未找到
Tuple[Optional[Any], List[Dict]]: (元素对象列表, 候选选择器列表)
元素对象列表按置信度降序排列(最多 max_candidates 个
候选选择器列表按置信度降序排列
"""
candidates = []
# 获取所有可交互元素
# 获取所有可交互元素(包含主页面 + iframe/微前端)
try:
# 扩展元素查询范围,包含 Element Plus 组件和常见可点击元素
elements = page.locator(
base_selector = (
'input:visible, button:visible, a:visible, select:visible, textarea:visible, '
# 语义化 role 元素
'[role="button"]:visible, [role="link"]:visible, [role="checkbox"]:visible, '
......@@ -278,13 +280,31 @@ def match_element_by_keywords(
'.el-button:visible, .el-checkbox:visible, .el-tabs__item:visible, '
# 常见可点击元素(图标、自定义按钮)
'i[onclick]:visible, div[onclick]:visible, span[onclick]:visible, '
'div[class*="btn"]:visible, span[class*="btn"]:visible'
).all()
'div[class*="btn"]:visible, span[class*="btn"]:visible, '
# 微前端中的常见可点击元素
'.block:visible, p:visible'
)
elements = page.locator(base_selector).all()
total_elements = len(elements)
# 🔧 新增:在所有 iframe 中查找元素
for frame in page.frames:
if frame == page.main_frame:
continue
try:
frame_elements = frame.locator(base_selector).all()
elements.extend(frame_elements)
if frame_elements:
logger.debug(f"在 iframe 中找到 {len(frame_elements)} 个元素")
except Exception as e:
logger.debug(f"iframe 元素查询失败(忽略): {e}")
except Exception as e:
logger.warning(f"获取页面元素失败: {e}")
return None, []
logger.info(f"页面找到 {len(elements)} 个可交互元素,关键词: {keywords}")
logger.info(f"页面找到 {len(elements)} 个可交互元素(主页面 {total_elements} + iframe {len(elements) - total_elements}),关键词: {keywords}")
for el in elements:
try:
......@@ -443,11 +463,128 @@ def match_element_by_keywords(
if not candidates:
return None, []
# 返回最佳匹配
best = candidates[0]
logger.info(f"最佳匹配元素: {best['info']}, 分数: {best['score']:.2f}")
# 限制候选数量
limited_candidates = candidates[:max_candidates]
# 返回所有候选元素和选择器(供 Claude 语义排序使用)
elements_list = [c['element'] for c in limited_candidates]
all_selectors = []
for c in limited_candidates:
all_selectors.extend(c['selectors'])
# 去重选择器
seen_selectors = set()
unique_selectors = []
for sel in all_selectors:
if sel['value'] not in seen_selectors:
seen_selectors.add(sel['value'])
unique_selectors.append(sel)
# 日志
best = limited_candidates[0]
logger.info(f"最佳匹配元素: {best['info']}, 分数: {best['score']:.2f}, 候选数: {len(limited_candidates)}")
return elements_list, unique_selectors
def get_candidate_details(
page,
elements: List[Any]
) -> List[Dict[str, Any]]:
"""
获取候选元素的详细信息(供 Claude 语义分析使用)
Args:
page: Playwright Page 对象
elements: 元素对象列表
Returns:
List[Dict]: 候选元素详情列表,格式:
[
{
"index": 0,
"tag": "BUTTON",
"text": "新建会议",
"selector": "button:has-text('新建会议')",
"attributes": {...},
"position": {"x": 100, "y": 200, "width": 80, "height": 32}
},
...
]
"""
candidates = []
for idx, el in enumerate(elements):
try:
# 提取元素属性
tag = el.evaluate('el => el.tagName')
text = ""
try:
text = el.inner_text().strip()[:100] # 限制长度
except Exception:
pass
# 提取关键属性
el_id = el.get_attribute('id') or ''
el_class = el.get_attribute('class') or ''
el_type = el.get_attribute('type') or ''
placeholder = el.get_attribute('placeholder') or ''
aria_label = el.get_attribute('aria-label') or ''
name = el.get_attribute('name') or ''
data_testid = el.get_attribute('data-testid') or el.get_attribute('data-test-id') or ''
# 获取元素位置
position = {}
try:
box = el.bounding_box()
if box:
position = {
'x': int(box.get('x', 0)),
'y': int(box.get('y', 0)),
'width': int(box.get('width', 0)),
'height': int(box.get('height', 0))
}
except Exception:
pass
# 构建选择器(优先级:ID > data-testid > class > text)
selector = ""
if el_id:
selector = f'#{el_id}'
elif data_testid:
selector = f'[data-testid="{data_testid}"]'
elif el_class and len(el_class.split()) > 0:
selector = f'.{el_class.split()[0]}'
elif text:
selector = f'{tag.lower()}:has-text("{text[:30]}")'
else:
selector = tag.lower()
# 构建候选元素信息
candidate = {
'index': idx,
'tag': tag,
'text': text,
'selector': selector,
'attributes': {
'id': el_id,
'class': el_class,
'type': el_type,
'placeholder': placeholder,
'aria-label': aria_label,
'name': name,
'data-testid': data_testid
},
'position': position
}
candidates.append(candidate)
except Exception as e:
logger.debug(f"获取元素详情失败(跳过): {e}")
continue
return best['element'], best['selectors']
return candidates
def find_element_by_semantic(
......
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:test_claude_enhanced_locate.py
模块描述:Claude 语义增强智能定位集成测试脚本
作者:czj
创建日期:2026-08-06
最后修改:2026-08-06
"""
import sys
import os
import json
import logging
# 添加项目路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# 设置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def test_claude_service_directly():
"""测试 Claude 服务直接调用"""
from app.services.claude_service import ClaudeService
print("=" * 60)
print("测试1: Claude 服务直接调用")
print("=" * 60)
test_candidates = [
{
'index': 0,
'tag': 'DIV',
'text': '会议预约',
'selector': 'div:has-text("会议预约")',
'attributes': {},
'position': {}
},
{
'index': 1,
'tag': 'BUTTON',
'text': '新建会议',
'selector': 'button:has-text("新建会议")',
'attributes': {'type': 'button'},
'position': {}
},
{
'index': 2,
'tag': 'INPUT',
'text': '',
'selector': 'input[placeholder]',
'attributes': {'type': 'text', 'placeholder': '请输入会议名称'},
'position': {}
}
]
service = ClaudeService()
# 测试1: 点击按钮
print("\n--- 场景1: 点击新建会议按钮 ---")
try:
idx, conf, reason = service.rank_candidates(
step_name='点击新建会议按钮',
action='click',
params={},
candidates=test_candidates
)
expected = 1 # 应该选择 BUTTON
status = "[OK]" if idx == expected else "[FAIL]"
print(f"{status} 结果: index={idx} (期望{expected}), confidence={conf:.2f}")
print(f" 理由: {reason}")
print(f" 选择器: {test_candidates[idx]['selector']}")
except Exception as e:
print(f"[FAIL] 失败: {e}")
# 测试2: 输入会议名称
print("\n--- 场景2: 会议名称输入:自动化测试 ---")
try:
idx, conf, reason = service.rank_candidates(
step_name='会议名称输入:自动化测试',
action='fill',
params={'value': '自动化测试'},
candidates=test_candidates
)
expected = 2 # 应该选择 INPUT
status = "[OK]" if idx == expected else "[FAIL]"
print(f"{status} 结果: index={idx} (期望{expected}), confidence={conf:.2f}")
print(f" 理由: {reason}")
print(f" 选择器: {test_candidates[idx]['selector']}")
except Exception as e:
print(f"[FAIL] 失败: {e}")
# 测试3: 点击菜单
print("\n--- 场景3: 点击会议预约分类 ---")
try:
idx, conf, reason = service.rank_candidates(
step_name='点击会议预约分类',
action='click',
params={},
candidates=test_candidates
)
expected = 0 # 应该选择 DIV "会议预约"
status = "[OK]" if idx == expected else "[FAIL]"
print(f"{status} 结果: index={idx} (期望{expected}), confidence={conf:.2f}")
print(f" 理由: {reason}")
print(f" 选择器: {test_candidates[idx]['selector']}")
except Exception as e:
print(f"[FAIL] 失败: {e}")
def test_claude_service_fallback():
"""测试 Claude 服务回退机制"""
from app.services.claude_service import ClaudeService
print("\n" + "=" * 60)
print("测试2: Claude 服务回退机制")
print("=" * 60)
service = ClaudeService()
service.enabled = False # 禁用 Claude
test_candidates = [
{'index': 0, 'tag': 'DIV', 'text': '会议预约', 'selector': 'div', 'attributes': {}, 'position': {}},
{'index': 1, 'tag': 'BUTTON', 'text': '新建会议', 'selector': 'button', 'attributes': {}, 'position': {}}
]
# 多候选回退
print("\n--- 场景1: 多候选回退(Claude 禁用)---")
try:
idx, conf, reason = service.rank_candidates(
step_name='点击新建会议按钮',
action='click',
params={},
candidates=test_candidates
)
print(f"[OK] 回退成功: index={idx}, confidence={conf:.2f}, reason={reason}")
except Exception as e:
print(f"[FAIL] 失败: {e}")
# 单候选
print("\n--- 场景2: 单候选(不调用 Claude)---")
try:
idx, conf, reason = service.rank_candidates(
step_name='点击按钮',
action='click',
params={},
candidates=[test_candidates[1]]
)
print(f"✅ 单候选: index={idx}, confidence={conf:.2f}, reason={reason}")
except Exception as e:
print(f"[FAIL] 失败: {e}")
def test_keyword_matcher_candidates():
"""测试关键词匹配返回候选列表"""
from app.services.keyword_matcher import extract_keywords, get_candidate_details
print("\n" + "=" * 60)
print("测试3: 关键词匹配候选列表")
print("=" * 60)
# 测试关键词提取
test_cases = [
("点击新建会议按钮", "click"),
("会议名称输入:自动化测试", "fill"),
("会议室选择:北京展厅会议室", "click"),
("等待页面加载", "wait"),
]
for desc, action in test_cases:
keywords = extract_keywords(desc)
print(f"\n步骤: {desc}")
print(f" 关键词: {keywords}")
print(f" 动作: {action}")
# 测试 get_candidate_details
print("\n--- get_candidate_details 空列表测试 ---")
details = get_candidate_details(None, [])
print(f"空列表结果: {details}")
def test_smart_locate_api():
"""测试智能定位 API"""
import requests
print("\n" + "=" * 60)
print("测试4: 智能定位 API(需要后端服务运行)")
print("=" * 60)
base_url = "http://localhost:8001"
# 健康检查
try:
resp = requests.get(f"{base_url}/api/element/health", timeout=5)
if resp.status_code == 200:
print(f"✅ 健康检查通过")
else:
print(f"❌ 健康检查失败: {resp.status_code}")
return
except Exception as e:
print(f"⚠️ 后端服务未启动,跳过 API 测试: {e}")
return
# 测试智能定位(不启用 Claude,快速测试)
print("\n--- 测试智能定位(use_claude=false)---")
try:
payload = {
"steps": [
{"order": 1, "name": "等待页面加载", "action": "wait", "params": {}},
],
"auto_login": False,
"navigate_menu": "",
"use_claude": False
}
resp = requests.post(f"{base_url}/api/element/smart-locate", json=payload, timeout=120)
result = resp.json()
print(f"状态码: {resp.status_code}")
print(f"结果: 成功 {result.get('located_steps', 0)}/{result.get('total_steps', 0)} 步骤")
except Exception as e:
print(f"❌ API 测试失败: {e}")
def main():
"""主测试入口"""
print("╔══════════════════════════════════════════════════════════╗")
print("║ Claude 语义增强智能定位 - 集成测试 ║")
print("╚══════════════════════════════════════════════════════════╝")
print()
# 测试1: Claude 服务直接调用
test_claude_service_directly()
# 测试2: 回退机制
test_claude_service_fallback()
# 测试3: 关键词匹配候选列表
test_keyword_matcher_candidates()
# 测试4: 智能定位 API
test_smart_locate_api()
print("\n" + "=" * 60)
print("测试完成!")
print("=" * 60)
if __name__ == "__main__":
main()
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论