From a5ea8197ab2ea72ca424c3a28f84cb6e3f7b529e Mon Sep 17 00:00:00 2001 From: WeiCHEN Date: Wed, 23 Sep 2026 14:51:09 +0800 Subject: [PATCH] fix(coordinator): force Tavily search for weather/realtime dialogue answers Anchor weather and other realtime questions on runner /v1/search results (or an explicit refuse constraint) so the model cannot invent dates or temperatures from training memory. --- internal/agent/pi_runner_search.go | 96 ++++++++++++++++++++ internal/agent/pi_runner_search_test.go | 69 ++++++++++++++ internal/app/app.go | 6 ++ internal/coordinator/context.go | 4 + internal/coordinator/dialogue_llm.go | 16 ++++ internal/coordinator/dialogue_llm_test.go | 95 +++++++++++++++++++ internal/coordinator/engine.go | 27 ++++-- internal/coordinator/memory_composer.go | 19 +++- internal/coordinator/memory_composer_test.go | 44 +++++++++ internal/coordinator/realtime_query.go | 38 ++++++++ internal/coordinator/realtime_query_test.go | 34 +++++++ internal/intent/recognizer_helpers.go | 5 + 12 files changed, 443 insertions(+), 10 deletions(-) create mode 100644 internal/agent/pi_runner_search.go create mode 100644 internal/agent/pi_runner_search_test.go create mode 100644 internal/coordinator/realtime_query.go create mode 100644 internal/coordinator/realtime_query_test.go diff --git a/internal/agent/pi_runner_search.go b/internal/agent/pi_runner_search.go new file mode 100644 index 00000000..876bfff0 --- /dev/null +++ b/internal/agent/pi_runner_search.go @@ -0,0 +1,96 @@ +package agent + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" + "time" +) + +const piRunnerSearchTimeout = 20 * time.Second + +// PiRunnerSearcher 通过 pi-agent runner 的 /v1/search 做 Tavily 检索。 +type PiRunnerSearcher struct { + URL string + Token string + Client *http.Client +} + +// Search 调用 runner /v1/search，返回 markdown 结果。 +func (s *PiRunnerSearcher) Search(ctx context.Context, query string) (string, error) { + if s == nil { + return "", fmt.Errorf("pi runner searcher is nil") + } + base := strings.TrimRight(strings.TrimSpace(s.URL), "/") + q := strings.TrimSpace(query) + if base == "" { + return "", fmt.Errorf("pi runner url empty") + } + if q == "" { + return "", fmt.Errorf("search query empty") + } + + body, err := json.Marshal(map[string]any{ + "query": q, + "max_results": 5, + }) + if err != nil { + return "", err + } + + reqCtx, cancel := context.WithTimeout(ctx, piRunnerSearchTimeout) + defer cancel() + + httpReq, err := http.NewRequestWithContext(reqCtx, http.MethodPost, base+"/v1/search", bytes.NewReader(body)) + if err != nil { + return "", err + } + httpReq.Header.Set("Content-Type", "application/json") + ApplyRunnerAuth(httpReq, s.Token) + + client := s.Client + if client == nil { + client = &http.Client{Timeout: piRunnerSearchTimeout} + } + resp, err := client.Do(httpReq) + if err != nil { + return "", err + } + defer resp.Body.Close() + + limited, err := io.ReadAll(io.LimitReader(resp.Body, MaxResultBodyBytes+1)) + if err != nil { + return "", err + } + if len(limited) > MaxResultBodyBytes { + return "", fmt.Errorf("search response too large") + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return "", fmt.Errorf("search http %d", resp.StatusCode) + } + + var parsed struct { + OK bool `json:"ok"` + Markdown string `json:"markdown"` + Error string `json:"error"` + } + if err := json.Unmarshal(limited, &parsed); err != nil { + return "", fmt.Errorf("decode search response: %w", err) + } + if !parsed.OK { + msg := strings.TrimSpace(parsed.Error) + if msg == "" { + msg = "search failed" + } + return "", fmt.Errorf("%s", msg) + } + md := strings.TrimSpace(parsed.Markdown) + if md == "" { + return "", fmt.Errorf("empty search markdown") + } + return md, nil +} diff --git a/internal/agent/pi_runner_search_test.go b/internal/agent/pi_runner_search_test.go new file mode 100644 index 00000000..2e26df9a --- /dev/null +++ b/internal/agent/pi_runner_search_test.go @@ -0,0 +1,69 @@ +package agent + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestPiRunnerSearcher_Success(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/search" { + t.Fatalf("path=%s", r.URL.Path) + } + if r.Header.Get("Authorization") != "Bearer tok-1" { + t.Fatalf("missing auth: %q", r.Header.Get("Authorization")) + } + var body map[string]any + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Fatalf("decode: %v", err) + } + if body["query"] != "今天北京天气" { + t.Fatalf("query=%v", body["query"]) + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "ok": true, + "markdown": "## 北京天气\n气温 21~33°C", + }) + })) + defer srv.Close() + + s := &PiRunnerSearcher{URL: srv.URL, Token: "tok-1", Client: srv.Client()} + md, err := s.Search(context.Background(), "今天北京天气") + if err != nil { + t.Fatalf("Search: %v", err) + } + if !strings.Contains(md, "21~33") { + t.Fatalf("markdown=%q", md) + } +} + +func TestPiRunnerSearcher_HTTPFailure(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusBadGateway) + _, _ = w.Write([]byte(`{"ok":false,"error":"upstream"}`)) + })) + defer srv.Close() + + s := &PiRunnerSearcher{URL: srv.URL, Client: srv.Client()} + _, err := s.Search(context.Background(), "今天天气") + if err == nil { + t.Fatal("expected error") + } +} + +func TestPiRunnerSearcher_EmptyMarkdown(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(map[string]any{"ok": true, "markdown": " "}) + })) + defer srv.Close() + + s := &PiRunnerSearcher{URL: srv.URL, Client: srv.Client()} + _, err := s.Search(context.Background(), "新闻") + if err == nil { + t.Fatal("expected empty markdown error") + } +} diff --git a/internal/app/app.go b/internal/app/app.go index 71b2af40..732fedd3 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -116,6 +116,12 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) { } dialogueBackend := agent.NewBackend(backendName, cfg) coordEngine := coordinator.NewCoordinatorEngineWithDialogue(nil, dialogueBackend, cfg.DataDir) + if cfg != nil && strings.TrimSpace(cfg.PiAgentRunnerURL) != "" { + coordEngine.SetRealtimeSearcher(&agent.PiRunnerSearcher{ + URL: cfg.PiAgentRunnerURL, + Token: cfg.PiAgentRunnerAuthToken, + }) + } parts := bizdev.Snapshot(svc) if ctxmgr.SemanticSummaryShadowEnabled() && parts.Ctx != nil && backendName == "pi-agent" { model := "" diff --git a/internal/coordinator/context.go b/internal/coordinator/context.go index f51c7fd6..c9039ab1 100644 --- a/internal/coordinator/context.go +++ b/internal/coordinator/context.go @@ -222,6 +222,10 @@ type UserProjectContext struct { UpgradeOffered bool `json:"-"` // DocumentSummaryReader 仅用于服务端鉴权后的只读正文访问，不序列化。 DocumentSummaryReader DocumentSummaryReader `json:"-"` + // RealtimeSnippet 本轮实时检索（天气/新闻等）成功结果，供 Prompt 注入；不序列化。 + RealtimeSnippet string `json:"-"` + // RealtimeUnavailable 本轮命中实时查询但检索失败/空结果，强制拒答约束；不序列化。 + RealtimeUnavailable bool `json:"-"` } // ContextAssembler 上下文组装服务 diff --git a/internal/coordinator/dialogue_llm.go b/internal/coordinator/dialogue_llm.go index 7da297c1..f7f43ff4 100644 --- a/internal/coordinator/dialogue_llm.go +++ b/internal/coordinator/dialogue_llm.go @@ -58,6 +58,22 @@ func (e *CoordinatorEngine) generateDialogueReply( return "", ReplySourceTemplate } + if userCtx != nil { + userCtx.RealtimeSnippet = "" + userCtx.RealtimeUnavailable = false + if e.realtimeSearcher != nil && IsRealtimeInformationQuery(userInput) { + if st := stream.FromContext(ctx); st != nil { + st.EmitPhase("dialogue_realtime_search", "正在检索实时信息...", "assistant", 0, 0, 40) + } + md, searchErr := e.realtimeSearcher.Search(ctx, userInput) + if searchErr != nil || strings.TrimSpace(md) == "" { + userCtx.RealtimeUnavailable = true + } else { + userCtx.RealtimeSnippet = strings.TrimSpace(md) + } + } + } + prompt := buildDialoguePrompt(userInput, userCtx) instructions := `你是「水木东方」医疗器械「项目助手」——温和、靠谱、会替用户着想的项目助理。 定位：以助理身份与用户友好协作。能力覆盖商务研判、工程方案与法规合规，但对外只以「项目助手」自称，不要分饰多角色，也不要自称「小队/参谋/总工/法规专家」。 diff --git a/internal/coordinator/dialogue_llm_test.go b/internal/coordinator/dialogue_llm_test.go index d8e6e8c6..f5a625b0 100644 --- a/internal/coordinator/dialogue_llm_test.go +++ b/internal/coordinator/dialogue_llm_test.go @@ -393,3 +393,98 @@ func TestDialogueLLM_AmbiguousSmalltalkDoesNotUseStageSalesTemplate(t *testing.T t.Fatalf("expected casual-chat template fallback, got: %s", res.NaturalReply) } } + + +type stubRealtimeSearcher struct { + markdown string + err error + calls int + lastQ string +} + +func (s *stubRealtimeSearcher) Search(ctx context.Context, query string) (string, error) { + s.calls++ + s.lastQ = query + return s.markdown, s.err +} + +func TestDialogueLLM_RealtimeSearchInjectsSnippet(t *testing.T) { + backend := &scriptedDialogueBackend{reply: "依据检索：北京今日约 25°C。"} + tmp := t.TempDir() + engine := NewCoordinatorEngineWithDialogue(nil, backend, tmp) + searcher := &stubRealtimeSearcher{markdown: "## 北京天气\n气温 21~33°C"} + engine.SetRealtimeSearcher(searcher) + + userCtx := &UserProjectContext{ + Project: ProjectInfo{ + ID: "p-rt-1", + Name: "全自动冲击波治疗仪", + CurrentStage: maturity.StageV01_Triage, + }, + BlackboardFacts: map[string]string{}, + } + rec := &intent.IntentRecognitionResult{ + RawInput: "今天北京天气怎么样，出去见客户要准备什么吗", + Branch: intent.BranchNewIntentPreempt, + TopIntent: &intent.CandidateIntent{ + Intent: intent.IntentCasualChat, + Confidence: 0.85, + Branch: intent.BranchNewIntentPreempt, + }, + } + res, err := engine.ProcessIntent(context.Background(), rec, userCtx) + if err != nil { + t.Fatalf("ProcessIntent: %v", err) + } + if searcher.calls != 1 { + t.Fatalf("expected 1 search call, got %d", searcher.calls) + } + if !strings.Contains(backend.lastPrompt, "【实时检索结果】") { + t.Fatalf("prompt missing realtime section: %s", backend.lastPrompt) + } + if !strings.Contains(backend.lastPrompt, "21~33") { + t.Fatalf("prompt missing snippet: %s", backend.lastPrompt) + } + if res.ReplySource != ReplySourcePiAgent { + t.Fatalf("reply_source=%q", res.ReplySource) + } +} + +func TestDialogueLLM_RealtimeSearchFailureInjectsRefuseConstraint(t *testing.T) { + backend := &scriptedDialogueBackend{reply: "抱歉，暂时无法获取实时天气数据，建议您查看天气 App。"} + tmp := t.TempDir() + engine := NewCoordinatorEngineWithDialogue(nil, backend, tmp) + searcher := &stubRealtimeSearcher{err: context.DeadlineExceeded} + engine.SetRealtimeSearcher(searcher) + + userCtx := &UserProjectContext{ + Project: ProjectInfo{ + ID: "p-rt-2", + Name: "全自动冲击波治疗仪", + CurrentStage: maturity.StageV01_Triage, + }, + BlackboardFacts: map[string]string{}, + } + rec := &intent.IntentRecognitionResult{ + RawInput: "今天北京天气怎么样", + Branch: intent.BranchNewIntentPreempt, + TopIntent: &intent.CandidateIntent{ + Intent: intent.IntentCasualChat, + Confidence: 0.9, + Branch: intent.BranchNewIntentPreempt, + }, + } + _, err := engine.ProcessIntent(context.Background(), rec, userCtx) + if err != nil { + t.Fatalf("ProcessIntent: %v", err) + } + if searcher.calls != 1 { + t.Fatalf("expected 1 search call, got %d", searcher.calls) + } + if strings.Contains(backend.lastPrompt, "【实时检索结果】") { + t.Fatalf("should not inject empty realtime section: %s", backend.lastPrompt) + } + if !strings.Contains(backend.lastPrompt, "本轮实时检索失败") { + t.Fatalf("missing refuse constraint: %s", backend.lastPrompt) + } +} diff --git a/internal/coordinator/engine.go b/internal/coordinator/engine.go index 43c4972c..01ceeada 100644 --- a/internal/coordinator/engine.go +++ b/internal/coordinator/engine.go @@ -65,15 +65,16 @@ type CoordinationResult struct { // CoordinatorEngine L2 协调引擎 type CoordinatorEngine struct { - stageTasks StageTaskRunner - instSvc *projectinstance.Service - assembler *ContextAssembler - dialogueBackend agent.Backend - dataDir string - ctxMgr *ctxmgr.ContextManager - kbClient *kb.Client - impact ImpactActions - projectAccess ProjectAccess + stageTasks StageTaskRunner + instSvc *projectinstance.Service + assembler *ContextAssembler + dialogueBackend agent.Backend + realtimeSearcher RealtimeSearcher + dataDir string + ctxMgr *ctxmgr.ContextManager + kbClient *kb.Client + impact ImpactActions + projectAccess ProjectAccess } // StageTaskRunner is the application owner for ExecuteTask / ExecuteStage. @@ -109,6 +110,14 @@ func (e *CoordinatorEngine) SetInstanceService(svc *projectinstance.Service) { } } + +// SetRealtimeSearcher 注入对话路径的实时检索（天气/新闻等）；nil 时跳过强制检索。 +func (e *CoordinatorEngine) SetRealtimeSearcher(s RealtimeSearcher) { + if e != nil { + e.realtimeSearcher = s + } +} + // SetImpactActions wires read-only impact analysis and confirmed revision proposals. func (e *CoordinatorEngine) SetImpactActions(a ImpactActions) { if e != nil { diff --git a/internal/coordinator/memory_composer.go b/internal/coordinator/memory_composer.go index 8b90bec6..6aa29259 100644 --- a/internal/coordinator/memory_composer.go +++ b/internal/coordinator/memory_composer.go @@ -230,6 +230,18 @@ func (mc *MemoryComposer) ComposeDialoguePrompt(userInput string, userCtx *UserP sb.WriteString("\n") } + // ------------------------------------------------------------- + // 6b. 【实时检索结果】 天气/新闻等权威实时数据（强制锚定） + // ------------------------------------------------------------- + if snip := strings.TrimSpace(userCtx.RealtimeSnippet); snip != "" { + sb.WriteString("【实时检索结果】(权威实时数据，回答天气/新闻等只准以此为准)\n") + sb.WriteString(snip) + if !strings.HasSuffix(snip, "\n") { + sb.WriteString("\n") + } + sb.WriteString("\n") + } + // ------------------------------------------------------------- // 7. 【用户偏好画像】 PersonalizedGuidance (～300 字) // ------------------------------------------------------------- @@ -259,7 +271,12 @@ func (mc *MemoryComposer) ComposeDialoguePrompt(userInput string, userCtx *UserP sb.WriteString("1. 优先级裁决：项目黑板核心事实 > 本轮用户输入 > 近期对话 > 早期摘要 > 检索片段。\n") sb.WriteString("2. 事实与引用：涉及法规条款与标准时，优先依据【参考知识】与【项目基准与协同上下文】并沿用〔引用键〕格式；无把握时明确提示依据不足，严禁生造标准号。\n") sb.WriteString("3. 隐私与跨用户边界：其他成员姓名与私人习惯仅限理解语境，禁止外泄复述。\n") - sb.WriteString("4. 答复目标：请结合上下文，以温和严谨的「项目助手」口吻作答，并写入 reply.md。") + sb.WriteString("4. 答复目标：请结合上下文，以温和严谨的「项目助手」口吻作答，并写入 reply.md。\n") + if userCtx.RealtimeUnavailable { + sb.WriteString("5. 本轮实时检索失败：必须明确告知无法获取实时数据（建议查看天气 App），严禁编造任何具体天气/新闻数值。\n") + } else if strings.TrimSpace(userCtx.RealtimeSnippet) != "" { + sb.WriteString("5. 实时数据：涉及天气/新闻等时效问题，必须严格依据【实时检索结果】作答，不得使用训练记忆中的过期数值。\n") + } return sb.String() } diff --git a/internal/coordinator/memory_composer_test.go b/internal/coordinator/memory_composer_test.go index 60f88ae4..70244f9f 100644 --- a/internal/coordinator/memory_composer_test.go +++ b/internal/coordinator/memory_composer_test.go @@ -133,3 +133,47 @@ func TestMemoryComposer_HierarchyAndPriority(t *testing.T) { idxWorkMem, idxProject, idxSummary, idxRecent, idxRecalled, idxReference, idxGuidance, idxUser, idxReq) } } + + +func TestMemoryComposer_RealtimeSnippetAndUnavailable(t *testing.T) { + composer := coordinator.NewMemoryComposer() + fixed := time.Date(2026, 9, 23, 10, 41, 0, 0, time.FixedZone("CST", 8*3600)) + composer.NowFn = func() time.Time { return fixed } + + okCtx := &coordinator.UserProjectContext{ + Project: coordinator.ProjectInfo{ + ID: "proj_rt", + Name: "冲击波治疗仪", + CurrentStage: maturity.StageV01_Triage, + }, + BlackboardFacts: map[string]string{}, + RealtimeSnippet: "北京今日气温约 21~33°C，晴。", + } + okPrompt := composer.ComposeDialoguePrompt("今天北京天气怎么样", okCtx) + if !strings.Contains(okPrompt, "【实时检索结果】") { + t.Fatalf("missing realtime section: %s", okPrompt) + } + if !strings.Contains(okPrompt, "21~33") { + t.Fatalf("missing snippet body: %s", okPrompt) + } + if !strings.Contains(okPrompt, "必须严格依据【实时检索结果】作答") { + t.Fatalf("missing success constraint: %s", okPrompt) + } + + failCtx := &coordinator.UserProjectContext{ + Project: coordinator.ProjectInfo{ + ID: "proj_rt", + Name: "冲击波治疗仪", + CurrentStage: maturity.StageV01_Triage, + }, + BlackboardFacts: map[string]string{}, + RealtimeUnavailable: true, + } + failPrompt := composer.ComposeDialoguePrompt("今天北京天气怎么样", failCtx) + if strings.Contains(failPrompt, "【实时检索结果】") { + t.Fatalf("should not render empty realtime section: %s", failPrompt) + } + if !strings.Contains(failPrompt, "本轮实时检索失败") { + t.Fatalf("missing failure constraint: %s", failPrompt) + } +} diff --git a/internal/coordinator/realtime_query.go b/internal/coordinator/realtime_query.go new file mode 100644 index 00000000..62c3a3e2 --- /dev/null +++ b/internal/coordinator/realtime_query.go @@ -0,0 +1,38 @@ +package coordinator + +import ( + "context" + "strings" + "unicode/utf8" + + "chat-agent-system/internal/intent" +) + +// RealtimeSearcher 对话路径的实时信息检索（天气/新闻等）；由 app 注入 Pi Runner /v1/search。 +type RealtimeSearcher interface { + Search(ctx context.Context, query string) (markdown string, err error) +} + +var realtimeInfoMarkers = []string{ + "天气", "气温", "温度", "下雨", "下雪", "降雨", "降雪", "台风", "雾霾", + "天气预报", "气象", "风力", "体感", + "新闻", "头条", "股价", "汇率", "实时", "今日新闻", "热点新闻", +} + +// IsRealtimeInformationQuery 判定是否为需要实时检索的时效性问句。 +// 含强业务词典时不触发（避免「型检温度」「项目进度」等误伤）。 +func IsRealtimeInformationQuery(input string) bool { + trimmed := strings.TrimSpace(input) + if trimmed == "" || utf8.RuneCountInString(trimmed) < 2 { + return false + } + if intent.HasStrongBusinessLexicon(trimmed) { + return false + } + for _, m := range realtimeInfoMarkers { + if strings.Contains(trimmed, m) { + return true + } + } + return false +} diff --git a/internal/coordinator/realtime_query_test.go b/internal/coordinator/realtime_query_test.go new file mode 100644 index 00000000..cb139431 --- /dev/null +++ b/internal/coordinator/realtime_query_test.go @@ -0,0 +1,34 @@ +package coordinator + +import "testing" + +func TestIsRealtimeInformationQuery(t *testing.T) { + positives := []string{ + "今天北京天气怎么样", + "明天会下雨吗", + "外面气温多少", + "今天有什么新闻", + "美元汇率多少", + "查一下实时股价", + } + for _, in := range positives { + if !IsRealtimeInformationQuery(in) { + t.Fatalf("expected realtime query: %q", in) + } + } + + negatives := []string{ + "", + "a", + "项目进度怎么样", + "型检温度指标怎么定", + "注册路径需要注意什么", + "帮我推进 V0.5 方案", + "你好", + } + for _, in := range negatives { + if IsRealtimeInformationQuery(in) { + t.Fatalf("did not expect realtime query: %q", in) + } + } +} diff --git a/internal/intent/recognizer_helpers.go b/internal/intent/recognizer_helpers.go index d84310ee..e0eb6e17 100644 --- a/internal/intent/recognizer_helpers.go +++ b/internal/intent/recognizer_helpers.go @@ -182,6 +182,11 @@ func IsSmallTalkOrGeneralQuestion(input string) bool { return false } +// HasStrongBusinessLexicon 是否含强业务词典（项目/法规/文档治理等），供闲聊与实时检索判定复用。 +func HasStrongBusinessLexicon(input string) bool { + return hasStrongBusinessLexicon(input) +} + func hasStrongBusinessLexicon(input string) bool { markers := []string{ "项目", "阶段", "推进", "法规", "注册", "文档", "审核", "批准", "驳回",