From cb0f37753b26699d0020181875cb0cb05c1db28a Mon Sep 17 00:00:00 2001 From: ymq Date: Sat, 8 Aug 2026 16:34:53 +0800 Subject: [PATCH] cockpit v3.1: streaming NDJSON backend + widget rendering frontend --- wwwroot/api/cockpit_chat.dspy | 198 +++++++++++++++------------------- 1 file changed, 86 insertions(+), 112 deletions(-) diff --git a/wwwroot/api/cockpit_chat.dspy b/wwwroot/api/cockpit_chat.dspy index 3dbef6a..52b7b3f 100644 --- a/wwwroot/api/cockpit_chat.dspy +++ b/wwwroot/api/cockpit_chat.dspy @@ -1,48 +1,39 @@ -// cockpit_chat.dspy - SDLC Agent Loop v3 (JSON, no streaming) +// 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，负责软件工程全生命周期管理。 +AGENT_PROMPT = """你是开发产线驾驶舱 Agent。需要数据时调工具，最终用 reply 回复。 -## 输出格式（每条消息二选一） -1. {"action":"tool_call","tool":"工具名","params":{"参数":"值"}} -2. {"action":"reply","message":"回复内容（中文，简洁专业）"} -规则：需要数据时调工具，不要编造；一个消息一个动作。 +输出格式二选一： +{"action":"tool_call","tool":"工具名","params":{}} +{"action":"reply","message":"中文回复"} -## 工具 -__TOOLS__ - -## 当前环境 -__ENV__""" +工具: __TOOLS__ +环境: __ENV__""" TOOLS = [ - {"name":"switch_project","description":"切换项目","params":{"project":"名称或关键字"}}, - {"name":"create_project","description":"创建项目","params":{"name":"名称","description":"描述(可选)"}}, - {"name":"create_task","description":"提交任务","params":{"title":"标题","description":"详细描述","role":"requirement/design/develop/test/deploy"}}, + {"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":"check_progress","description":"检查项目进度","params":{"task_id":"任务ID(可选)"}}, - {"name":"add_bug","description":"报告Bug","params":{"title":"标题","description":"描述","severity":"严重程度(可选)"}}, - {"name":"list_bugs","description":"列出Bug","params":{}}, - {"name":"get_deliverable","description":"获取交付件","params":{"task_id":"任务ID"}}, - {"name":"shell_exec","description":"执行shell","params":{"command":"命令","workdir":"工作目录(可选)"}}, - {"name":"list_skills","description":"列出技能","params":{}}, + {"name":"add_bug","description":"报告Bug","params":{"title":"标题","description":"描述"}}, ] -TOOLS_TEXT = json.dumps(TOOLS, ensure_ascii=False, indent=2) +TOOLS_TEXT = json.dumps(TOOLS, ensure_ascii=False) def _security_scan(text): t = (text or '').lower() - for kw in ('drop table','drop database','truncate','rm -rf','密钥','api_key'): - if kw in t: return True, '危险操作/提示注入' + 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,role in [('需求','requirement'),('设计','design'),('开发','develop'),('实现','develop'),('编码','develop'),('测试','test'),('部署','deploy'),('发布','deploy')]: - if kw in t: return role + for kw,r in [('需求','requirement'),('设计','design'),('开发','develop'),('实现','develop'),('测试','test'),('部署','deploy')]: + if kw in t: return r return 'develop' async def _find_project(sor, name): @@ -92,7 +83,7 @@ async def _call_llm(model, msgs, 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}: {(await r.text())[:300]}") + if r.status != 200: raise ValueError(f"LLM {r.status}") return (await r.json())["choices"][0]["message"]["content"] def _parse(raw): @@ -104,6 +95,23 @@ def _parse(raw): 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"}}} +def _w_line(): + parts = [] + parts.append(json.dumps({"type":"widget","widget":w}, ensure_ascii=False)) + return '\n'.join(parts) + +_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: @@ -112,23 +120,9 @@ async def _exec_tool(sor, tool, params, ctx, uid, org_id): if proj: await _save_ctx(sor, uid, proj.id) ctx['pid'] = proj.id; ctx['pname'] = proj.name - return 'OK: 已切换到「' + 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_project': - name = p.get('name','')[:80] - if not name: return 'FAIL: 缺少项目名称' - ex = await sor.sqlExe("SELECT id FROM sd_projects WHERE name=${n}$",{"n":name}) - if ex: return 'FAIL: 已存在' - pid = getID(); ws = os.path.expanduser(f'~/pipeline_ws/{name}') - os.makedirs(ws, exist_ok=True) - await sor.C('sd_projects',{'id':pid,'name':name,'description':p.get('description',''),'project_type':'software','org_id':org_id,'created_by':uid,'status':'active','workspace_dir':ws}) - iid = getID() - await sor.C('sd_iterations',{'id':iid,'project_id':pid,'iteration_name':'默认迭代','iteration_type':'sprint','org_id':org_id,'created_by':uid,'status':'active'}) - await _save_ctx(sor, uid, pid) - ctx['pid'] = pid; ctx['pname'] = name - return f'OK: 项目「{name}」已创建 ({ws})' + return 'FAIL: 未找到。可用: '+', '.join([getattr(r,'name','') for r in (allp or [])]) elif tool == 'create_task': if not ctx['pid']: return 'FAIL: 请先切换到项目' @@ -138,7 +132,7 @@ async def _exec_tool(sor, tool, params, ctx, uid, org_id): 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})" + if rd.get('success'): return f"OK: 任务「{title}」已创建(角色:{role})" return f"FAIL: {rd.get('message','')}" elif tool == 'check_progress': @@ -149,12 +143,12 @@ async def _exec_tool(sor, tool, params, ctx, uid, org_id): 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 [])]) + 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 [])]) + 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 [])]) - return f"任务: {summary or '无'}\nAgent: {al or '未配置'}\n交付:\n{dl or '无'}" + 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: 请先切换到项目' @@ -163,60 +157,31 @@ async def _exec_tool(sor, tool, params, ctx, uid, org_id): 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']}) - if not ts: return '当前无任务' 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]) + 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: 请提供仓库地址' + 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}' + 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 '暂无仓库' + 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':p.get('severity','major'),'priority':'P1','status':'open','reporter_type':'human','reporter_id':uid}) - return f'OK: Bug已记录: {title}' - - elif tool == 'list_bugs': - bs = await sor.sqlExe("SELECT title,severity,status FROM sd_bugs ORDER BY created_at DESC LIMIT 20",{}) - return '\n'.join([f"[{b.severity}] {b.title}({b.status})" for b in bs]) if bs else '暂无Bug' - - elif tool == 'get_deliverable': - tid = p.get('task_id','') - if not tid: return 'FAIL: 请提供任务ID' - ds = await sor.sqlExe("SELECT deliverable_type,title,content FROM pipeline_deliverables WHERE task_id=${t}$ ORDER BY created_at DESC LIMIT 1",{"t":tid}) - if not ds: return '无交付件' - d = ds[0]; return f"[{d.deliverable_type}] {d.title}\n{(d.content or '')[:3000]}" - - elif tool == 'shell_exec': - cmd = p.get('command',''); wd = p.get('workdir','') or ctx.get('ws','') - if not cmd: return 'FAIL: 请提供命令' - r = await shell_exec(cmd, workdir=wd, timeout=120) - return f"OK:\n{r['stdout'][:2000]}" if r['rc']==0 else f"FAIL({r['rc']}):\n{r['stderr'][:1000]}" - - elif tool == 'list_skills': - sd = ctx.get('skills_dir','') - if not sd or not os.path.isdir(sd): return 'FAIL: Skills目录未配置' - found = {} - for rd in sorted(os.listdir(sd)): - rp = os.path.join(sd,rd) - if os.path.isdir(rp): - ss = [d for d in os.listdir(rp) if os.path.isdir(os.path.join(rp,d)) and os.path.isfile(os.path.join(rp,d,'SKILL.md'))] - if ss: found[rd] = ss - return '\n'.join([f"[{r}]{', '.join(sorted(n))}" for r,n in sorted(found.items())]) if found else '无技能' + 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: @@ -234,42 +199,51 @@ if action == 'send_message': uid = await get_user() org_id = await get_userorgid() or '0' - async with DBPools().sqlorContext(dbname) as sor: - ctx = await _load_ctx(sor, uid) - model = await _sel_model(sor, user_model_id) - if not model: return json.dumps({"success":True,"agent_reply":"❌ 没有可用的LLM模型配置"},ensure_ascii=False) + 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)+'\n' + return - repos_str = ', '.join([r['n'] for r in ctx['repos']]) or '无' - env_text = f"项目: {ctx['pname'] or '未选择'}\n仓库: {repos_str}\nSkills: {ctx['skills_dir'] or '未配置'}" - system = AGENT_PROMPT.replace('__TOOLS__',TOOLS_TEXT).replace('__ENV__',env_text) + 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",{}) - msgs = [{"role":"system","content":system}] - 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}) + 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}) + 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','') - result = await _exec_tool(sor, tool, act.get('params',{}), ctx, uid, org_id) - msgs.append({"role":"assistant","content":raw}) - msgs.append({"role":"user","content":f"工具 {tool} 结果:\n{result}"}) - continue - agent_reply = raw; break + 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)+'\n' + 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)+'\n' + 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'}) + 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)+'\n' - return json.dumps({"success":True,"agent_reply":agent_reply},ensure_ascii=False) + return await stream_response(request, agent_stream, 'application/x-ndjson') elif action == 'list_messages': async with DBPools().sqlorContext(dbname) as sor: