Refactor: restructure directory and clean up redundant scripts

This commit is contained in:
Liang Jiaqing
2026-02-01 08:58:26 +08:00
parent 669095699e
commit 35a1b70208
9 changed files with 109 additions and 85 deletions

View File

@@ -7,7 +7,7 @@ sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
import streamlit as st import streamlit as st
import time, json, re import time, json, re
with open('tools_schema.json', 'r', encoding='utf-8') as f: with open('assets/tools_schema.json', 'r', encoding='utf-8') as f:
TOOLS_SCHEMA = json.load(f) TOOLS_SCHEMA = json.load(f)
st.set_page_config(page_title="Cowork", layout="wide") st.set_page_config(page_title="Cowork", layout="wide")
@@ -34,8 +34,11 @@ def get_system_prompt():
if not os.path.exists('memory/global_mem.txt'): if not os.path.exists('memory/global_mem.txt'):
with open('memory/global_mem.txt', 'w', encoding='utf-8') as f: f.write('') with open('memory/global_mem.txt', 'w', encoding='utf-8') as f: f.write('')
if not os.path.exists('memory/global_mem_insight.txt'): if not os.path.exists('memory/global_mem_insight.txt'):
with open('memory/global_mem_insight.txt', 'w', encoding='utf-8') as f: f.write('') content = "## Global Memory Index (Logic)\n\n[CONSTITUTION]\n1. 改我自身源码前必须先问用户\n\n[STORES]\n- global_mem: ../memory/global_mem.txt\n\n[ACCESS]\n- global_mem: 按 TOPIC 检索索引\n\n[TOPICS.GLOBAL_MEM]"
with open('sys_prompt.txt', 'r', encoding='utf-8') as f: prompt = f.read() if os.path.exists('assets/global_mem_insight_template.txt'):
with open('assets/global_mem_insight_template.txt', 'r', encoding='utf-8') as f: content = f.read()
with open('memory/global_mem_insight.txt', 'w', encoding='utf-8') as f: f.write(content)
with open('assets/sys_prompt.txt', 'r', encoding='utf-8') as f: prompt = f.read()
prompt += get_global_memory() prompt += get_global_memory()
return prompt return prompt

View File

@@ -0,0 +1,28 @@
## Global Memory Index (Logic)
[CONSTITUTION]
1. 改我自身源码前必须先问用户;在 ./ 内自由实验可直接做。
2. 要下结论/依赖环境/做操作前:先枚举可用存储(store)与索引;再读取相关条目;未验证不硬断言。
3. 没有用户授权不要直接文本读取或移动密钥/密码文件,能引用即可。或者先请求用户许可。
[STORES]
- global_mem: ../memory/global_mem.txt (Facts, Policy, Prefs)
- memory_dir: ../memory/ (SOPs, Tools)
[ACCESS]
- global_mem: 按 TOPIC 检索索引 → file_read 对应条目。
- memory_dir: ls ../memory/
[TOPICS.GLOBAL_MEM]
# 【引导注释 - 使用后请删除】
# 请模仿以下概括性 Tag 格式记录 global_mem.txt 中的事实:
# - env/network: 网络代理或环境描述 (Global Facts ## [SECTION])
# - auth/platform: 平台账号或认证信息 (Global Facts ## [SECTION])
# - pref/download_dir: 文件存储偏好路径 (Global Facts ## [SECTION])
# - tool/config: 核心工具运行参数 (Global Facts ## [SECTION])
[TOPICS.MEMORY_DIR]
# 【引导注释 - 使用后请删除】
# 请模仿以下格式登记 memory 目录下的独立文件:
# - sop/workflow: 标准操作流程文档 (.md)
# - tool/automation: 自动化执行脚本 (.py)

View File

@@ -15,7 +15,7 @@ def generate_tool_schema():
""" """
通过代码内省 Handler 的逻辑映射为高语义的工具描述 通过代码内省 Handler 的逻辑映射为高语义的工具描述
""" """
with open('ga.py', 'r', encoding='utf-8') as f: with open('../ga.py', 'r', encoding='utf-8') as f:
ga_code = f.read() ga_code = f.read()
# 极简且具备高度概括能力的元 Prompt # 极简且具备高度概括能力的元 Prompt
meta_prompt = f""" meta_prompt = f"""
@@ -79,7 +79,7 @@ def generate_tool_schema():
return None return None
def make_system_prompt(ga_code_path='ga.py'): def make_system_prompt(ga_code_path='../ga.py'):
with open(ga_code_path, 'r', encoding='utf-8') as f: with open(ga_code_path, 'r', encoding='utf-8') as f:
ga_code = f.read() ga_code = f.read()

View File

@@ -1,7 +1,7 @@
[ [
{"type": "function", "function": { {"type": "function", "function": {
"name": "code_run", "name": "code_run",
"description": "针对 Windows 优化的双模态代码执行器。优先使用python仅在必要系统操作时使用 powershell。注意执行的代码必须以 ```python 或 ```powershell 代码块的形式包含在回复正文中。严禁在代码中硬编码大量数据,如有需要应通过文件读取。执行时间限制为 60s。", "description": "针对 Windows 优化的双模态代码执行器。优先使用python仅在必要系统操作时使用 powershell。注意执行的代码必须放在在回复正文中,以 ```python 或 ```powershell 代码块的形式。严禁在代码中硬编码大量数据,如有需要应通过文件读取。",
"parameters": {"type": "object", "properties": { "parameters": {"type": "object", "properties": {
"type": {"type": "string", "enum": ["python", "powershell"], "description": "执行环境类型,默认为 python。", "default": "python"}, "type": {"type": "string", "enum": ["python", "powershell"], "description": "执行环境类型,默认为 python。", "default": "python"},
"timeout": {"type": "integer", "description": "执行超时时间(秒),默认 60。", "default": 60}, "timeout": {"type": "integer", "description": "执行超时时间(秒),默认 60。", "default": 60},

3
ga.py
View File

@@ -97,7 +97,7 @@ def ask_user(question: str, candidates: list = None):
} }
} }
from web_tools import execute_js_rich, get_html from simphtml import execute_js_rich, get_html
driver = None driver = None
@@ -109,6 +109,7 @@ def first_init_driver():
time.sleep(1) time.sleep(1)
sess = driver.get_all_sessions() sess = driver.get_all_sessions()
if len(sess) > 0: break if len(sess) > 0: break
if len(sess) == 1:
driver.newtab() driver.newtab()
time.sleep(5) time.sleep(5)

View File

@@ -863,3 +863,70 @@ def find_changed_elements(before_html, after_html):
"similarity": similarity "similarity": similarity
#"changed_elements": changed_elements[:3] #"changed_elements": changed_elements[:3]
} }
def get_html(driver, cutlist=False, maxchars=28000, instruction=""):
page = get_main_block(driver)
soup = optimize_html_for_tokens(page)
html = str(soup)
if not cutlist or len(html) <= maxchars: return html
rr = driver.execute_js(js_findMainList + js_findMainContent + """
return findMainList(findMainContent(document.body));""")
sel = rr.get("selector", None)
if not sel: return html[:maxchars]
s = BeautifulSoup(str(soup), "html.parser"); items = s.select(sel)
hit = [it for it in items if instruction and instruction.strip() and instruction in it.get_text(" ",strip=True)]
keep = hit[:6] if hit else items[:3]
for it in items:
if it not in keep: it.decompose()
s = optimize_html_for_tokens(s)
return str(s)[:maxchars]
def execute_js_rich(script, driver):
start_temp_monitor(driver)
curr_session = driver.default_session_id
last_html = get_html(driver)
result = None; error_msg = None
new_tab = False; reloaded = False
try:
print(f"⚡ Executing: {script[:250]} ...")
result = driver.execute_js(script, auto_switch_newtab=True)
if type(result) is dict and result.get('closed', 0) == 1: reloaded = True
time.sleep(2)
except Exception as e:
error = e.args[0] if e.args else str(e)
if isinstance(error, dict): error.pop('stack', None)
error_msg = str(error)
print(f"❌ Error: {error_msg}")
transients = get_temp_texts(driver)
if driver.default_session_id != curr_session:
curr_session = driver.latest_session_id
print('Session changed')
new_tab = True
current_html = get_html(driver)
diff_summary = "无需对比 (报错)"
is_significant_change = False
if not error_msg:
diff_data = find_changed_elements(last_html, current_html)
change_count = diff_data.get('changed', 0)
diff_summary = f"DOM变化量: {change_count}"
if change_count < 5 and not transients and not new_tab:
diff_summary += " (页面几乎无静默变化)"
else:
is_significant_change = True
return {
"status": "failed" if error_msg else "success",
"js_return": result,
"error": error_msg,
"transients": transients,
"environment": {
"new_tab": new_tab,
"reloaded": reloaded
},
"diff": diff_summary,
"suggestion": "" if is_significant_change else "页面无明显变化"
}

View File

@@ -1,75 +0,0 @@
import sys, os, re
import pyperclip
import json, time
import subprocess
import tempfile
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from simphtml import get_main_block, start_temp_monitor, get_temp_texts, find_changed_elements, optimize_html_for_tokens
from simphtml import js_findMainContent, js_findMainList
from bs4 import BeautifulSoup
def get_html(driver, cutlist=False, maxchars=28000, instruction=""):
page = get_main_block(driver)
soup = optimize_html_for_tokens(page)
html = str(soup)
if not cutlist or len(html) <= maxchars: return html
rr = driver.execute_js(js_findMainList + js_findMainContent + """
return findMainList(findMainContent(document.body));""")
sel = rr.get("selector", None)
if not sel: return html[:maxchars]
s = BeautifulSoup(str(soup), "html.parser"); items = s.select(sel)
hit = [it for it in items if instruction and instruction.strip() and instruction in it.get_text(" ",strip=True)]
keep = hit[:6] if hit else items[:3]
for it in items:
if it not in keep: it.decompose()
s = optimize_html_for_tokens(s)
return str(s)[:maxchars]
def execute_js_rich(script, driver):
start_temp_monitor(driver)
curr_session = driver.default_session_id
last_html = get_html(driver)
result = None; error_msg = None
new_tab = False; reloaded = False
try:
print(f"⚡ Executing: {script[:250]} ...")
result = driver.execute_js(script, auto_switch_newtab=True)
if type(result) is dict and result.get('closed', 0) == 1: reloaded = True
time.sleep(2)
except Exception as e:
error = e.args[0] if e.args else str(e)
if isinstance(error, dict): error.pop('stack', None)
error_msg = str(error)
print(f"❌ Error: {error_msg}")
transients = get_temp_texts(driver)
if driver.default_session_id != curr_session:
curr_session = driver.latest_session_id
print('Session changed')
new_tab = True
current_html = get_html(driver)
diff_summary = "无需对比 (报错)"
is_significant_change = False
if not error_msg:
diff_data = find_changed_elements(last_html, current_html)
change_count = diff_data.get('changed', 0)
diff_summary = f"DOM变化量: {change_count}"
if change_count < 5 and not transients and not new_tab:
diff_summary += " (页面几乎无静默变化)"
else:
is_significant_change = True
return {
"status": "failed" if error_msg else "success",
"js_return": result,
"error": error_msg,
"transients": transients,
"environment": {
"new_tab": new_tab,
"reloaded": reloaded
},
"diff": diff_summary,
"suggestion": "" if is_significant_change else "页面无明显变化"
}