252 lines
14 KiB
Plaintext
252 lines
14 KiB
Plaintext
# cockpit_chat.dspy - SDLC Agent Loop v3.1 (streaming NDJSON)
|
||
import aiohttp, json, os
|
||
|
||
action = (params_kw or {}).get('action', 'list_messages')
|
||
dbname = get_module_dbname('pipeline-sdlc')
|
||
|
||
AGENT_PROMPT = """你是开发产线驾驶舱 Agent。需要数据时调工具，最终用 reply 回复。
|
||
|
||
输出格式二选一：
|
||
{"action":"tool_call","tool":"工具名","params":{}}
|
||
{"action":"reply","message":"中文回复"}
|
||
|
||
工具: __TOOLS__
|
||
环境: __ENV__"""
|
||
|
||
TOOLS = [
|
||
{"name":"switch_project","description":"切换项目","params":{"project":"名称"}},
|
||
{"name":"create_task","description":"提交任务","params":{"title":"标题","description":"描述","role":"requirement/design/develop/test/deploy"}},
|
||
{"name":"check_progress","description":"查进度","params":{"task_id":"任务ID(可选)"}},
|
||
{"name":"list_tasks","description":"列出任务","params":{"state":"状态(可选)"}},
|
||
{"name":"add_repo","description":"关联仓库","params":{"url":"git地址","name":"名称(可选)"}},
|
||
{"name":"list_repos","description":"列出仓库","params":{}},
|
||
{"name":"add_bug","description":"报告Bug","params":{"title":"标题","description":"描述"}},
|
||
]
|
||
TOOLS_TEXT = json.dumps(TOOLS, ensure_ascii=False)
|
||
|
||
def _security_scan(text):
|
||
t = (text or '').lower()
|
||
for kw in ('drop table','truncate','rm -rf','密钥','api_key'):
|
||
if kw in t: return True, '危险操作'
|
||
return False, ''
|
||
|
||
def _guess_role(title):
|
||
t = (title or '').lower()
|
||
for kw,r in [('需求','requirement'),('设计','design'),('开发','develop'),('实现','develop'),('测试','test'),('部署','deploy')]:
|
||
if kw in t: return r
|
||
return 'develop'
|
||
|
||
async def _find_project(sor, name):
|
||
if not name: return None
|
||
recs = await sor.sqlExe("SELECT id,name FROM sd_projects WHERE name=${n}$",{"n":name})
|
||
if recs: return recs[0]
|
||
allp = await sor.sqlExe("SELECT id,name FROM sd_projects ORDER BY created_at DESC",{})
|
||
kw = name.lower()
|
||
for r in (allp or []):
|
||
if kw in (getattr(r,'name','') or '').lower(): return r
|
||
return None
|
||
|
||
async def _load_ctx(sor, uid):
|
||
recs = await sor.sqlExe("SELECT current_project_id FROM pipeline_agent_settings WHERE user_id=${u}$",{"u":uid})
|
||
ctx = {'pid':'','pname':'','ws':'','skills_dir':'','repos':[]}
|
||
if not recs: return ctx
|
||
ctx['pid'] = getattr(recs[0],'current_project_id','') or ''
|
||
if ctx['pid']:
|
||
projs = await sor.sqlExe("SELECT name,workspace_dir,org_id FROM sd_projects WHERE id=${p}$",{"p":ctx['pid']})
|
||
if projs:
|
||
ctx['pname'] = getattr(projs[0],'name','')
|
||
ctx['ws'] = getattr(projs[0],'workspace_dir','') or ''
|
||
oid = getattr(projs[0],'org_id','') or '0'
|
||
orgs = await sor.sqlExe("SELECT skills_dir FROM sd_org_settings WHERE org_id=${o}$",{"o":oid})
|
||
if orgs: ctx['skills_dir'] = getattr(orgs[0],'skills_dir','') or ''
|
||
repos = await sor.sqlExe("SELECT repo_name,repo_url FROM sd_project_repos WHERE project_id=${p}$",{"p":ctx['pid']})
|
||
ctx['repos'] = [{'n':r.repo_name,'u':r.repo_url} for r in (repos or [])]
|
||
return ctx
|
||
|
||
async def _save_ctx(sor, uid, pid):
|
||
ex = await sor.sqlExe("SELECT id FROM pipeline_agent_settings WHERE user_id=${u}$",{"u":uid})
|
||
if ex:
|
||
await sor.sqlExe("UPDATE pipeline_agent_settings SET current_project_id=${p}$ WHERE user_id=${u}$",{"p":pid or '','u':uid})
|
||
else:
|
||
await sor.C('pipeline_agent_settings',{'id':getID(),'user_id':uid,'default_llm_id':'','current_project_id':pid or '','current_iteration_id':''})
|
||
|
||
async def _sel_model(sor, pref):
|
||
if pref:
|
||
recs = await sor.sqlExe("SELECT id,name,provider,model_id,api_base,api_key FROM llm WHERE (id=${l}$ OR name=${l}$) AND status='active'",{"l":pref})
|
||
if recs: return recs[0]
|
||
recs = await sor.sqlExe("SELECT id,name,provider,model_id,api_base,api_key FROM llm WHERE status='active' LIMIT 1",{})
|
||
return recs[0] if recs else None
|
||
|
||
async def _call_llm(model, msgs, temp):
|
||
hdrs = {"Authorization": f"Bearer {model.api_key or ''}", "Content-Type": "application/json"}
|
||
payload = {"model": model.model_id, "messages": msgs, "temperature": temp}
|
||
to = aiohttp.ClientTimeout(total=180)
|
||
async with aiohttp.ClientSession(timeout=to) as s:
|
||
async with s.post(f"{model.api_base.rstrip('/')}/chat/completions", headers=hdrs, json=payload) as r:
|
||
if r.status != 200: raise ValueError(f"LLM {r.status}")
|
||
return (await r.json())["choices"][0]["message"]["content"]
|
||
|
||
def _parse(raw):
|
||
raw = (raw or "").strip()
|
||
if raw.startswith("```"): raw = raw.split("\n",1)[1].rsplit("```",1)[0].strip()
|
||
try:
|
||
d = json.loads(raw)
|
||
if isinstance(d,dict) and 'action' in d: return d
|
||
except: pass
|
||
return {"action":"reply","message":raw}
|
||
|
||
# Widget helpers
|
||
def _w_text(t): return {"widgettype":"Text","options":{"text":t,"css":"agent-text"}}
|
||
def _w_card(title, body, kind="success"):
|
||
colors = {"success":"#10b981","error":"#ef4444","reply":"#6366f1","progress":"#f59e0b"}
|
||
c = colors.get(kind,"#6b7280")
|
||
return {"widgettype":"VBox","options":{"css":"agent-card","style":{"borderLeft":f"3px solid {c}","padding":"8px 12px","margin":"4px 0"}},"subwidgets":[
|
||
{"widgettype":"Text","options":{"text":title,"css":"agent-card-title","style":{"fontWeight":"bold","fontSize":"13px","color":c}}},
|
||
body
|
||
]}
|
||
def _w_progress(text): return {"widgettype":"Text","options":{"text":text,"css":"agent-progress","style":{"color":"#f59e0b","fontSize":"12px","padding":"2px 8px"}}}
|
||
|
||
_tool_labels = {'switch_project':'切换项目','create_task':'创建任务','check_progress':'检查进度','list_tasks':'查询任务','add_repo':'关联仓库','list_repos':'查看仓库','add_bug':'报告Bug'}
|
||
|
||
async def _exec_tool(sor, tool, params, ctx, uid, org_id):
|
||
p = params or {}
|
||
try:
|
||
if tool == 'switch_project':
|
||
proj = await _find_project(sor, p.get('project',''))
|
||
if proj:
|
||
await _save_ctx(sor, uid, proj.id)
|
||
ctx['pid'] = proj.id; ctx['pname'] = proj.name
|
||
return 'OK: 已切换到「'+proj.name+'」'
|
||
allp = await sor.sqlExe("SELECT name FROM sd_projects ORDER BY created_at DESC LIMIT 15",{})
|
||
return 'FAIL: 未找到。可用: '+', '.join([getattr(r,'name','') for r in (allp or [])])
|
||
|
||
elif tool == 'create_task':
|
||
if not ctx['pid']: return 'FAIL: 请先切换到项目'
|
||
title = p.get('title','')[:100]
|
||
if not title: return 'FAIL: 缺少标题'
|
||
role = p.get('role','') or _guess_role(title)
|
||
tp = {'description':p.get('description',''),'project_id':ctx['pid']}
|
||
r = await pipeline_role_submit(ctx['pid'],'role_task',uid,title,tp,role)
|
||
rd = json.loads(r)
|
||
if rd.get('success'): return f"OK: 任务「{title}」已创建(角色:{role})"
|
||
return f"FAIL: {rd.get('message','')}"
|
||
|
||
elif tool == 'check_progress':
|
||
if not ctx['pid']: return 'FAIL: 请先切换到项目'
|
||
tid = p.get('task_id','')
|
||
if tid:
|
||
ts = await sor.sqlExe("SELECT title,state,role FROM pipeline_tasks WHERE id=${t}$",{"t":tid})
|
||
if not ts: return 'FAIL: 任务不存在'
|
||
t = ts[0]; return f"[{t.state}][{t.role}] {t.title}"
|
||
ts = await sor.sqlExe("SELECT state,count(*)c FROM pipeline_tasks WHERE tenant_id=${p}$ GROUP BY state",{"p":ctx['pid']})
|
||
summary = ', '.join([f"{r.state}:{r.c}" for r in (ts or [])]) or '无'
|
||
ags = await sor.sqlExe("SELECT role_name,status FROM pipeline_project_agents WHERE project_id=${p}$",{"p":ctx['pid']})
|
||
al = ', '.join([f"{r.role_name}({r.status})" for r in (ags or [])]) or '未配置'
|
||
ds = await sor.sqlExe("SELECT deliverable_type,title,review_status FROM pipeline_deliverables WHERE project_id=${p}$ ORDER BY created_at DESC LIMIT 3",{"p":ctx['pid']})
|
||
dl = '\n'.join([f"[{d.deliverable_type}] {d.title[:40]}({d.review_status})" for d in (ds or [])]) or '无'
|
||
return f"任务: {summary}\nAgent: {al}\n交付:\n{dl}"
|
||
|
||
elif tool == 'list_tasks':
|
||
if not ctx['pid']: return 'FAIL: 请先切换到项目'
|
||
sf = p.get('state','')
|
||
sql = "SELECT title,state,role FROM pipeline_tasks WHERE tenant_id=${p}$"
|
||
if sf: sql += " AND state=${s}$"
|
||
sql += " ORDER BY created_at DESC LIMIT 20"
|
||
ts = await sor.sqlExe(sql, {"p":ctx['pid'],"s":sf} if sf else {"p":ctx['pid']})
|
||
em = {'submitted':'⏳','running':'🔄','review':'👀','approved':'✅','completed':'🏁','failed':'❌'}
|
||
return '\n'.join([f"{em.get(t.state,'?')}[{t.state}][{t.role}]{t.title}" for t in ts]) if ts else '无任务'
|
||
|
||
elif tool == 'add_repo':
|
||
if not ctx['pid']: return 'FAIL: 请先切换到项目'
|
||
url = p.get('url','')
|
||
if not url: return 'FAIL: 需要仓库地址'
|
||
name = p.get('name','') or url.rstrip('/').split('/')[-1].replace('.git','')
|
||
ex = await sor.sqlExe("SELECT id FROM sd_project_repos WHERE project_id=${p}$ AND repo_url=${u}$",{"p":ctx['pid'],"u":url})
|
||
if ex: return '仓库已关联'
|
||
await sor.C('sd_project_repos',{'id':getID(),'project_id':ctx['pid'],'repo_name':name,'repo_url':url,'default_branch':'main','local_path':'','org_id':org_id})
|
||
ctx['repos'].append({'n':name,'u':url})
|
||
return f"OK: 已关联 {name}"
|
||
|
||
elif tool == 'list_repos':
|
||
if not ctx['pid']: return 'FAIL: 请先切换到项目'
|
||
rs = await sor.sqlExe("SELECT repo_name,repo_url FROM sd_project_repos WHERE project_id=${p}$",{"p":ctx['pid']})
|
||
return '\n'.join([f"{r.repo_name}: {r.repo_url}" for r in rs]) if rs else '无仓库'
|
||
|
||
elif tool == 'add_bug':
|
||
if not ctx['pid']: return 'FAIL: 请先切换到项目'
|
||
title = p.get('title','')[:100]
|
||
if not title: return 'FAIL: 缺少标题'
|
||
await sor.C('sd_bugs',{'id':getID(),'iteration_id':'','title':title,'description':p.get('description',''),'severity':'major','priority':'P1','status':'open','reporter_type':'human','reporter_id':uid})
|
||
return f"OK: Bug已记录 {title}"
|
||
|
||
return f'未实现: {tool}'
|
||
except Exception as e:
|
||
return f'ERROR: {str(e)[:300]}'
|
||
|
||
|
||
if action == 'send_message':
|
||
message_text = (params_kw or {}).get('message_text','').strip()
|
||
user_model_id = (params_kw or {}).get('model_id','')
|
||
if not message_text: return json.dumps({"error":"message_text is required"},ensure_ascii=False)
|
||
|
||
blocked, reason = _security_scan(message_text)
|
||
if blocked: return json.dumps({"success":True,"agent_reply":f"⚠️ {reason}"},ensure_ascii=False)
|
||
|
||
uid = await get_user()
|
||
org_id = await get_userorgid() or '0'
|
||
|
||
async def agent_stream():
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
ctx = await _load_ctx(sor, uid)
|
||
model = await _sel_model(sor, user_model_id)
|
||
if not model:
|
||
yield json.dumps({"type":"widget","widget":_w_card("❌ 错误",_w_text("无可用LLM"),"error")},ensure_ascii=False)+chr(10)
|
||
return
|
||
|
||
repos_str = ', '.join([r['n'] for r in ctx['repos']]) or '无'
|
||
env_text = f"项目: {ctx['pname'] or '未选择'}\n仓库: {repos_str}"
|
||
system = AGENT_PROMPT.replace('__TOOLS__',TOOLS_TEXT).replace('__ENV__',env_text)
|
||
msgs = [{"role":"system","content":system}]
|
||
|
||
history = await sor.sqlExe("SELECT role,content FROM pipeline_conversations ORDER BY created_at ASC LIMIT 20",{})
|
||
for h in (history or []):
|
||
msgs.append({"role":'user' if getattr(h,'role','')=='user' else 'assistant',"content":getattr(h,'content','')})
|
||
msgs.append({"role":"user","content":message_text})
|
||
|
||
await sor.C('pipeline_conversations',{'id':getID(),'role':'user','content':message_text,'msg_type':'text','org_id':org_id,'created_by':uid})
|
||
|
||
agent_reply = ''
|
||
for turn in range(10):
|
||
raw = await _call_llm(model, msgs, 0.4)
|
||
act = _parse(raw)
|
||
if act.get('action') == 'reply':
|
||
agent_reply = act.get('message', raw); break
|
||
if act.get('action') == 'tool_call':
|
||
tool = act.get('tool','')
|
||
label = _tool_labels.get(tool,tool)
|
||
yield json.dumps({"type":"widget","widget":_w_progress(f"🔄 {label}")},ensure_ascii=False)+chr(10)
|
||
result = await _exec_tool(sor, tool, act.get('params',{}), ctx, uid, org_id)
|
||
ok = result.startswith("OK:")
|
||
yield json.dumps({"type":"widget","widget":_w_card(
|
||
f"{'✅' if ok else '❌'} {label}", _w_text(result),
|
||
"success" if ok else "error")},ensure_ascii=False)+chr(10)
|
||
msgs.append({"role":"assistant","content":raw})
|
||
msgs.append({"role":"user","content":f"工具 {tool} 结果:\n{result}"})
|
||
continue
|
||
agent_reply = raw; break
|
||
|
||
if not agent_reply: agent_reply = "处理超时"
|
||
await sor.C('pipeline_conversations',{'id':getID(),'role':'agent','content':agent_reply,'msg_type':'text','org_id':org_id,'created_by':'system'})
|
||
yield json.dumps({"type":"widget","widget":_w_card("🤖 驾驶舱 Agent",_w_text(agent_reply),"reply")},ensure_ascii=False)+chr(10)
|
||
|
||
return await stream_response(request, agent_stream, 'application/x-ndjson')
|
||
|
||
elif action == 'list_messages':
|
||
async with DBPools().sqlorContext(dbname) as sor:
|
||
ms = await sor.sqlExe("SELECT role,content,created_at FROM pipeline_conversations ORDER BY created_at ASC LIMIT 100",{})
|
||
result = [{"role":getattr(m,'role',''),"content":getattr(m,'content','')} for m in (ms or [])]
|
||
return json.dumps({"success":True,"messages":result},ensure_ascii=False,default=str)
|
||
|
||
else:
|
||
return json.dumps({"error":f"Unknown: {action}"},ensure_ascii=False)
|