6c4b063f76
Adds RBAC matrix API/UI behind feature flags, agent×role toggles with audit, and extends Agentic Ops with streaming chat, Kimi LLM support, and UI updates.
137 lines
4.3 KiB
Python
137 lines
4.3 KiB
Python
"""SSE streaming chat — respostas token-a-token como assistente."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from collections.abc import Iterator
|
|
from typing import Any
|
|
|
|
from app import auth
|
|
from app.agents import llm_client, store
|
|
from app.agents import messages as agent_messages
|
|
from app.agents.chat_context_builder import build_ops_context, build_suggested_actions
|
|
|
|
|
|
def _sse(payload: dict[str, Any]) -> str:
|
|
return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
|
|
|
|
|
|
def _thread_history_messages(conn, thread_id: int, limit: int = 16) -> list[dict]:
|
|
msgs = agent_messages.thread_messages(conn, thread_id)[-limit:]
|
|
out: list[dict] = []
|
|
for m in msgs:
|
|
if m["from_type"] == "human":
|
|
out.append({"role": "user", "content": m["body"]})
|
|
elif m["from_type"] == "agent":
|
|
out.append({"role": "assistant", "content": m["body"]})
|
|
return out
|
|
|
|
|
|
def chat_stream_generator(
|
|
*,
|
|
username: str,
|
|
user_role: str,
|
|
question: str,
|
|
target_agent: str,
|
|
agent_name: str,
|
|
agent_role: str,
|
|
include_findings: bool,
|
|
thread_id: int | None = None,
|
|
) -> Iterator[str]:
|
|
conn = auth.db()
|
|
try:
|
|
kb = store.search_kb(conn, question)
|
|
kb_snippets = [k["snippet"] for k in kb]
|
|
ops_context = build_ops_context(
|
|
conn,
|
|
question,
|
|
target_agent,
|
|
include_findings=include_findings,
|
|
kb_snippets=kb_snippets,
|
|
)
|
|
suggested_actions = build_suggested_actions(conn, question, target_agent)
|
|
|
|
tid = thread_id
|
|
history_msgs: list[dict] = []
|
|
if tid:
|
|
row = conn.execute("SELECT * FROM agent_threads WHERE id=?", (tid,)).fetchone()
|
|
if not row:
|
|
yield _sse({"type": "error", "message": "thread not found"})
|
|
return
|
|
history_msgs = _thread_history_messages(conn, tid)
|
|
else:
|
|
tid = agent_messages.create_thread(
|
|
conn,
|
|
subject=f"Chat: {question[:60]}",
|
|
primary_agent=target_agent,
|
|
severity="info",
|
|
)
|
|
|
|
agent_messages.post_message(
|
|
conn,
|
|
thread_id=tid,
|
|
from_type="human",
|
|
from_id=username,
|
|
to_type="agent",
|
|
to_id=target_agent,
|
|
body=question,
|
|
)
|
|
conn.commit()
|
|
|
|
messages = llm_client.build_chat_messages(
|
|
question=question,
|
|
kb_snippets=kb_snippets,
|
|
ops_context=ops_context,
|
|
user_role=user_role,
|
|
target_agent=target_agent,
|
|
agent_name=agent_name,
|
|
agent_role=agent_role,
|
|
history_messages=history_msgs,
|
|
)
|
|
|
|
yield _sse({"type": "start", "thread_id": tid, "agent": target_agent, "actions": suggested_actions})
|
|
|
|
token_iter, model_label = llm_client.stream_llm_chat(messages)
|
|
parts: list[str] = []
|
|
for tok in token_iter:
|
|
parts.append(tok)
|
|
yield _sse({"type": "token", "text": tok})
|
|
|
|
answer = "".join(parts).strip()
|
|
if not answer:
|
|
answer = (
|
|
"Não consegui gerar resposta agora. Verifique Groq/Ollama no health do Agentic Ops."
|
|
)
|
|
model_label = "error"
|
|
yield _sse({"type": "token", "text": answer})
|
|
|
|
agent_messages.post_message(
|
|
conn,
|
|
thread_id=tid,
|
|
from_type="agent",
|
|
from_id=target_agent,
|
|
to_type="human",
|
|
to_id=username,
|
|
body=answer,
|
|
context={"model": model_label, "kb_hits": len(kb), "streamed": True, "actions": suggested_actions},
|
|
)
|
|
store.log_event(
|
|
conn,
|
|
event_type="chat.stream" if thread_id else "chat.query",
|
|
message=question[:120],
|
|
agent_id=target_agent,
|
|
payload={"user": username, "model": model_label, "thread_id": tid},
|
|
)
|
|
conn.commit()
|
|
yield _sse({
|
|
"type": "done",
|
|
"answer": answer,
|
|
"model": model_label,
|
|
"thread_id": tid,
|
|
"kb_hits": len(kb),
|
|
"actions": suggested_actions,
|
|
})
|
|
except Exception as exc:
|
|
yield _sse({"type": "error", "message": str(exc)[:300]})
|
|
finally:
|
|
conn.close()
|