ys/GenericAgent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: working memory dict, merge chat paths, simplify SKILL docs

Browse Source
This commit is contained in:
Jiaqing Liang
2026-03-16 13:18:41 +08:00
parent 3e6be97ec9
commit 065a5ee4bf
5 changed files with 80 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
agentmain.py
View File

@@ -38,8 +38,7 @@ def get_system_prompt():
class GeneraticAgent: class GeneraticAgent:
def __init__(self): def __init__(self):
script_dir = os.path.dirname(os.path.abspath(__file__)) script_dir = os.path.dirname(os.path.abspath(__file__))
temp_dir = os.path.join(script_dir, 'temp') os.makedirs(os.path.join(script_dir, 'temp'), exist_ok=True)
if not os.path.exists(temp_dir): os.makedirs(temp_dir)
from llmcore import mykeys from llmcore import mykeys
llm_sessions = [] llm_sessions = []
for k, cfg in mykeys.items(): for k, cfg in mykeys.items():
@@ -99,10 +98,11 @@ class GeneraticAgent:
sys_prompt = get_system_prompt() sys_prompt = get_system_prompt()
script_dir = os.path.dirname(os.path.abspath(__file__)) script_dir = os.path.dirname(os.path.abspath(__file__))
handler = GenericAgentHandler(None, self.history, os.path.join(script_dir, 'temp')) handler = GenericAgentHandler(None, self.history, os.path.join(script_dir, 'temp'))
if self.handler and self.handler.key_info: if self.handler and 'key_info' in self.handler.working:
handler.key_info = self.handler.key_info ki = re.sub(r'\n\[SYSTEM\] 此为.*?工作记忆[。\n]*', '', self.handler.working['key_info']) # 去旧
if '清除工作记忆' not in handler.key_info: handler.working['key_info'] = ki
handler.key_info += '\n[SYSTEM] 若开始新任务，先更新或清除工作记忆\n' handler.working['passed_sessions'] = ps = self.handler.working.get('passed_sessions', 0) + 1
if ps > 0: handler.working['key_info'] += f'\n[SYSTEM] 此为 {ps} 个对话前设置的工作记忆。若已在新任务，先更新或清除工作记忆\n'
self.handler = handler self.handler = handler
self.llmclient.backend = self.llmclient.backends[self.llm_no] self.llmclient.backend = self.llmclient.backends[self.llm_no]
user_input = raw_query user_input = raw_query
@@ -154,7 +154,6 @@ if __name__ == '__main__':
threading.Thread(target=agent.run, daemon=True).start() threading.Thread(target=agent.run, daemon=True).start()
if args.task: if args.task:
script_dir = os.path.dirname(os.path.abspath(__file__))
d = os.path.join(script_dir, f'temp/{args.task}'); rp = os.path.join(d, 'reply.txt'); nround = '' d = os.path.join(script_dir, f'temp/{args.task}'); rp = os.path.join(d, 'reply.txt'); nround = ''
with open(os.path.join(d, 'input.txt'), encoding='utf-8') as f: raw = f.read() with open(os.path.join(d, 'input.txt'), encoding='utf-8') as f: raw = f.read()
while True: while True:
@@ -193,8 +192,9 @@ if __name__ == '__main__':
except Exception as e: except Exception as e:
if once: raise if once: raise
print(f'[Reflect] drain error: {e}'); result = f'[ERROR] {e}' print(f'[Reflect] drain error: {e}'); result = f'[ERROR] {e}'
script_dir = os.path.dirname(os.path.abspath(__file__)) log_dir = os.path.join(script_dir, 'temp/reflect_logs'); os.makedirs(log_dir, exist_ok=True)
open(os.path.join(script_dir, './temp/reflect.log'), 'a', encoding='utf-8').write(f'[{datetime.now():%m-%d %H:%M}]\n{result}\n\n') script_name = os.path.splitext(os.path.basename(args.reflect))[0]
open(os.path.join(log_dir, f'{script_name}_{datetime.now():%Y-%m-%d}.log'), 'a', encoding='utf-8').write(f'[{datetime.now():%m-%d %H:%M}]\n{result}\n\n')
if on_done: if on_done:
try: on_done(result) try: on_done(result)
except Exception as e: print(f'[Reflect] on_done error: {e}') except Exception as e: print(f'[Reflect] on_done error: {e}')

18
ga.py
View File

@@ -246,8 +246,7 @@ class GenericAgentHandler(BaseHandler):
'''Generic Agent 工具库，包含多种工具的实现。工具函数自动加上了 do_ 前缀。实际工具名没有前缀。''' '''Generic Agent 工具库，包含多种工具的实现。工具函数自动加上了 do_ 前缀。实际工具名没有前缀。'''
def __init__(self, parent, last_history=None, cwd='./'): def __init__(self, parent, last_history=None, cwd='./'):
self.parent = parent self.parent = parent
self.key_info = "" self.working = {}
self.related_sop = ""
self.cwd = cwd; self.current_turn = 0 self.cwd = cwd; self.current_turn = 0
self.history_info = last_history if last_history else [] self.history_info = last_history if last_history else []
self.code_stop_signal = [] self.code_stop_signal = []
@@ -408,11 +407,12 @@ class GenericAgentHandler(BaseHandler):
''' '''
key_info = args.get("key_info", "") key_info = args.get("key_info", "")
related_sop = args.get("related_sop", "") related_sop = args.get("related_sop", "")
if "key_info" in args: self.key_info = key_info if "key_info" in args: self.working['key_info'] = key_info
if "related_sop" in args: self.related_sop = related_sop if "related_sop" in args: self.working['related_sop'] = related_sop
self.working['passed_sessions'] = 0
yield f"[Info] Updated key_info and related_sop.\n" yield f"[Info] Updated key_info and related_sop.\n"
yield f"key_info:\n{self.key_info}\n\n" yield f"key_info:\n{self.working.get('key_info', '')}\n\n"
yield f"related_sop:\n{self.related_sop}\n\n" yield f"related_sop:\n{self.working.get('related_sop', '')}\n\n"
next_prompt = self._get_anchor_prompt() next_prompt = self._get_anchor_prompt()
#next_prompt += '\n[SYSTEM TIPS] 此函数一般在任务开始或中间时调用，如果任务已成功完成应该是start_long_term_update用于结算长期记忆。\n' #next_prompt += '\n[SYSTEM TIPS] 此函数一般在任务开始或中间时调用，如果任务已成功完成应该是start_long_term_update用于结算长期记忆。\n'
return StepOutcome({"status": "success"}, next_prompt=next_prompt) return StepOutcome({"status": "success"}, next_prompt=next_prompt)
@@ -477,14 +477,14 @@ class GenericAgentHandler(BaseHandler):
h_str = "\n".join(self.history_info[-20:]) h_str = "\n".join(self.history_info[-20:])
prompt = f"\n### [WORKING MEMORY]\n<history>\n{h_str}\n</history>" prompt = f"\n### [WORKING MEMORY]\n<history>\n{h_str}\n</history>"
prompt += f"\nCurrent turn: {self.current_turn}\n" prompt += f"\nCurrent turn: {self.current_turn}\n"
if self.key_info: prompt += f"\n<key_info>{self.key_info}</key_info>" if self.working.get('key_info'): prompt += f"\n<key_info>{self.working.get('key_info')}</key_info>"
if self.related_sop: prompt += f"\n有不清晰的地方请再次读取{self.related_sop}" if self.working.get('related_sop'): prompt += f"\n有不清晰的地方请再次读取{self.working.get('related_sop')}"
try: print(prompt) try: print(prompt)
except: pass except: pass
return prompt return prompt
def next_prompt_patcher(self, next_prompt, outcome, turn): def next_prompt_patcher(self, next_prompt, outcome, turn):
if turn % 35 == 0 and 'plan' not in str(self.related_sop): if turn % 35 == 0 and 'plan' not in str(self.working.get('related_sop')):
next_prompt += f"\n\n[DANGER] 已连续执行第 {turn} 轮。你必须总结情况进行ask_user，不允许继续重试。" next_prompt += f"\n\n[DANGER] 已连续执行第 {turn} 轮。你必须总结情况进行ask_user，不允许继续重试。"
elif turn % 7 == 0: elif turn % 7 == 0:
next_prompt += f"\n\n[DANGER] 已连续执行第 {turn} 轮。禁止无效重试。若无有效进展，必须切换策略：1. 探测物理边界 2. 请求用户协助。如有需要，可调用 update_working_checkpoint 保存关键上下文。" next_prompt += f"\n\n[DANGER] 已连续执行第 {turn} 轮。禁止无效重试。若无有效进展，必须切换策略：1. 探测物理边界 2. 请求用户协助。如有需要，可调用 update_working_checkpoint 保存关键上下文。"

37
llmcore.py
View File

@@ -413,23 +413,28 @@ class ToolClient:
self.total_cd_tokens = 0 self.total_cd_tokens = 0
def chat(self, messages, tools=None): def chat(self, messages, tools=None):
script_dir = os.path.dirname(os.path.abspath(__file__))
log_path = os.path.join(script_dir, f'./temp/model_responses_{os.getpid()}.txt')
if self._should_use_structured_messages(messages): if self._should_use_structured_messages(messages):
return (yield from self._chat_structured(messages, tools)) backend_messages = self._build_backend_messages(messages, tools)
print("Structured prompt length:", sum(self._estimate_content_len(m.get("content")) for m in backend_messages), 'chars')
prompt_log = self._serialize_messages_for_log(backend_messages)
gen = self.backend.raw_ask(backend_messages)
else:
full_prompt = self._build_protocol_prompt(messages, tools) full_prompt = self._build_protocol_prompt(messages, tools)
print("Full prompt length:", len(full_prompt), 'chars') print("Full prompt length:", len(full_prompt), 'chars')
script_dir = os.path.dirname(os.path.abspath(__file__)) prompt_log = full_prompt
with open(os.path.join(script_dir, f'./temp/model_responses_{os.getpid()}.txt'), 'a', encoding='utf-8', errors="replace") as f:
f.write(f"=== Prompt === {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n{full_prompt}\n")
gen = self.backend.ask(full_prompt, stream=True) gen = self.backend.ask(full_prompt, stream=True)
with open(log_path, 'a', encoding='utf-8', errors="replace") as f:
f.write(f"=== Prompt === {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n{prompt_log}\n")
raw_text = ''; summarytag = '[NextWillSummary]' raw_text = ''; summarytag = '[NextWillSummary]'
for chunk in gen: for chunk in gen:
raw_text += chunk; raw_text += chunk
if chunk != summarytag: yield chunk if chunk != summarytag: yield chunk
print('Complete response received.') print('Complete response received.')
if raw_text.endswith(summarytag): if raw_text.endswith(summarytag):
self.last_tools = ''; raw_text = raw_text[:-len(summarytag)] self.last_tools = ''; raw_text = raw_text[:-len(summarytag)]
script_dir = os.path.dirname(os.path.abspath(__file__)) with open(log_path, 'a', encoding='utf-8', errors="replace") as f:
with open(os.path.join(script_dir, f'./temp/model_responses_{os.getpid()}.txt'), 'a', encoding='utf-8', errors="replace") as f:
f.write(f"=== Response === {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n{raw_text}\n\n") f.write(f"=== Response === {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n{raw_text}\n\n")
return self._parse_mixed_response(raw_text) return self._parse_mixed_response(raw_text)
@@ -506,24 +511,6 @@ class ToolClient:
logged.append(msg) logged.append(msg)
return json.dumps(logged, ensure_ascii=False, indent=2) return json.dumps(logged, ensure_ascii=False, indent=2)
def _chat_structured(self, messages, tools):
backend_messages = self._build_backend_messages(messages, tools)
print("Structured prompt length:", sum(self._estimate_content_len(m.get("content")) for m in backend_messages), 'chars')
script_dir = os.path.dirname(os.path.abspath(__file__))
with open(os.path.join(script_dir, f'./temp/model_responses_{os.getpid()}.txt'), 'a', encoding='utf-8', errors="replace") as f:
f.write(f"=== Prompt === {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n{self._serialize_messages_for_log(backend_messages)}\n")
gen = self.backend.raw_ask(backend_messages)
raw_text = ''; summarytag = '[NextWillSummary]'
for chunk in gen:
raw_text += chunk
if chunk != summarytag: yield chunk
print('Complete response received.')
if raw_text.endswith(summarytag):
self.last_tools = ''; raw_text = raw_text[:-len(summarytag)]
with open(os.path.join(script_dir, f'./temp/model_responses_{os.getpid()}.txt'), 'a', encoding='utf-8', errors="replace") as f:
f.write(f"=== Response === {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n{raw_text}\n\n")
return self._parse_mixed_response(raw_text)
def _build_protocol_prompt(self, messages, tools): def _build_protocol_prompt(self, messages, tools):
system_content = next((m['content'] for m in messages if m['role'].lower() == 'system'), "") system_content = next((m['content'] for m in messages if m['role'].lower() == 'system'), "")
history_msgs = [m for m in messages if m['role'].lower() != 'system'] history_msgs = [m for m in messages if m['role'].lower() != 'system']

2
memory/autonomous_operation_sop.md
View File

@@ -15,7 +15,7 @@
## 执行 ## 执行
**启动**： **启动**：
- update_working_checkpoint: `自主行动｜报告→./autonomous_reports/R{XX}_简短描述.md｜≤30回合｜收尾：重读sop，写报告+更新history+标记TODO` - update_working_checkpoint: `自主行动｜报告→./autonomous_reports/R{XX}_简短描述.md｜≤30回合｜收尾：重读sop，写报告+更新history+标记TODO | 报告严禁放错位置`
- 读 `./autonomous_reports/history.txt` 推断下一编号RXX + 了解历史避免重复 - 读 `./autonomous_reports/history.txt` 推断下一编号RXX + 了解历史避免重复
**执行**： **执行**：

144
memory/skill_search/SKILL.md
View File

@@ -1,116 +1,64 @@
# Skill Search — API 客户端 # Skill Search — 105K 技能卡检索
> 从 10 万+ 技能卡中智能检索最适合当前环境的 skill。 > 从 105K+ 技能卡中语义搜索最匹配的 skill。零依赖，内置默认 API 地址，开箱即用。
## 架构 ## 最简调用
```
┌─────────────┐ HTTPS/JSON ┌──────────────────┐
│ 客户端 CLI │ ──────────────────▶ │ Skill Search │
│ (本项目) │ ◀────────────────── │ API Server │
└─────────────┘ └──────────────────┘
• 环境检测 • 105K+ 技能卡索引
• 结果格式化 • 四层漏斗检索引擎
• 零数据依赖 • 环境过滤 → 安全标注
→ 语义匹配 → 质量排序
```
## 快速开始
```bash
# 设置 API 地址
export SKILL_SEARCH_API="https://your-server.com/api"
export SKILL_SEARCH_KEY="your-api-key" # 可选
# 搜索
python -m skill_search "python testing"
# 限定类别
python -m skill_search "docker deploy" --category devops
# JSON 输出（适合程序集成）
python -m skill_search "git workflow" --json
# 查看环境信息
python -m skill_search --env
# 查看索引统计
python -m skill_search --stats
```
## 编程接口
```python ```python
from skill_search import search, detect_environment import sys; sys.path.append('../memory/skill_search')
from skill_search import search
env = detect_environment()
results = search(query="python testing", env=env, top_k=5)
results = search("python send email") # ⚠️ 必须用英文查询，中文匹配效果极差
for r in results: for r in results:
print(f"{r.skill.name} (score: {r.final_score:.2f})") s = r.skill
print(f" {r.skill.one_line_summary}") print(f"[{r.final_score:.2f}] {s.name} — {s.one_line_summary}")
print(f" key: {s.key} category: {s.category} tags: {s.tags[:3]}")
``` ```
## 文件结构 ## API 签名
``` ```python
skill_search/ search(query, env=None, category=None, top_k=10) -> list[SearchResult]
├── SKILL.md # 本文档 # env: 自动检测，一般不传
└── skill_search/ # Python 包 # category: 可选过滤，如 "devops"
├── __init__.py # 公开 API # top_k: 返回数量，默认10
├── __main__.py # CLI 入口
├── engine.py # HTTP 客户端（替代本地检索）
├── index.py # SkillIndex 数据模型
├── env_detect.py # 本地环境检测
└── formatter.py # 结果格式化输出
``` ```
## 环境变量 ## 返回结构
| 变量 | 说明 | 默认值 | ```
|------|------|--------| SearchResult
| `SKILL_SEARCH_API` | API 服务地址 | `https://skill-search.example.com/api` | .final_score float 综合评分 (0~1)
| `SKILL_SEARCH_KEY` | API 密钥（可选） | 无 | .relevance float 语义相关度
.quality float 质量分
.match_reasons list[str] 匹配原因
.warnings list[str] 警告
.skill SkillIndex ↓
## API 协议 SkillIndex (常用字段)
.key str 唯一标识/路径
### POST /search .name str 名称
.one_line_summary str 一句话摘要
请求: .description str 详细描述
```json .category str 类别
{ .tags list[str] 标签
"query": "python testing", .form str 形式(sop/script/...)
"env": { "os": "windows", "shell": "powershell", ... }, .autonomous_safe bool 是否自主安全
"category": "coding",
"top_k": 10
}
``` ```
响应: ## CLI
```json
{ ```bash
"results": [ python -m skill_search "python testing"
{ python -m skill_search "docker deployment" --category devops --top 5
"skill": { "key": "org/repo/skill", "name": "...", ... }, python -m skill_search "git" --json
"relevance": 0.85, python -m skill_search --stats
"quality": 7.2, python -m skill_search --env
"final_score": 0.78,
"match_reasons": ["完整短语匹配", "标签匹配: python"],
"warnings": []
}
]
}
``` ```
### POST /stats ## 配置
请求: `{}` 或 `{"env": {...}}` | 项 | 默认值 | 说明 |
|---|---|---|
响应: | API地址 | `http://www.fudankw.cn:58787` | 环境变量 `SKILL_SEARCH_API` 可覆盖 |
```json | API密钥 | 无(可选) | 环境变量 `SKILL_SEARCH_KEY` |
{
"total": 105586,
"safe_count": 98234,
"categories": { "coding": 45000, "devops": 12000, ... }
}
```