feat: implement streaming chat, skill routing, and SAFe PI planning tools

- Add /api/chat/stream endpoint with Server-Sent Events (SSE) for real-time message streaming
  * Implement StreamEvent types (thought, tool_call, tool_result, final, error)
  * Add StreamEventCallback mechanism for event propagation
  * Create StreamChatHandler in webui/bot with proper HTTP headers and flushing

- Implement LLM-based skill router for intelligent capability selection
  * Add optional routerLLM client for semantic routing
  * Implement routeSkillsWithLLM() to match user intent to available skills
  * Add matchSkillsByName() for fuzzy skill matching
  * Update buildUnifiedSystemPrompt() to use routed skills

- Add streaming support to ReAct pipeline
  * Implement runUnifiedReActStream() for streaming thought/action/observation
  * Emit StreamEvent at each ReAct step
  * Support callback error handling in streaming mode

- Integrate three new DevOps tools
  * tools/filedoc: Extract document content from file_id via OpenAI
  * tools/giteaticket: Create Gitea issues from PI plan items with SAFe metadata
  * tools/piplan: Publish PI planning blueprints with dependency tracking

- Add SAFe PI Planning skill
  * Implement PM/SA/RTE (iron triangle) workflow
  * Support for Feature, Enabler, and Dependency definition
  * Automatic task decomposition and Gitea integration

- Create frontend integration documentation
  * Complete SSE protocol specification
  * TypeScript fetch + ReadableStream example
  * LLM-ready refactoring template for other projects

- Simplify file handling
  * Remove legacy file context structures and dual-mode processing
  * Consolidate file operations into UploadAndCacheFiles()
  * Remove FilePromptMode configuration and related complexity

- Update configuration
  * Add Router model support (LLM_ROUTER_MODEL)
  * Add Gitea configuration (BaseURL, Token, Owner, Repo)
  * WebSearch and additional tool infrastructure

Tests: All 22 test packages passing, 8/8 webui tests including 3 new stream tests
This commit is contained in:
2026-03-11 17:58:19 +08:00
parent 0e1a800646
commit 8dc5354fa4
17 changed files with 3086 additions and 565 deletions

View File

@@ -18,13 +18,33 @@ import (
)
type IncomingMessage struct {
ChatID string
UserID string
Text string
FileIDs []string
ChatID string
UserID string
Text string
}
// StreamEventType 定义流式输出的事件类型
type StreamEventType string
const (
StreamEventTypeThought StreamEventType = "thought" // LLM 思考过程
StreamEventTypeToolCall StreamEventType = "tool_call" // 工具调用请求
StreamEventTypeToolResult StreamEventType = "tool_result" // 工具执行结果
StreamEventTypeFinal StreamEventType = "final" // 最终答案
StreamEventTypeError StreamEventType = "error" // 错误信息
)
// StreamEvent 代表流式输出中的一个事件
type StreamEvent struct {
Type StreamEventType `json:"type"`
Content string `json:"content"`
Step int `json:"step,omitempty"`
ToolName string `json:"tool_name,omitempty"`
}
type ChatHandler func(context.Context, IncomingMessage) (string, error)
type StreamChatHandler func(context.Context, IncomingMessage, StreamEventCallback) (string, error)
type StreamEventCallback func(event StreamEvent) error
type UploadHandler func(context.Context, string, string, []llm.InputFile) ([]string, error)
type Bot struct {
@@ -32,29 +52,25 @@ type Bot struct {
maxUploadBytes int64
log *logger.Logger
chatHandler ChatHandler
uploadHandler UploadHandler
counter uint64
chatHandler ChatHandler
streamChatHandler StreamChatHandler
uploadHandler UploadHandler
counter uint64
}
type chatRequest struct {
Text string `json:"text"`
SessionID string `json:"session_id"`
UserID string `json:"user_id"`
FileIDs []string `json:"file_ids"`
Text string `json:"text"`
SessionID string `json:"session_id"`
UserID string `json:"user_id"`
}
func (r *chatRequest) UnmarshalJSON(data []byte) error {
type rawChatRequest struct {
Text string `json:"text"`
SessionID string `json:"session_id"`
SessionIDCamel string `json:"sessionId"`
UserID string `json:"user_id"`
UserIDCamel string `json:"userId"`
FileIDs json.RawMessage `json:"file_ids"`
FileIDsCamel json.RawMessage `json:"fileIds"`
FileIDsFlat json.RawMessage `json:"fileids"`
FileID json.RawMessage `json:"file_id"`
Text string `json:"text"`
SessionID string `json:"session_id"`
SessionIDCamel string `json:"sessionId"`
UserID string `json:"user_id"`
UserIDCamel string `json:"userId"`
}
var raw rawChatRequest
@@ -65,13 +81,6 @@ func (r *chatRequest) UnmarshalJSON(data []byte) error {
r.Text = raw.Text
r.SessionID = firstNonEmpty(raw.SessionID, raw.SessionIDCamel)
r.UserID = firstNonEmpty(raw.UserID, raw.UserIDCamel)
rawIDs := firstNonEmptyRaw(raw.FileIDs, raw.FileIDsCamel, raw.FileIDsFlat, raw.FileID)
ids, err := decodeStringList(rawIDs)
if err != nil {
return err
}
r.FileIDs = ids
return nil
}
@@ -109,7 +118,7 @@ func NewBot(cfg config.WebUIConfig, log *logger.Logger) (*Bot, error) {
}, nil
}
func (b *Bot) Run(ctx context.Context, chatHandler ChatHandler, uploadHandler UploadHandler) error {
func (b *Bot) Run(ctx context.Context, chatHandler ChatHandler, streamChatHandler StreamChatHandler, uploadHandler UploadHandler) error {
if chatHandler == nil {
return fmt.Errorf("nil webui chat handler")
}
@@ -117,10 +126,12 @@ func (b *Bot) Run(ctx context.Context, chatHandler ChatHandler, uploadHandler Up
return fmt.Errorf("nil webui upload handler")
}
b.chatHandler = chatHandler
b.streamChatHandler = streamChatHandler
b.uploadHandler = uploadHandler
mux := http.NewServeMux()
mux.HandleFunc("/api/chat", b.handleChat)
mux.HandleFunc("/api/chat/stream", b.handleChatStream)
mux.HandleFunc("/api/upload", b.handleUpload)
srv := &http.Server{
@@ -191,10 +202,9 @@ func (b *Bot) handleChat(w http.ResponseWriter, r *http.Request) {
userID := b.resolveID(req.UserID, "user")
reply, err := b.chatHandler(r.Context(), IncomingMessage{
ChatID: sessionID,
UserID: userID,
Text: req.Text,
FileIDs: req.FileIDs,
ChatID: sessionID,
UserID: userID,
Text: req.Text,
})
if err != nil {
if b.log != nil {
@@ -210,37 +220,8 @@ func (b *Bot) handleChat(w http.ResponseWriter, r *http.Request) {
})
}
func decodeStringList(raw json.RawMessage) ([]string, error) {
if len(raw) == 0 {
return nil, nil
}
var list []string
if err := json.Unmarshal(raw, &list); err == nil {
return nonEmptyIDs(list), nil
}
var single string
if err := json.Unmarshal(raw, &single); err == nil {
if strings.TrimSpace(single) == "" {
return nil, nil
}
return nonEmptyIDs(strings.Split(single, ",")), nil
}
return nil, fmt.Errorf("invalid file ids format")
}
func firstNonEmptyRaw(vals ...json.RawMessage) json.RawMessage {
for _, v := range vals {
if len(v) > 0 {
return v
}
}
return nil
}
func firstNonEmpty(vals ...string) string {
for _, v := range vals {
if strings.TrimSpace(v) != "" {
return v
@@ -249,24 +230,82 @@ func firstNonEmpty(vals ...string) string {
return ""
}
func nonEmptyIDs(ids []string) []string {
if len(ids) == 0 {
func (b *Bot) handleChatStream(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
return
}
if !strings.Contains(strings.ToLower(strings.TrimSpace(r.Header.Get("Content-Type"))), "application/json") {
writeJSON(w, http.StatusBadRequest, errorResponse{Error: "content-type must be application/json"})
return
}
if b.streamChatHandler == nil {
writeJSON(w, http.StatusInternalServerError, errorResponse{Error: "stream chat handler not ready"})
return
}
var req chatRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeJSON(w, http.StatusBadRequest, errorResponse{Error: "invalid json body"})
return
}
req.Text = strings.TrimSpace(req.Text)
if req.Text == "" {
writeJSON(w, http.StatusBadRequest, errorResponse{Error: "text is required"})
return
}
sessionID := b.resolveID(req.SessionID, "sess")
userID := b.resolveID(req.UserID, "user")
// 设置 SSE 响应头
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
w.Header().Set("Access-Control-Allow-Origin", "*")
w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
w.WriteHeader(http.StatusOK)
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "streaming not supported", http.StatusInternalServerError)
return
}
// 创建回调函数来推送 SSE 事件
callback := func(event StreamEvent) error {
data, err := json.Marshal(event)
if err != nil {
return err
}
fmt.Fprintf(w, "data: %s\n\n", string(data))
flusher.Flush()
return nil
}
out := make([]string, 0, len(ids))
seen := map[string]struct{}{}
for _, id := range ids {
id = strings.TrimSpace(id)
if id == "" {
continue
// 调用流式处理器
reply, err := b.streamChatHandler(r.Context(), IncomingMessage{
ChatID: sessionID,
UserID: userID,
Text: req.Text,
}, callback)
if err != nil {
if b.log != nil {
b.log.Errorf("webui stream chat handler failed session_id=%s user_id=%s err=%v", sessionID, userID, err)
}
if _, ok := seen[id]; ok {
continue
// 推送错误事件
errEvent := StreamEvent{
Type: StreamEventTypeError,
Content: "stream error: " + err.Error(),
}
seen[id] = struct{}{}
out = append(out, id)
data, _ := json.Marshal(errEvent)
fmt.Fprintf(w, "data: %s\n\n", string(data))
flusher.Flush()
return
}
if b.log != nil {
b.log.Infof("webui stream chat completed session_id=%s user_id=%s reply_len=%d", sessionID, userID, len(reply))
}
return out
}
func (b *Bot) handleUpload(w http.ResponseWriter, r *http.Request) {

View File

@@ -4,6 +4,7 @@ import (
"bytes"
"context"
"encoding/json"
"errors"
"mime/multipart"
"net/http"
"net/http/httptest"
@@ -51,66 +52,6 @@ func TestHandleChatSuccess(t *testing.T) {
}
}
func TestHandleChatWithFileIDs(t *testing.T) {
b := newTestBot(t, 1024*1024)
b.chatHandler = func(_ context.Context, msg IncomingMessage) (string, error) {
if msg.ChatID != "s1" || msg.UserID != "u1" || msg.Text != "hello" {
t.Fatalf("unexpected message: %+v", msg)
}
if len(msg.FileIDs) != 2 || msg.FileIDs[0] != "file_a" || msg.FileIDs[1] != "file_b" {
t.Fatalf("unexpected file ids: %+v", msg.FileIDs)
}
return "ok", nil
}
body := strings.NewReader(`{"text":"hello","session_id":"s1","user_id":"u1","file_ids":["file_a","file_b"]}`)
req := httptest.NewRequest(http.MethodPost, "/api/chat", body)
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
b.handleChat(w, req)
if w.Code != http.StatusOK {
t.Fatalf("expected 200, got %d body=%s", w.Code, w.Body.String())
}
}
func TestHandleChatWithFileIDsAliases(t *testing.T) {
tests := []struct {
name string
body string
}{
{name: "camel array", body: `{"text":"hello","sessionId":"s1","userId":"u1","fileIds":["file_a","file_b"]}`},
{name: "flat array", body: `{"text":"hello","session_id":"s1","user_id":"u1","fileids":["file_a","file_b"]}`},
{name: "single key", body: `{"text":"hello","session_id":"s1","user_id":"u1","file_id":"file_a"}`},
{name: "csv string", body: `{"text":"hello","session_id":"s1","user_id":"u1","file_ids":"file_a, file_b"}`},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
b := newTestBot(t, 1024*1024)
b.chatHandler = func(_ context.Context, msg IncomingMessage) (string, error) {
if msg.ChatID != "s1" || msg.UserID != "u1" || msg.Text != "hello" {
t.Fatalf("unexpected message: %+v", msg)
}
if len(msg.FileIDs) == 0 {
t.Fatalf("expected file ids from alias payload, got empty")
}
return "ok", nil
}
body := strings.NewReader(tt.body)
req := httptest.NewRequest(http.MethodPost, "/api/chat", body)
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
b.handleChat(w, req)
if w.Code != http.StatusOK {
t.Fatalf("expected 200, got %d body=%s", w.Code, w.Body.String())
}
})
}
}
func TestHandleChatMissingText(t *testing.T) {
b := newTestBot(t, 1024*1024)
b.chatHandler = func(_ context.Context, _ IncomingMessage) (string, error) { return "", nil }
@@ -215,3 +156,149 @@ func TestHandleUploadMissingFile(t *testing.T) {
t.Fatalf("expected 400, got %d", w.Code)
}
}
func TestHandleChatStreamSuccess(t *testing.T) {
b := newTestBot(t, 1024*1024)
b.streamChatHandler = func(_ context.Context, msg IncomingMessage, cb StreamEventCallback) (string, error) {
if msg.ChatID != "s1" || msg.UserID != "u1" || msg.Text != "hello" {
t.Fatalf("unexpected message: %+v", msg)
}
if err := cb(StreamEvent{Type: StreamEventTypeThought, Content: "thinking", Step: 1}); err != nil {
return "", err
}
if err := cb(StreamEvent{Type: StreamEventTypeToolCall, Content: "{\"input\":\"pwd\"}", Step: 1, ToolName: "shell"}); err != nil {
return "", err
}
if err := cb(StreamEvent{Type: StreamEventTypeToolResult, Content: "C:/Project", Step: 1, ToolName: "shell"}); err != nil {
return "", err
}
if err := cb(StreamEvent{Type: StreamEventTypeFinal, Content: "done", Step: 2}); err != nil {
return "", err
}
return "done", nil
}
body := strings.NewReader(`{"text":"hello","session_id":"s1","user_id":"u1"}`)
req := httptest.NewRequest(http.MethodPost, "/api/chat/stream", body)
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
b.handleChatStream(w, req)
if w.Code != http.StatusOK {
t.Fatalf("expected 200, got %d body=%s", w.Code, w.Body.String())
}
if got := w.Header().Get("Content-Type"); got != "text/event-stream" {
t.Fatalf("expected text/event-stream, got %q", got)
}
var events []StreamEvent
chunks := strings.Split(strings.TrimSpace(w.Body.String()), "\n\n")
for _, chunk := range chunks {
line := strings.TrimSpace(chunk)
if line == "" {
continue
}
if !strings.HasPrefix(line, "data: ") {
t.Fatalf("invalid sse line: %q", line)
}
payload := strings.TrimPrefix(line, "data: ")
var ev StreamEvent
if err := json.Unmarshal([]byte(payload), &ev); err != nil {
t.Fatalf("unmarshal stream event failed: %v payload=%s", err, payload)
}
events = append(events, ev)
}
if len(events) != 4 {
t.Fatalf("expected 4 events, got %d", len(events))
}
if events[0].Type != StreamEventTypeThought {
t.Fatalf("event[0] type mismatch: %s", events[0].Type)
}
if events[1].Type != StreamEventTypeToolCall || events[1].ToolName != "shell" {
t.Fatalf("event[1] mismatch: %+v", events[1])
}
if events[2].Type != StreamEventTypeToolResult || events[2].ToolName != "shell" {
t.Fatalf("event[2] mismatch: %+v", events[2])
}
if events[3].Type != StreamEventTypeFinal || events[3].Content != "done" {
t.Fatalf("event[3] mismatch: %+v", events[3])
}
}
func TestHandleChatStreamHandlerError(t *testing.T) {
b := newTestBot(t, 1024*1024)
b.streamChatHandler = func(_ context.Context, _ IncomingMessage, _ StreamEventCallback) (string, error) {
return "", errors.New("boom")
}
body := strings.NewReader(`{"text":"hello"}`)
req := httptest.NewRequest(http.MethodPost, "/api/chat/stream", body)
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
b.handleChatStream(w, req)
if w.Code != http.StatusOK {
t.Fatalf("expected 200, got %d", w.Code)
}
respBody := w.Body.String()
if !strings.Contains(respBody, `"type":"error"`) {
t.Fatalf("expected error event in stream, body=%q", respBody)
}
if !strings.Contains(respBody, "stream error: boom") {
t.Fatalf("expected error detail in stream, body=%q", respBody)
}
}
func TestHandleChatStreamValidation(t *testing.T) {
t.Run("method not allowed", func(t *testing.T) {
b := newTestBot(t, 1024*1024)
req := httptest.NewRequest(http.MethodGet, "/api/chat/stream", nil)
w := httptest.NewRecorder()
b.handleChatStream(w, req)
if w.Code != http.StatusMethodNotAllowed {
t.Fatalf("expected 405, got %d", w.Code)
}
})
t.Run("content type must be json", func(t *testing.T) {
b := newTestBot(t, 1024*1024)
req := httptest.NewRequest(http.MethodPost, "/api/chat/stream", strings.NewReader(`{"text":"hello"}`))
w := httptest.NewRecorder()
b.handleChatStream(w, req)
if w.Code != http.StatusBadRequest {
t.Fatalf("expected 400, got %d", w.Code)
}
})
t.Run("handler not ready", func(t *testing.T) {
b := newTestBot(t, 1024*1024)
req := httptest.NewRequest(http.MethodPost, "/api/chat/stream", strings.NewReader(`{"text":"hello"}`))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
b.handleChatStream(w, req)
if w.Code != http.StatusInternalServerError {
t.Fatalf("expected 500, got %d", w.Code)
}
})
t.Run("text required", func(t *testing.T) {
b := newTestBot(t, 1024*1024)
b.streamChatHandler = func(_ context.Context, _ IncomingMessage, _ StreamEventCallback) (string, error) {
return "", nil
}
req := httptest.NewRequest(http.MethodPost, "/api/chat/stream", strings.NewReader(`{"text":" "}`))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
b.handleChatStream(w, req)
if w.Code != http.StatusBadRequest {
t.Fatalf("expected 400, got %d", w.Code)
}
})
}