yumoqing/pipeline-sdlc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SSE mode: text/event-stream + data: prefix + [DONE] end

Browse Source 
Backend: _sse() wraps each widget as 'data: <json>\n\n', ends with [DONE]
Frontend: XHR.onprogress splits on 'data: ', extracts json before \n\n
This commit is contained in:
ymq 2026-08-08 21:22:34 +08:00
parent 620b12dee0
commit bac65a61aa
2 changed files with 21 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

BIN
wwwroot/api/.cockpit_chat.dspy.swp Normal file
View File

Binary file not shown.

36
wwwroot/api/cockpit_chat.dspy
View File

@ -1,4 +1,4 @@
# cockpit_chat.dspy - SDLC Agent Loop v3.1 # cockpit_chat.dspy - SDLC Agent Loop v3.1 (SSE streaming)
import aiohttp, json, os import aiohttp, json, os
action = (params_kw or {}).get('action', 'list_messages') action = (params_kw or {}).get('action', 'list_messages')
@ -95,6 +95,10 @@ def _parse(raw):
except: pass except: pass
return {"action":"reply","message":raw} return {"action":"reply","message":raw}
# SSE helper: format as "data: <json>\n\n"
def _sse(data):
return f"data: {data}\n\n"
def _w_text(t): return {"widgettype":"Text","options":{"text":t,"css":"agent-text"}} def _w_text(t): return {"widgettype":"Text","options":{"text":t,"css":"agent-text"}}
def _w_card(title, body, kind="success"): def _w_card(title, body, kind="success"):
colors = {"success":"#10b981","error":"#ef4444","reply":"#6366f1","progress":"#f59e0b"} colors = {"success":"#10b981","error":"#ef4444","reply":"#6366f1","progress":"#f59e0b"}
@ -194,14 +198,15 @@ if action == 'send_message':
uid = await get_user() uid = await get_user()
org_id = await get_userorgid() or '0' org_id = await get_userorgid() or '0'
async def agent_stream(): async def agent_sse():
async with DBPools().sqlorContext(dbname) as sor: async with DBPools().sqlorContext(dbname) as sor:
ctx = await _load_ctx(sor, uid) ctx = await _load_ctx(sor, uid)
model = await _sel_model(sor, user_model_id) model = await _sel_model(sor, user_model_id)
if not model: if not model:
d = json.dumps({"type":"widget","widget":_w_card("❌ 错误",_w_text("无可用LLM"),"error")},ensure_ascii=False)+chr(10) d = json.dumps({"type":"widget","widget":_w_card("❌ 错误",_w_text("无可用LLM"),"error")},ensure_ascii=False)
debug(f"AGENT_STREAM yield error: {d}") debug(f"SSE yield error: {d[:80]}")
yield d yield _sse(d)
yield _sse("[DONE]")
return return
repos_str = ', '.join([r['n'] for r in ctx['repos']]) or '无' repos_str = ', '.join([r['n'] for r in ctx['repos']]) or '无'
@ -225,27 +230,28 @@ if action == 'send_message':
if act.get('action') == 'tool_call': if act.get('action') == 'tool_call':
tool = act.get('tool','') tool = act.get('tool','')
label = _tool_labels.get(tool,tool) label = _tool_labels.get(tool,tool)
d = json.dumps({"type":"widget","widget":_w_progress(f"🔄 {label}")},ensure_ascii=False)+chr(10) d = json.dumps({"type":"widget","widget":_w_progress(f"🔄 {label}")},ensure_ascii=False)
debug(f"AGENT_STREAM yield: {d}") debug(f"SSE yield progress: {d[:80]}")
yield d yield _sse(d)
result = await _exec_tool(sor, tool, act.get('params',{}), ctx, uid, org_id) result = await _exec_tool(sor, tool, act.get('params',{}), ctx, uid, org_id)
ok = result.startswith("OK:") ok = result.startswith("OK:")
d = 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) d = json.dumps({"type":"widget","widget":_w_card(f"{'✅' if ok else '❌'} {label}",_w_text(result),"success" if ok else "error")},ensure_ascii=False)
debug(f"AGENT_STREAM yield: {d}") debug(f"SSE yield result: {d[:80]}")
yield d yield _sse(d)
msgs.append({"role":"assistant","content":raw}) msgs.append({"role":"assistant","content":raw})
msgs.append({"role":"user","content":f"工具 {tool} 结果:\n{result}"}) msgs.append({"role":"user","content":f"工具 {tool} 结果:\n{result}"})
continue continue
agent_reply = raw; break agent_reply = raw; break
if not agent_reply: agent_reply = "处理超时，请简化需求。" if not agent_reply: agent_reply = "处理超时，请简化需求。"
d = json.dumps({"type":"widget","widget":_w_card("🤖 驾驶舱 Agent",_w_text(agent_reply),"reply")},ensure_ascii=False)+chr(10) d = json.dumps({"type":"widget","widget":_w_card("🤖 驾驶舱 Agent",_w_text(agent_reply),"reply")},ensure_ascii=False)
debug(f"AGENT_STREAM yield final: {d}") debug(f"SSE yield final: {d[:80]}")
yield d yield _sse(d)
yield _sse("[DONE]")
await sor.C('pipeline_conversations',{'id':getID(),'role':'agent','content':agent_reply,'msg_type':'text','org_id':org_id,'created_by':'system'}) await sor.C('pipeline_conversations',{'id':getID(),'role':'agent','content':agent_reply,'msg_type':'text','org_id':org_id,'created_by':'system'})
return await stream_response(request, agent_stream, 'application/x-ndjson') return await stream_response(request, agent_sse, 'text/event-stream')
elif action == 'list_messages': elif action == 'list_messages':
async with DBPools().sqlorContext(dbname) as sor: async with DBPools().sqlorContext(dbname) as sor: