diff --git a/llmcore.py b/llmcore.py index 8de9d8b6..33857cf3 100644 --- a/llmcore.py +++ b/llmcore.py @@ -618,30 +618,40 @@ class LLMSession(BaseSession): def make_messages(self, raw_list): return _msgs_claude2oai(_fix_messages(raw_list)) def _fix_messages(messages): - """修复 messages 符合 Claude API:交替、tool_use/tool_result 配对""" if not messages: return messages - _wrap = lambda c: c if isinstance(c, list) else [{"type": "text", "text": str(c)}] - fixed = [] + W = lambda c: c if isinstance(c, list) else [{"type": "text", "text": str(c)}] + merged = [] for m in messages: - if fixed and m['role'] == fixed[-1]['role']: - fixed[-1] = {**fixed[-1], 'content': _wrap(fixed[-1].get('content', [])) + [{"type": "text", "text": "\n"}] + _wrap(m.get('content', []))}; continue - if fixed and fixed[-1]['role'] == 'assistant' and m['role'] == 'user': - uses = [b.get('id') for b in _wrap(fixed[-1].get('content', [])) if isinstance(b, dict) and b.get('type') == 'tool_use' and b.get('id')] - if uses: - got, rest = {}, [] - for b in _wrap(m.get('content', [])): - if isinstance(b, dict) and b.get('type') == 'tool_result': - tid = b.get('tool_use_id') - if tid in uses and tid not in got: got[tid] = b - else: rest.append({"type":"text","text":str(b.get('content',''))}) - else: rest.append(b) - m = {**m, 'content': [got.get(uid) or {"type": "tool_result", "tool_use_id": uid, "content": "(error)"} for uid in uses] + rest} - else: m = {**m, 'content': [{"type":"text","text":str(b.get('content',''))} if isinstance(b,dict) and b.get('type')=='tool_result' else b for b in _wrap(m.get('content', []))]} - fixed.append(m) - while fixed and fixed[0]['role'] != 'user': fixed.pop(0) - if fixed and fixed[0]['role'] == 'user': - fixed[0] = {**fixed[0], 'content': [{"type":"text","text":str(b.get('content',''))} if isinstance(b,dict) and b.get('type')=='tool_result' else b for b in _wrap(fixed[0].get('content', []))]} - return fixed + if m.get('role') not in ('user', 'assistant'): continue + blocks = W(m.get('content', [])) + if merged and m['role'] == merged[-1]['role']: + merged[-1]['content'] += [{"type": "text", "text": "\n"}] + blocks + else: + merged.append({"role": m['role'], "content": list(blocks)}) + while merged and merged[0]['role'] != 'user': merged.pop(0) + if not merged: return [] + prev_uses = [] + for m in merged: + c = m['content'] + if m['role'] == 'assistant': + seen, out = set(), [] + for b in c: + uid = b.get('id') if isinstance(b, dict) and b.get('type') == 'tool_use' else None + if uid and uid in seen: continue + if uid: seen.add(uid) + out.append(b) + m['content'] = out + prev_uses = [b.get('id') for b in out if isinstance(b, dict) and b.get('type') == 'tool_use'] + else: + got, rest = {}, [] + for b in c: + tid = b.get('tool_use_id') if isinstance(b, dict) and b.get('type') == 'tool_result' else None + if tid and tid in prev_uses and tid not in got: got[tid] = b + elif isinstance(b, dict) and b.get('type') == 'tool_result': rest.append({"type": "text", "text": str(b.get('content', ''))}) + else: rest.append(b) + m['content'] = [got.get(u) or {"type": "tool_result", "tool_use_id": u, "content": "(error)"} for u in prev_uses] + rest + prev_uses = [] + return merged class NativeClaudeSession(BaseSession): def __init__(self, cfg): diff --git a/memory/ljqCtrl.py b/memory/ljqCtrl.py index e1e77d4d..5d9ef32f 100644 --- a/memory/ljqCtrl.py +++ b/memory/ljqCtrl.py @@ -2,7 +2,7 @@ CRITICAL: 严禁在此工具链中 import pyautogui (会污染 win32api 导致逻辑冲突)。 ljqCtrl Quick Reference: - dpi_scale: float (Logical = Physical * dpi_scale) -- Click(x, y): Use Physical Coordinates (from screenshots) +- Click(x, y, check=True): Use Physical Coordinates. check=True → 自动比前后像素变化,返回周边图像 - SetCursorPos(z): Use Physical Coordinates z=(x, y) - Press(cmd, staytime=0): Keyboard shortcuts (e.g. 'ctrl+v') - FindBlock(fn, wrect=None, threshold=0.8) -> (obj_center_phys, is_found) @@ -47,10 +47,19 @@ def SetCursorPos(z): win32api.SetCursorPos(z) time.sleep(0.05) -def Click(x, y=None): +def Click(x, y=None, check=True): if type(x) is type(tuple()): x, y = int(x[0]), int(x[1]) + if check: before = ScreenCapAt(x, y) SetCursorPos( (x, y) ) MouseClick() + if check: + time.sleep(0.3) + after = ScreenCapAt(x, y) + b, a = np.array(before), np.array(after) + diff = np.sum(np.any(b != a, axis=2)) + total = b.shape[0] * b.shape[1] + print(f'[Click check] near 100px rect {diff}/{total} pixels changed ({diff/total*100:.1f}%)') + return after click = Click def Press(cmd, staytime=0): @@ -127,6 +136,11 @@ def FindBlock(fn, wrect=None, verbose=0, threshold=0.8): sscr.show() return obj, max_val > threshold +def ScreenCapAt(x, y, r=100): + """物理坐标(x,y)为中心±r的屏幕截图 → PIL Image""" + from PIL import ImageGrab + return ImageGrab.grab((x-r, y-r, x+r, y+r)) + if __name__ == '__main__': #time.sleep(3) #SetCursorPos( (1640, 131) ) diff --git a/memory/ui_detect.py b/memory/ui_detect.py index d3f50ff0..8dede28f 100644 --- a/memory/ui_detect.py +++ b/memory/ui_detect.py @@ -1,11 +1,13 @@ #!/usr/bin/env python3 """ -UI元素检测 - 基于OmniParser YOLO + RapidOCR +UI元素检测 - YOLO + RapidOCR 用法: from ui_detect import detect - elements = detect("screenshot.png", mode='crop') # 或 'match' + elements = detect("screenshot.png") # 默认match模式 + elements = detect(pil_image) # 支持PIL.Image + elements = detect(img, mode='crop') # crop备选 返回: [{'bbox':[x1,y1,x2,y2], 'type':'icon'|'text', 'label':str|None, 'confidence':float}] -模式: crop=YOLO+逐块OCR(label全) | match=YOLO+全图OCR空间匹配(快,label=None可VLM保底) +模式: match=YOLO+全图OCR IoU匹配(推荐,1.2s,无文字图标label=None可VLM保底) | crop=拼接crop OCR(备选,更精确,2.3s) 依赖: ultralytics, rapidocr-onnxruntime, pillow, numpy """ from pathlib import Path @@ -43,15 +45,38 @@ def _ocr_full(image_path): out.append([int(min(xs)), int(min(ys)), int(max(xs)), int(max(ys)), text, conf]) return out -def _ocr_crop(img, bbox): - """裁剪区域OCR → text or None""" - if not _ocr: return None - x1, y1, x2, y2 = bbox - crop = img.crop((x1, y1, x2, y2)) - arr = np.array(crop) - result, _ = _ocr(arr) - if not result: return None - return ' '.join(t for _, t, _ in result) +def _ocr_crops_batch(img, yolo_boxes): + """批量OCR:将所有YOLO框crop垂直拼接为一张图,一次OCR,按y坐标映射回各box → {box_idx: text}""" + if not _ocr or not yolo_boxes: return {} + + crops, offsets = [], [] # offsets: [(y_off, orig_x1, orig_y1, box_idx)] + max_w, y_cursor = 0, 0 + for idx, (x1, y1, x2, y2, _) in enumerate(yolo_boxes): + crop = img.crop((x1, y1, x2, y2)) + w, h = crop.size + max_w = max(max_w, w) + crops.append(crop) + offsets.append((y_cursor, x1, y1, idx)) + y_cursor += h + if max_w == 0: return {} + # 垂直拼接 + stitched = Image.new('RGB', (max_w, y_cursor), (255, 255, 255)) + for i, crop in enumerate(crops): + stitched.paste(crop, (0, offsets[i][0])) + + result, _ = _ocr(np.array(stitched)) + if not result: return {} + # 映射:OCR框中心y → 归属的crop + labels = {} + for bbox, text, _ in result: + cy = sum(p[1] for p in bbox) / len(bbox) + for y_off, ox1, oy1, idx in offsets: + h = yolo_boxes[idx][3] - yolo_boxes[idx][1] + if y_off <= cy < y_off + h: + old = labels.get(idx) + labels[idx] = (old + ' ' + text) if old else text + break + return labels def _iou(a, b): """计算两个bbox的交集占b面积的比例(包含率)""" @@ -60,11 +85,11 @@ def _iou(a, b): area_b = (b[2]-b[0]) * (b[3]-b[1]) return inter / area_b if area_b > 0 else 0 -def detect(image_path, mode='crop', model_path=None, conf=0.25, iou_thresh=0.5): +def detect(image_path, mode='match', model_path=None, conf=0.25, iou_thresh=0.5): """ 统一检测入口,返回元素列表: [{'bbox':[x1,y1,x2,y2], 'type':'icon'|'text', 'label':str|None, 'confidence':float}] - mode: 'crop' = YOLO+逐块OCR | 'match' = YOLO+全图OCR空间匹配 + mode: 'match' = YOLO+全图OCR空间匹配(推荐, 快) | 'crop' = YOLO+拼接OCR(备选, 更精确) 支持 image_path: str 路径 或 PIL.Image 对象 """ # 归一化:PIL Image → 临时文件 @@ -79,10 +104,10 @@ def detect(image_path, mode='crop', model_path=None, conf=0.25, iou_thresh=0.5): elements = [] if mode == 'crop': - # YOLO元素逐块OCR - for x1, y1, x2, y2, c in yolo_boxes: - label = _ocr_crop(img, [x1, y1, x2, y2]) - elements.append({'bbox': [x1,y1,x2,y2], 'type': 'icon', 'label': label, 'confidence': c}) + # YOLO元素批量OCR(拼接一次推理) + labels_map = _ocr_crops_batch(img, yolo_boxes) + for idx, (x1, y1, x2, y2, c) in enumerate(yolo_boxes): + elements.append({'bbox': [x1,y1,x2,y2], 'type': 'icon', 'label': labels_map.get(idx), 'confidence': c}) # 补充:全图OCR找未被覆盖的纯文本 for ox1, oy1, ox2, oy2, text, oc in _ocr_full(image_path): covered = any(_iou([x1,y1,x2,y2,_,__], [ox1,oy1,ox2,oy2]) > iou_thresh