mirror of
https://github.com/NoFxAiOS/nofx.git
synced 2026-07-11 23:07:01 +08:00
feat: real SSE streaming for agent chat + data cleanup on startup
SSE Streaming: - Add POST /api/agent/chat/stream endpoint with real SSE streaming - Agent.HandleMessageStream() streams final LLM response via onEvent callback - Tool-calling rounds use non-streaming; tool status sent as 'tool' events - Frontend uses ReadableStream API instead of fake word-by-word setTimeout - SSE events: tool (tool execution), delta (text chunk), done (complete), error Data Maintenance: - Add cleanupOldData() on server startup (decisions >90d, equity >180d) - Add CleanOldRecords to DecisionStore and EquityStore - Add store-level logger for cleanup operations Exchange Traders: - Add GetAccountInfo() to Bitget, Bybit, Indodax, KuCoin, Lighter, OKX traders - Remove unused imports in Aster/Lighter traders
This commit is contained in:
127
agent/agent.go
127
agent/agent.go
@@ -151,6 +151,45 @@ func (a *Agent) HandleMessage(ctx context.Context, userID int64, text string) (s
|
||||
return a.thinkAndAct(ctx, userID, lang, text)
|
||||
}
|
||||
|
||||
// HandleMessageStream is like HandleMessage but streams the final LLM response via SSE.
|
||||
// onEvent is called with (eventType, data) — see StreamEvent* constants.
|
||||
// Non-streamable responses (commands, trade confirmations) return immediately without events.
|
||||
func (a *Agent) HandleMessageStream(ctx context.Context, userID int64, text string, onEvent func(event, data string)) (string, error) {
|
||||
lang := a.config.Language
|
||||
if strings.HasPrefix(text, "[lang:") {
|
||||
if end := strings.Index(text, "] "); end > 0 {
|
||||
lang = text[6:end]; text = text[end+2:]
|
||||
}
|
||||
}
|
||||
|
||||
a.logger.Info("message (stream)", "user_id", userID, "text", text)
|
||||
|
||||
if resp, handled := a.handleSetupFlow(userID, text, lang); handled {
|
||||
return resp, nil
|
||||
}
|
||||
if text == "/help" || text == "/start" {
|
||||
return a.msg(lang, "help"), nil
|
||||
}
|
||||
if text == "/status" {
|
||||
return a.handleStatus(lang), nil
|
||||
}
|
||||
if text == "/clear" {
|
||||
a.history.Clear(userID)
|
||||
if lang == "zh" { return "🧹 对话记忆已清除。", nil }
|
||||
return "🧹 Conversation history cleared.", nil
|
||||
}
|
||||
if resp, handled := a.handleTradeConfirmation(ctx, userID, text, lang); handled {
|
||||
return resp, nil
|
||||
}
|
||||
if trade := parseTradeCommand(text); trade != nil {
|
||||
a.pending.Add(trade)
|
||||
a.pending.CleanExpired()
|
||||
return formatTradeConfirmation(trade, lang), nil
|
||||
}
|
||||
|
||||
return a.thinkAndActStream(ctx, userID, lang, text, onEvent)
|
||||
}
|
||||
|
||||
// thinkAndAct sends the user message to LLM with full context and tools.
|
||||
// The LLM decides what to do — analyze, query, trade, or just chat.
|
||||
// Supports a tool-calling loop: LLM can call tools, get results, and continue.
|
||||
@@ -263,6 +302,94 @@ func (a *Agent) thinkAndAct(ctx context.Context, userID int64, lang, text string
|
||||
return finalResp, nil
|
||||
}
|
||||
|
||||
// StreamEvent types sent via SSE to the frontend.
|
||||
const (
|
||||
StreamEventTool = "tool" // Tool is being called (shows status to user)
|
||||
StreamEventDelta = "delta" // Text chunk from LLM streaming
|
||||
StreamEventDone = "done" // Stream complete
|
||||
StreamEventError = "error" // Error occurred
|
||||
)
|
||||
|
||||
// thinkAndActStream is like thinkAndAct but streams the final LLM response via SSE.
|
||||
// Tool-calling rounds use non-streaming CallWithRequestFull.
|
||||
// Once tools are done, the final response is streamed via onEvent("delta", accumulated_text).
|
||||
// onEvent("tool", toolName) is sent when a tool is being called.
|
||||
func (a *Agent) thinkAndActStream(ctx context.Context, userID int64, lang, text string, onEvent func(event, data string)) (string, error) {
|
||||
if a.aiClient == nil {
|
||||
return a.noAIFallback(lang, text)
|
||||
}
|
||||
|
||||
systemPrompt := a.buildSystemPrompt(lang)
|
||||
enrichment := a.gatherContext(text)
|
||||
userPrompt := text
|
||||
if enrichment != "" {
|
||||
userPrompt = text + "\n\n---\n[NOFXi System Context - real-time data for reference]\n" + enrichment
|
||||
}
|
||||
|
||||
messages := []mcp.Message{mcp.NewSystemMessage(systemPrompt)}
|
||||
for _, msg := range a.history.Get(userID) {
|
||||
messages = append(messages, mcp.NewMessage(msg.Role, msg.Content))
|
||||
}
|
||||
messages = append(messages, mcp.NewUserMessage(userPrompt))
|
||||
a.history.Add(userID, "user", text)
|
||||
|
||||
tools := agentTools()
|
||||
|
||||
// Tool-calling loop: use non-streaming to detect tool calls.
|
||||
// Once no tool calls remain, stream the final response for real-time UX.
|
||||
const maxToolRounds = 5
|
||||
toolsUsed := false
|
||||
for round := 0; round < maxToolRounds; round++ {
|
||||
req := &mcp.Request{Messages: messages, Tools: tools, ToolChoice: "auto"}
|
||||
resp, err := a.aiClient.CallWithRequestFull(req)
|
||||
if err != nil {
|
||||
a.logger.Error("LLM call failed (stream)", "error", err, "round", round)
|
||||
return a.noAIFallback(lang, text)
|
||||
}
|
||||
|
||||
// No tool calls → final text response
|
||||
if len(resp.ToolCalls) == 0 {
|
||||
if !toolsUsed {
|
||||
// No tools at all — response was fast, just return it
|
||||
a.history.Add(userID, "assistant", resp.Content)
|
||||
return resp.Content, nil
|
||||
}
|
||||
// After tool rounds, we already have the response (non-streamed).
|
||||
// Still return it — the SSE handler sends it as "done".
|
||||
a.history.Add(userID, "assistant", resp.Content)
|
||||
return resp.Content, nil
|
||||
}
|
||||
|
||||
// Process tool calls
|
||||
toolsUsed = true
|
||||
a.logger.Info("LLM tool calls (stream)", "count", len(resp.ToolCalls), "round", round)
|
||||
assistantMsg := mcp.Message{Role: "assistant", ToolCalls: resp.ToolCalls}
|
||||
if resp.Content != "" {
|
||||
assistantMsg.Content = resp.Content
|
||||
}
|
||||
messages = append(messages, assistantMsg)
|
||||
|
||||
for _, tc := range resp.ToolCalls {
|
||||
onEvent(StreamEventTool, tc.Function.Name)
|
||||
a.logger.Info("executing tool", "name", tc.Function.Name, "call_id", tc.ID)
|
||||
result := a.handleToolCall(ctx, userID, lang, tc)
|
||||
messages = append(messages, mcp.Message{Role: "tool", Content: result, ToolCallID: tc.ID})
|
||||
}
|
||||
}
|
||||
|
||||
// Exhausted tool rounds — stream the final synthesis response
|
||||
finalReq := &mcp.Request{Messages: messages}
|
||||
finalText, err := a.aiClient.CallWithRequestStream(finalReq, func(chunk string) {
|
||||
onEvent(StreamEventDelta, chunk)
|
||||
})
|
||||
if err != nil {
|
||||
a.logger.Error("stream final response failed", "error", err)
|
||||
return a.noAIFallback(lang, text)
|
||||
}
|
||||
a.history.Add(userID, "assistant", finalText)
|
||||
return finalText, nil
|
||||
}
|
||||
|
||||
// buildSystemPrompt creates the system prompt that makes NOFXi behave like a real agent.
|
||||
func (a *Agent) buildSystemPrompt(lang string) string {
|
||||
// Gather live system state
|
||||
|
||||
72
agent/web.go
72
agent/web.go
@@ -83,6 +83,78 @@ func (w *WebHandler) HandleChat(rw http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(rw, 200, map[string]string{"response": resp})
|
||||
}
|
||||
|
||||
// HandleChatStream handles POST /api/agent/chat/stream — SSE streaming chat.
|
||||
// Sends server-sent events with types: tool, delta, done, error.
|
||||
func (w *WebHandler) HandleChatStream(rw http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
http.Error(rw, "method not allowed", 405)
|
||||
return
|
||||
}
|
||||
var req struct {
|
||||
Message string `json:"message"`
|
||||
UserID int64 `json:"user_id"`
|
||||
Lang string `json:"lang"`
|
||||
}
|
||||
if err := json.NewDecoder(io.LimitReader(r.Body, 64*1024)).Decode(&req); err != nil {
|
||||
writeJSON(rw, 400, map[string]string{"error": "invalid request"})
|
||||
return
|
||||
}
|
||||
if req.Message == "" {
|
||||
writeJSON(rw, 400, map[string]string{"error": "message required"})
|
||||
return
|
||||
}
|
||||
if req.UserID == 0 {
|
||||
req.UserID = 1
|
||||
}
|
||||
msg := req.Message
|
||||
if req.Lang != "" {
|
||||
msg = "[lang:" + req.Lang + "] " + msg
|
||||
}
|
||||
|
||||
// Set SSE headers
|
||||
rw.Header().Set("Content-Type", "text/event-stream")
|
||||
rw.Header().Set("Cache-Control", "no-cache")
|
||||
rw.Header().Set("Connection", "keep-alive")
|
||||
rw.Header().Set("X-Accel-Buffering", "no") // Disable nginx buffering
|
||||
rw.WriteHeader(200)
|
||||
|
||||
flusher, ok := rw.(http.Flusher)
|
||||
if !ok {
|
||||
writeSSE(rw, nil, "error", "streaming not supported")
|
||||
return
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(r.Context(), 120*time.Second)
|
||||
defer cancel()
|
||||
|
||||
resp, err := w.agent.HandleMessageStream(ctx, req.UserID, msg, func(event, data string) {
|
||||
writeSSE(rw, flusher, event, data)
|
||||
})
|
||||
if err != nil {
|
||||
w.logger.Error("agent HandleMessageStream failed", "error", err, "user_id", req.UserID)
|
||||
writeSSE(rw, flusher, "error", "Failed to process message. Please try again.")
|
||||
return
|
||||
}
|
||||
// Send final done event with complete response
|
||||
writeSSE(rw, flusher, "done", resp)
|
||||
}
|
||||
|
||||
// writeSSE writes a single SSE event.
|
||||
func writeSSE(w http.ResponseWriter, flusher http.Flusher, event, data string) {
|
||||
fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event, sseEscape(data))
|
||||
if flusher != nil {
|
||||
flusher.Flush()
|
||||
}
|
||||
}
|
||||
|
||||
// sseEscape escapes newlines in SSE data (each line needs a "data: " prefix).
|
||||
func sseEscape(s string) string {
|
||||
// SSE spec: multi-line data uses multiple "data:" lines
|
||||
// But we use JSON encoding to avoid this complexity
|
||||
b, _ := json.Marshal(s)
|
||||
return string(b)
|
||||
}
|
||||
|
||||
// HandleKlines proxies kline data from Binance.
|
||||
func (w *WebHandler) HandleKlines(rw http.ResponseWriter, r *http.Request) {
|
||||
symbol := r.URL.Query().Get("symbol")
|
||||
|
||||
Reference in New Issue
Block a user