Compare commits

..

38 Commits

Author SHA1 Message Date
姚章浩 1bbaf2f07a fix(outboundtext): keep resource handles intact when normalizing formulas 2026-09-18 17:14:45 +08:00
姚章浩 48b741984f fix(reasoning): never cut a lead-in that carries an image 2026-09-18 16:49:16 +08:00
姚章浩 5ccaeb3b96 fix(outbound): keep single-run answers whole and show formulas
The DingTalk binding can point at the quick-answer agent, which publishes the
whole answer as one answer run plus tool_call/tool_result steps. The narration
boundary was rejected for that shape (start=0), so the text-only heuristic ran
and dropped the answer lead-in together with its image.

- answer_observation: a single answer run means there is no narration to cut.
- outboundtext: rewrite LaTeX formulas into plain Unicode text, because
  DingTalk cards render no math typesetting (r = sqrt(X^2 + Y^2) -> r = √(X² + Y²)).
- dingtalk handler: normalize formulas on both the partial and final card paths.
2026-09-18 16:21:42 +08:00
姚章浩 dfce8cb9f6 test(qa): pin the answer-run boundary used to trim narration
Adds the production CRA5 shape (narration run + tool steps + final answer run
that opens with a lead-in and an image) and a control case where the last
answer run is interrupted by a tool call and the heuristic still applies.
2026-09-18 15:58:14 +08:00
姚章浩 149e94b34b fix(reasoning): keep the answer body's lead-in and images when trimming narration
The DingTalk card dropped the answer's opening (and any image reference in
it) whenever the agent narrated its steps before answering: the text-only
boundary heuristic (reasoningmode.BodyStartOffset) treats everything before
the first structural line as narration, so a lead-in like "我已经找到了…"
plus the image that follows it disappeared, leaving the card to start at
the first heading.

The event stream already knows the boundary: every agent step publishes its
answer chunks under its own answer run id (data.event_id), and the final
body is the last run that no tool/thinking event interrupts. WeKnora
StreamEvent now carries the run id and run start offset; qa.Service tracks
the observation and prefers it over the heuristic, falling back to the
heuristic when the stream carries no run information.

Verified: production turn efae317e (给我CRA5的负载图) lost 131 chars —
lead-in + the load-curve image + separator — and now keeps them.
2026-09-18 15:53:43 +08:00
hermes dd229ec0ad feat(dingtalk): per-frame wide-table grouping (stream keeps tables flowing while streaming) 2026-09-18 11:55:45 +08:00
hermes 92324b4242 feat(dingtalk): make wide-table grouping switchable for wide-screen PC cards
DINGTALK_CARD_WIDE_TABLE_GROUP=off keeps over-wide answer tables as tables on the final frame (still replaces <br> inside cells) so the PC wide-screen card can lay them out itself.
2026-09-18 11:45:55 +08:00
agent 882ac62433 perf(dingtalk): compress outbound screenshots, reuse uploaded objects, raise rewrite budget
- outboundimage: re-encode large PNG/JPEG screenshots as downscaled JPEG (1600px long edge, q82, transparency flattened onto white) before upload; undecodable, small, GIF/WebP payloads are published untouched
- outboundimage: StableHashPublisher skips the upload when the content-addressed object already exists (ObjectStater); OSS store implements it via IsObjectExist
- cmd/dobotai: DOBOTAI_OUTBOUND_IMAGE_REWRITE_TIMEOUT default 60s -> 180s; DOBOTAI_OUTBOUND_IMAGE_COMPRESS (on by default, off restores raw bytes)
2026-09-18 11:14:13 +08:00
agent 8a9c1eaf82 perf(dingtalk): rewrite answer images concurrently and make the rewrite budget configurable
- outboundimage: fetch and publish up to 4 images in parallel, dedupe in document order, log per-image outcome plus a rewrite summary
- dingtalk handler: OutboundRewriteTimeout applies to the image rewrite stage only, falling back to finalDeliveryTimeout
- cmd/dobotai: DOBOTAI_OUTBOUND_IMAGE_REWRITE_TIMEOUT (default 60s, invalid values fall back)
2026-09-18 10:57:32 +08:00
agent 9de1ed66aa feat(dingtalk): make outbound image count configurable (DOBOTAI_OUTBOUND_IMAGE_MAX_COUNT, default 12) 2026-09-18 10:16:14 +08:00
姚章浩 cfe3271fa2 feat(dingtalk): fall back to the public image URL and report image ref shape
The loopback fetch now degrades to the public URL ...
2026-09-17 18:20:02 +08:00
姚章浩 dfc7c8dba9 feat(dingtalk): deliver card images through DingTalk media ids
AI card bodies render only DingTalk-hosted media, so published image URLs are
rewritten into media ids inside card content (streaming frames never block,
final frame waits briefly), while the markdown fallback keeps external URLs.
2026-09-17 18:14:03 +08:00
姚章浩 06d11135e1 deploy(prod): 生产切换脚本(门控字面量+回滚标签+env 0600 备份) 2026-09-17 14:25:30 +08:00
姚章浩 5f1206b545 fix(dingtalk): 推理正文边界支持「首行即正文」并补齐卡片结构诊断 + 流式分帧重放守卫 2026-09-17 14:16:40 +08:00
姚章浩 a1a2aabab4 fix(dingtalk): 推理正文边界支持「首行即正文」并补齐卡片结构诊断
- BodyStartOffset:内容首个非空行即结构行时正文从头开始,避免收尾裁剪
  把标题+首段一起删掉(118 条生产样本中 31 条、最多 526 字)
- 语料回归守卫:隐藏前缀不得含结构行、上限 600 字;单测补 suffix 断言
- 卡片诊断新增 body_ok/body_line/body_prefix_runes/banner_pos/first_heading_line
  (定位流式残留推理末句,只看结构不看文本)
2026-09-17 14:01:10 +08:00
姚章浩 fcca33050a test: 通用的测试服镜像切换脚本(先打回滚标签+env 0600 备份,读回校验) 2026-09-17 11:08:23 +08:00
姚章浩 cd3f92aaaa verify: 合并宽屏自适应与推理输出改造(测试服验证用) 2026-09-17 11:03:05 +08:00
姚章浩 15f2a11162 fix(dingtalk): 推理模式卡片隐藏推理并只留正文
- reasoningmode: 新增正文边界识别(结构行 + 前置空行)与流式 PartialFeed:
  分步叙述按行拼接显示,正文开始时只输出正文,卡片更新即刷新清掉推理区
- dingtalk: 事件流与流式回调统一改走 PartialFeed
- qa: 推理执行的最终答案在入库/投递前去掉步骤叙述(快速模式行为不变)

语料回归:生产 118 条内建智能推理答案中 106 条识别出正文边界(被隐藏的
均为 59-164 字过渡句),12 条无推理的问候类答案保持原样;go vet 与
go test ./... 全绿。
2026-09-17 10:58:59 +08:00
hermes-agent a096ba5f51 fix(dingtalk): 投放 AI 卡片时开启宽屏自适应布局
PC 端卡片宽度过窄:创建卡片(createAndDeliver)请求体从未请求宽屏布局。

修复落在创建卡片的 JSON 载荷层 —— cardData.cardParamMap 的钉钉保留系统
字段 sys_full_json_obj 内传 {"config":{"autoLayout":true}}。config 为
对象类型,钉钉规定复杂类型变量必须走该保留字段;放在 cardParamMap 兄弟
节点会因变量未赋值而静默失效(宽度不变、无报错)。

- 仅影响 PC 宽屏渲染;移动端保持自适应布局不变。
- 未改动富文本组件宽度/padding,也未改动模板内静态内容。
- 流式更新(streaming)请求体不带该字段,避免中途变更布局。

新增回归守卫测试:投放体(direct/group)必须含该字段且 autoLayout=true,
流式更新体必须不含;防止退化为静默失效写法。

验证:gofmt 干净、go vet ok、go test ./... 全绿;
A/B 探针(A 带字段 / B 对照)经 PC 端人工确认 A 卡片变宽。
2026-09-17 09:59:39 +08:00
zheromiku ba090d1705 Merge pull request 'feat: default smart reasoning and robust channel delivery' (#16) from feat/default-reasoning-mode into main 2026-09-15 17:13:32 +08:00
姚章浩 4e65028d1d fix(dingtalk): strip leaked internal source handles 2026-09-15 17:08:33 +08:00
姚章浩 c7d2629a77 fix(coverage): let references override partial no-answer wording 2026-09-15 16:53:12 +08:00
姚章浩 b49cb5eaf9 feat(dingtalk): normalize wide tables into grouped lists on final card 2026-09-15 16:09:43 +08:00
姚章浩 16aa981028 test(reasoning): cover default-mode routing, user pinning and 000022 contract 2026-09-15 14:01:00 +08:00
姚章浩 f800127461 feat(reasoning): support operator-configurable default mode via DOBOTAI_DEFAULT_MODE 2026-09-15 12:25:09 +08:00
姚章浩 d6a781ba5b fix(dingtalk): send answer language via Accept-Language header
WeKnora's system prompt hard-enforces 'ALWAYS respond in {{language}}',
fed by its Language middleware (WEKNORA_LANGUAGE env -> Accept-Language
header -> zh-CN fallback). The previous localizedQuery() prompt prefix
lost against that instruction, so a user-set 'en' preference still got
Chinese answers on English questions.

- AgentChatRequest.Locale is transmitted as the Accept-Language header
  (never in the JSON body); WeKnora maps it to the prompt language
- 'auto' detects the query script client-side (latin->en, Han->zh-CN,
  kana->ja, Hangul->ko); es is never auto-guessed
- unset/unknown locale sends no header, preserving prior server behavior
- drop localizedQuery prompt prefixes entirely
2026-09-14 16:58:52 +08:00
姚章浩 168e401bc0 docs(deploy): table the optional outbound-image OSS fields 2026-09-11 20:38:06 +08:00
zheromiku 58b37f1c24 Merge pull request 'fix(deploy): configure outbound image OSS as an optional group' (#15) from feat/docker-external-rds into main 2026-09-11 18:48:56 +08:00
姚章浩 6a3f505965 fix(deploy): configure outbound image OSS as an optional group 2026-09-11 18:48:15 +08:00
zheromiku a0f1fbf391 Merge pull request 'feat(deploy): external-RDS Docker deployment with fielded RDS config' (#14) from feat/docker-external-rds into main 2026-09-11 18:39:20 +08:00
姚章浩 49e0789fa5 feat(deploy): prompt RDS config as separate fields 2026-09-11 18:04:38 +08:00
姚章浩 d77fbd675b fix(deploy): honor configured Go module proxy 2026-09-11 17:27:20 +08:00
姚章浩 a615693bd0 fix(deploy): configure distinct quick and reasoning agents 2026-09-11 13:26:17 +08:00
姚章浩 febf99b549 fix(deploy): generate internal tenant key automatically 2026-09-11 13:20:44 +08:00
姚章浩 b4b61c0aa5 fix(deploy): explain missing RDS database to administrators 2026-09-11 12:25:31 +08:00
姚章浩 6153533b46 fix(deploy): isolate installer command directory 2026-09-11 12:10:08 +08:00
姚章浩 27d4d45af4 feat(deploy): run Docker application against external RDS 2026-09-11 12:09:29 +08:00
zheromiku a31156dc4e Merge pull request 'docs: 清理历史计划并说明仓库目录结构' (#13) from feat/cli-admin-bootstrap into main 2026-09-11 10:18:28 +08:00
72 changed files with 5345 additions and 681 deletions
+2
View File
@@ -2,6 +2,8 @@ FROM golang:1.26.5-bookworm AS build
WORKDIR /src
COPY go.mod go.sum ./
ARG GOPROXY=https://proxy.golang.org,direct
ENV GOPROXY=${GOPROXY}
RUN go mod download
COPY . ./
+29
View File
@@ -76,6 +76,35 @@ DobotAI 是 WeKnora 的企业外部渠道集成层,负责钉钉协议、身份
目录职责规则:`cmd/`只装配进程,`internal/`承载业务模块,`internal/store/postgres/`只实现数据库持久化,`db/migrations/`只放Schema变更,`deploy/`只放容器运行资产。历史计划、备份、镜像归档和真实环境文件均不进入Git。
## 部署方式
| 方式 | 适用场景 | 数据库 | 入口 |
| --- | --- | --- | --- |
| 源码/原生二进制 | 本地开发、快速调试 | 手工配置阿里云RDS连接 | `go run ./cmd/dobotai` |
| Docker | 测试、预发布、生产候选 | 既有阿里云RDS;Docker不运行数据库 | `sudo bash ./install-docker.sh` |
### 源码部署
```bash
go run ./cmd/dobotai
curl --noproxy '*' -fsS http://127.0.0.1:8090/healthz
```
源码方式不自动安装Docker、不创建目录、不检查RDS表,也不适合作为长期渠道消费者。
### Docker + 阿里云RDS部署
克隆仓库后在根目录执行:
```bash
sudo bash ./install-docker.sh
sudo dobotai-dingtalk-configure
```
第一条命令自动安装Docker/Compose(如缺失)、构建应用镜像、准备目录和配置模板;第二条命令安全录入RDS、WeKnora与钉钉配置,然后检查RDS表:空库初始化`000001`至`000021`,完整表不写入,异常/半成品表拒绝启动。Docker只运行`dobotai-dingtalk`,不创建`dobotai-postgres`、本地数据卷或本地数据库。
完整安装、RDS检查、升级和回滚见 [Docker部署手册](deploy/dobotai-dingtalk/README.md)。
## 钉钉 Stream
使用官方 `dingtalk-stream-sdk-go`,不暴露公网回调入口。
+27
View File
@@ -0,0 +1,27 @@
package main
import (
"log"
"os"
"strings"
)
// cardWideTableGroupingMode returns the configured handling of over-wide answer
// tables (see dingtalk.tableGroupingForMode):
//
// unset / on → grouped list on every frame (historical behaviour)
// off → real tables on every frame
// stream → grouped list while streaming, real table on the final frame
//
// The default stays "on" so an environment without the variable renders exactly
// as before; DINGTALK_CARD_WIDE_TABLE_GROUP selects the other modes.
func cardWideTableGroupingMode() string {
mode := strings.ToLower(strings.TrimSpace(os.Getenv("DINGTALK_CARD_WIDE_TABLE_GROUP")))
switch mode {
case "off", "0", "false", "no", "stream", "streaming", "partial":
log.Printf("DingTalk wide-table grouping mode=%s", mode)
return mode
default:
return ""
}
}
+17
View File
@@ -79,10 +79,22 @@ func main() {
if err != nil {
log.Fatal(err)
}
defaultMode := reasoningmode.ModeQuick
if raw := strings.TrimSpace(os.Getenv("DOBOTAI_DEFAULT_MODE")); raw != "" {
switch raw {
case "reasoning":
defaultMode = reasoningmode.ModeReasoning
case "quick":
defaultMode = reasoningmode.ModeQuick
default:
log.Fatal("DOBOTAI_DEFAULT_MODE must be quick or reasoning")
}
}
reasoningService, err := reasoningmode.NewService(
store,
os.Getenv("DOBOTAI_QUICK_AGENT_ID"),
os.Getenv("DOBOTAI_REASONING_AGENT_ID"),
reasoningmode.WithDefaultMode(defaultMode),
)
if err != nil {
log.Fatal(err)
@@ -295,6 +307,7 @@ func main() {
}
cardSender = cs
}
cardImageMedia := buildCardImageMedia(outboundImages.publicURLPrefix, dingCfg.ClientID, dingCfg.ClientSecret, address)
handler, err := dingtalk.NewHandlerWithOptions(
dingtalk.TenantConfig{TenantKey: dingCfg.TenantKey, Channel: "dingtalk"},
qaService,
@@ -305,9 +318,13 @@ func main() {
CardSender: cardSender,
Directory: directoryResolver,
OutboundRewriter: outboundImages.rewriter,
CardImageMedia: cardImageMedia,
SettingsRouter: &dingTalkSettingsRouter{controller: settingsRuntime.controller},
ReasoningRouter: &dingTalkReasoningRouter{service: reasoningService},
LanguageResolver: settingsRuntime.repository,
OutboundRewriteTimeout: outboundImages.rewriteTimeout,
WideTableGrouping: cardWideTableGroupingMode(),
},
)
if err != nil {
+50
View File
@@ -5,6 +5,7 @@ import (
"os"
"strings"
"testing"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/audit"
)
@@ -184,6 +185,55 @@ func TestOutboundOSSConfiguredRequiresStablePublicBaseURL(t *testing.T) {
}
}
func TestOutboundImageBudgetAndCompressionDefaults(t *testing.T) {
if outboundImageRewriteTimeoutDefault != 180*time.Second {
t.Fatalf("default rewrite budget=%s", outboundImageRewriteTimeoutDefault)
}
t.Setenv("DOBOTAI_OUTBOUND_IMAGE_REWRITE_TIMEOUT", "")
if got := outboundImageRewriteTimeout(); got != outboundImageRewriteTimeoutDefault {
t.Fatalf("unset budget=%s", got)
}
t.Setenv("DOBOTAI_OUTBOUND_IMAGE_REWRITE_TIMEOUT", "2m30s")
if got := outboundImageRewriteTimeout(); got != 150*time.Second {
t.Fatalf("parsed budget=%s", got)
}
for _, invalid := range []string{"-1s", "180", "soon"} {
t.Setenv("DOBOTAI_OUTBOUND_IMAGE_REWRITE_TIMEOUT", invalid)
if got := outboundImageRewriteTimeout(); got != outboundImageRewriteTimeoutDefault {
t.Fatalf("invalid budget %q resolved to %s", invalid, got)
}
}
t.Setenv("DOBOTAI_OUTBOUND_IMAGE_COMPRESS", "")
if !outboundImageCompressionEnabled() {
t.Fatal("compression must default to on")
}
for _, disabled := range []string{"off", "0", "false", "no"} {
t.Setenv("DOBOTAI_OUTBOUND_IMAGE_COMPRESS", disabled)
if outboundImageCompressionEnabled() {
t.Fatalf("compression must be off for %q", disabled)
}
}
}
func TestCardWideTableGroupingMode(t *testing.T) {
t.Setenv("DINGTALK_CARD_WIDE_TABLE_GROUP", "")
if got := cardWideTableGroupingMode(); got != "" {
t.Fatalf("unset must keep the historical mode, got %q", got)
}
t.Setenv("DINGTALK_CARD_WIDE_TABLE_GROUP", "off")
if got := cardWideTableGroupingMode(); got != "off" {
t.Fatalf("off must be passed through, got %q", got)
}
t.Setenv("DINGTALK_CARD_WIDE_TABLE_GROUP", " STREAM ")
if got := cardWideTableGroupingMode(); got != "stream" {
t.Fatalf("stream must be normalised, got %q", got)
}
t.Setenv("DINGTALK_CARD_WIDE_TABLE_GROUP", "on")
if got := cardWideTableGroupingMode(); got != "" {
t.Fatalf("on is the default and must not be forwarded as a mode, got %q", got)
}
}
func TestMainDoesNotExposeWebAdministrationOrSetupOnlyMode(t *testing.T) {
data, err := os.ReadFile("main.go")
if err != nil {
+115 -7
View File
@@ -2,7 +2,11 @@ package main
import (
"errors"
"log"
"net"
"net/http"
"os"
"strconv"
"strings"
"time"
@@ -11,12 +15,63 @@ import (
)
const (
outboundImageMaxCount = 5
outboundImageMaxBytes = 10 << 20
outboundImagePrefix = "dobotai/im-images"
outboundImageRoute = "/dingtalk/i/"
outboundImageMaxCountDefault = 12
outboundImageMaxBytes = 10 << 20
outboundImagePrefix = "dobotai/im-images"
outboundImageRoute = "/dingtalk/i/"
// outboundImageRewriteTimeoutDefault is the budget for one answer's image
// rewrite. Manual screenshots are multi-megabyte uploads and the container's
// uplink is the real ceiling (measured ~0.6MB/s), so long answers need a
// budget with headroom: the historical 15s delivery timeout truncated them.
outboundImageRewriteTimeoutDefault = 180 * time.Second
)
// outboundImageMaxCount resolves how many images one answer may rewrite.
// DOBOTAI_OUTBOUND_IMAGE_MAX_COUNT overrides the default; invalid values fall back.
func outboundImageMaxCount() int {
raw := strings.TrimSpace(os.Getenv("DOBOTAI_OUTBOUND_IMAGE_MAX_COUNT"))
if raw == "" {
return outboundImageMaxCountDefault
}
n, err := strconv.Atoi(raw)
if err != nil || n < 1 {
log.Printf("outbound image: invalid DOBOTAI_OUTBOUND_IMAGE_MAX_COUNT=%q; using default %d", raw, outboundImageMaxCountDefault)
return outboundImageMaxCountDefault
}
return n
}
// outboundImageRewriteTimeout resolves the budget for the image rewrite stage.
// DOBOTAI_OUTBOUND_IMAGE_REWRITE_TIMEOUT accepts a Go duration ("90s", "2m");
// invalid or non-positive values fall back to the default.
func outboundImageRewriteTimeout() time.Duration {
raw := strings.TrimSpace(os.Getenv("DOBOTAI_OUTBOUND_IMAGE_REWRITE_TIMEOUT"))
if raw == "" {
return outboundImageRewriteTimeoutDefault
}
duration, err := time.ParseDuration(raw)
if err != nil || duration <= 0 {
log.Printf("outbound image: invalid DOBOTAI_OUTBOUND_IMAGE_REWRITE_TIMEOUT=%q; using default %s", raw, outboundImageRewriteTimeoutDefault)
return outboundImageRewriteTimeoutDefault
}
return duration
}
// outboundImageCompressionEnabled reports whether published screenshots are
// re-encoded as smaller JPEGs before upload. It is on by default because the
// container uplink, not the rewrite code, caps how many images fit in the
// delivery budget; DOBOTAI_OUTBOUND_IMAGE_COMPRESS=off restores the original
// bytes exactly as fetched.
func outboundImageCompressionEnabled() bool {
switch strings.ToLower(strings.TrimSpace(os.Getenv("DOBOTAI_OUTBOUND_IMAGE_COMPRESS"))) {
case "off", "0", "false", "no":
log.Printf("outbound image: compression disabled by configuration")
return false
default:
return true
}
}
type outboundImageConfig struct {
WeKnoraBaseURL string
WeKnoraAPIKey string
@@ -32,6 +87,11 @@ type outboundImageRuntime struct {
rewriter dingtalk.OutboundContentRewriter
publisher *outboundimage.StableHashPublisher
handler http.Handler
// rewriteTimeout is the budget the channel gives the image rewrite stage.
rewriteTimeout time.Duration
// publicURLPrefix is the prefix of every published image URL, used by the
// DingTalk card image uploader to recognise (and refetch) our own images.
publicURLPrefix string
}
func buildOutboundImageRuntime(cfg outboundImageConfig) (outboundImageRuntime, error) {
@@ -81,9 +141,11 @@ func buildOutboundImageRuntime(cfg outboundImageConfig) (outboundImageRuntime, e
return outboundImageRuntime{}, err
}
return outboundImageRuntime{
rewriter: outboundimage.NewRewriter(fetcher, publisher, outboundImageMaxCount),
publisher: publisher,
handler: handler,
rewriter: outboundimage.NewRewriter(fetcher, publisher, outboundImageMaxCount(), outboundimage.WithImageCompression(outboundImageCompressionEnabled())),
publisher: publisher,
handler: handler,
rewriteTimeout: outboundImageRewriteTimeout(),
publicURLPrefix: strings.TrimRight(cfg.PublicBaseURL, "/") + "/",
}, nil
}
@@ -91,3 +153,49 @@ func buildOutboundImageRewriter(cfg outboundImageConfig) (dingtalk.OutboundConte
runtime, err := buildOutboundImageRuntime(cfg)
return runtime.rewriter, err
}
// buildCardImageMedia wires the DingTalk media-id rewriter used for AI card
// bodies. It returns nil when the feature is switched off or not configured, so
// a disabled uploader simply keeps external image URLs in place.
func buildCardImageMedia(publicImagePrefix, clientID, clientSecret, listenAddr string) dingtalk.CardImageMediaRewriter {
switch strings.ToLower(strings.TrimSpace(os.Getenv("DINGTALK_CARD_IMAGE_MEDIA"))) {
case "off", "0", "false", "no":
log.Printf("DingTalk card image media ids disabled by configuration")
return nil
}
resolver, err := dingtalk.NewCardImageMediaRewriter(dingtalk.CardImageMediaOptions{
ClientID: clientID,
ClientSecret: clientSecret,
ImageURLPrefix: publicImagePrefix,
FetchURLPrefix: cardImageFetchPrefix(listenAddr),
})
if err != nil {
log.Printf("DingTalk card image media ids disabled: %v", err)
return nil
}
if resolver == nil {
return nil
}
log.Printf("DingTalk card images will be delivered through DingTalk media ids")
return resolver
}
// cardImageFetchPrefix derives the prefix used to download our own published
// images before uploading them to DingTalk. An explicit
// DINGTALK_CARD_IMAGE_FETCH_BASE_URL (host only, without the image route) wins;
// otherwise the local listen port is reused so the fetch stays on loopback.
func cardImageFetchPrefix(listenAddr string) string {
base := strings.TrimSpace(os.Getenv("DINGTALK_CARD_IMAGE_FETCH_BASE_URL"))
if base == "" {
addr := strings.TrimSpace(listenAddr)
if addr == "" {
return ""
}
_, port, err := net.SplitHostPort(addr)
if err != nil || port == "" {
return ""
}
base = "http://127.0.0.1:" + port
}
return strings.TrimRight(base, "/") + outboundImageRoute
}
+118
View File
@@ -0,0 +1,118 @@
// 临时探针(不入库,用完即删):用真实 WeKnora + OSS 端点跑
// internal/outboundimage 的改写链路,量化"一条含 N 张真实图的答案"在
// 并发取图 + 预算可配下的出图结果与耗时。
package main
import (
"bufio"
"context"
"flag"
"fmt"
"log"
"net/http"
"os"
"regexp"
"strings"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/outboundimage"
)
func main() {
refsFile := flag.String("refs-file", "", "文件:每行一个 resource:// 句柄")
budget := flag.Duration("budget", 180*time.Second, "改写预算")
maxImages := flag.Int("max", 50, "图片上限")
compress := flag.Bool("compress", true, "上传前压缩截图")
passes := flag.Int("passes", 1, "重复改写轮数(第二轮应命中已存在对象)")
flag.Parse()
refs := readRefs(*refsFile)
if len(refs) == 0 {
log.Fatal("probe: no references")
}
fetcher, err := outboundimage.NewWeKnoraFetcher(
os.Getenv("WEKNORA_BASE_URL"), os.Getenv("WEKNORA_API_KEY"),
&http.Client{Timeout: 30 * time.Second}, 10<<20)
if err != nil {
log.Fatalf("probe: fetcher: %v", err)
}
store, err := outboundimage.NewAliyunOSSStore(
os.Getenv("DOBOTAI_OUTBOUND_IMAGE_OSS_ENDPOINT"),
os.Getenv("DOBOTAI_OUTBOUND_IMAGE_OSS_REGION"),
os.Getenv("DOBOTAI_OUTBOUND_IMAGE_OSS_ACCESS_KEY_ID"),
os.Getenv("DOBOTAI_OUTBOUND_IMAGE_OSS_ACCESS_KEY_SECRET"),
os.Getenv("DOBOTAI_OUTBOUND_IMAGE_OSS_BUCKET"))
if err != nil {
log.Fatalf("probe: store: %v", err)
}
publicBase := os.Getenv("DOBOTAI_OUTBOUND_IMAGE_PUBLIC_BASE_URL")
publisher, err := outboundimage.NewStableHashPublisher(store, "dobotai/im-images", publicBase)
if err != nil {
log.Fatalf("probe: publisher: %v", err)
}
rewriter := outboundimage.NewRewriter(fetcher, publisher, *maxImages, outboundimage.WithImageCompression(*compress))
var answer strings.Builder
answer.WriteString("## 探针答案\n")
for i, ref := range refs {
fmt.Fprintf(&answer, "\n![图%d](%s)\n", i+1, ref)
}
log.Printf("probe: refs=%d budget=%s max=%d compress=%v passes=%d bucket=%s", len(refs), *budget, *maxImages, *compress, *passes, os.Getenv("DOBOTAI_OUTBOUND_IMAGE_OSS_BUCKET"))
body := answer.String()
for pass := 1; pass <= *passes; pass++ {
ctx, cancel := context.WithTimeout(context.Background(), *budget)
started := time.Now()
out := rewriter.Rewrite(ctx, body)
elapsed := time.Since(started)
cancel()
published := regexp.MustCompile(`!\[图\d+\]\(https://`).FindAllString(out, -1)
leaked := strings.Count(out, "resource://")
fmt.Printf("RESULT pass=%d elapsed=%s refs=%d published=%d dropped=%d residual_resource_refs=%d out_bytes=%d\n",
pass, elapsed.Round(time.Millisecond), len(refs), len(published), len(refs)-len(published), leaked, len(out))
if len(published) > 0 {
fmt.Printf("FIRST_URL_%d=%s\nLAST_URL_%d=%s\n", pass, firstImageURL(out), pass, lastImageURL(out))
}
}
}
func readRefs(path string) []string {
if path == "" {
return nil
}
file, err := os.Open(path)
if err != nil {
log.Fatalf("probe: refs file: %v", err)
}
defer file.Close()
seen := map[string]bool{}
refs := make([]string, 0, 64)
scanner := bufio.NewScanner(file)
for scanner.Scan() {
ref := strings.TrimSpace(scanner.Text())
if !strings.HasPrefix(ref, "resource://") || seen[ref] {
continue
}
seen[ref] = true
refs = append(refs, ref)
}
return refs
}
var imageURLRE = regexp.MustCompile(`!\[图\d+\]\((https://[^)]+)\)`)
func firstImageURL(content string) string {
if match := imageURLRE.FindStringSubmatch(content); len(match) > 1 {
return match[1]
}
return ""
}
func lastImageURL(content string) string {
if matches := imageURLRE.FindAllStringSubmatch(content, -1); len(matches) > 0 {
return matches[len(matches)-1][1]
}
return ""
}
@@ -0,0 +1,2 @@
ALTER TABLE reasoning_mode_states
DROP COLUMN user_selected;
@@ -0,0 +1,6 @@
-- 000022_reasoning_default_mode: support an operator-configurable default
-- mode for the /think reasoning router. The new column records whether a
-- user has explicitly chosen a mode (/think on|off); users who never have
-- will follow DOBOTAI_DEFAULT_MODE instead of the hardcoded quick answer.
ALTER TABLE reasoning_mode_states
ADD COLUMN user_selected BOOLEAN NOT NULL DEFAULT FALSE;
@@ -0,0 +1,34 @@
package migrations_test
import (
"os"
"strings"
"testing"
)
func TestUp022AddsReasoningUserSelectedColumn(t *testing.T) {
data, err := os.ReadFile("000022_reasoning_default_mode.up.sql")
if err != nil {
t.Fatal(err)
}
source := strings.ToUpper(string(data))
for _, required := range []string{
"ALTER TABLE REASONING_MODE_STATES",
"ADD COLUMN USER_SELECTED BOOLEAN NOT NULL DEFAULT FALSE",
} {
if !strings.Contains(source, required) {
t.Errorf("up migration missing %q", required)
}
}
}
func TestDown022DropsUserSelectedColumn(t *testing.T) {
data, err := os.ReadFile("000022_reasoning_default_mode.down.sql")
if err != nil {
t.Fatal(err)
}
source := strings.ToUpper(string(data))
if !strings.Contains(source, "DROP COLUMN USER_SELECTED") {
t.Errorf("down migration missing %q", "DROP COLUMN USER_SELECTED")
}
}
+121
View File
@@ -0,0 +1,121 @@
#!/usr/bin/env python3
"""One-shot DingTalk AI card probe (test bot, private IM_ROBOT space).
Sends two cards to the owner's private robot chat so PC width can be compared:
A: body carries sys_full_json_obj = {"config":{"autoLayout":true}}
B: control, same body without the layout field
Card body mirrors internal/channels/dingtalk/card.go buildCardBody() exactly.
Credentials are read from the running container's environment and are never
printed. Only masked ids, HTTP statuses and DingTalk response bodies are shown.
"""
import json
import subprocess
import sys
import urllib.error
import urllib.request
import uuid
CONTAINER = "dobotai-dingtalk"
TOKEN_URL = "https://api.dingtalk.com/v1.0/oauth2/accessToken"
CREATE_URL = "https://api.dingtalk.com/v1.0/card/instances/createAndDeliver"
STREAM_URL = "https://api.dingtalk.com/v1.0/card/streaming"
OWNER_ID = "243143382023108067" # 姚章浩 (channel_actors.dingtalk, test tenant)
WIDE = json.dumps({"config": {"autoLayout": True}}, separators=(",", ":"))
CONTENT = (
"**{tag}**\n\n"
"这是一条 PC 宽度对照测试卡片,由测试机器人在私聊里直接推送。\n\n"
"- 投放体里的布局字段:{field}\n"
"- 期望:{expect}\n"
"- 请在 PC 端对比两张卡片的宽度差异\n\n"
"移动端请忽略,宽屏只影响桌面端渲染。"
)
def env(key: str) -> str:
out = subprocess.run(
["docker", "exec", CONTAINER, "printenv", key],
capture_output=True, text=True, check=True,
).stdout.strip()
if not out:
sys.exit(f"missing env {key}")
return out
def call(method: str, url: str, payload: dict, token: str = "") -> tuple[int, str]:
data = json.dumps(payload).encode()
req = urllib.request.Request(url, data=data, method=method)
req.add_header("Content-Type", "application/json")
if token:
req.add_header("x-acs-dingtalk-access-token", token)
try:
with urllib.request.urlopen(req, timeout=30) as resp:
return resp.status, resp.read(4096).decode("utf-8", "replace")
except urllib.error.HTTPError as e:
return e.code, e.read(4096).decode("utf-8", "replace")
def mask(v: str) -> str:
return v[:6] + "..." + v[-4:] if len(v) > 12 else "***"
def main() -> None:
app_key, app_secret = env("DINGTALK_CLIENT_ID"), env("DINGTALK_CLIENT_SECRET")
tpl = env("DINGTALK_CARD_TEMPLATE_ID")
print(f"app_key={mask(app_key)} template={mask(tpl)} target={mask(OWNER_ID)}")
status, body = call("POST", TOKEN_URL, {"appKey": app_key, "appSecret": app_secret})
token = json.loads(body).get("accessToken", "")
print(f"token: http={status} got={'yes' if token else 'NO'}")
if not token:
sys.exit("no token")
for tag, with_wide in (("A · 已带 autoLayout=true", True), ("B · 对照组(无布局字段)", False)):
param_map = {
"content": "",
"feedbackSelected": "",
"feedbackReasonSelected": "",
"showFeedbackReasons": "false",
"feedbackMessage": "",
}
if with_wide:
param_map["sys_full_json_obj"] = WIDE
out_track_id = str(uuid.uuid4())
card_body = {
"cardTemplateId": tpl,
"outTrackId": out_track_id,
"callbackType": "STREAM",
"cardData": {"cardParamMap": param_map},
"userIdType": 1,
"openSpaceId": "dtv1.card//IM_ROBOT." + OWNER_ID,
"imRobotOpenSpaceModel": {"supportForward": True},
"imRobotOpenDeliverModel": {
"robotCode": app_key,
"spaceType": "IM_ROBOT",
"extension": {},
},
}
st, rb = call("POST", CREATE_URL, card_body, token)
print(f"[{tag}] create http={st} resp={rb[:300]}")
text = CONTENT.format(
tag=tag,
field=("sys_full_json_obj = " + WIDE) if with_wide else "(无)",
expect="PC 端铺满窗口(宽屏)" if with_wide else "PC 端窄卡片(默认宽度)",
)
stream_body = {
"outTrackId": out_track_id,
"guid": str(uuid.uuid4()),
"key": "content",
"content": text,
"isFull": True,
"isFinalize": True,
"isError": False,
}
st2, rb2 = call("PUT", STREAM_URL, stream_body, token)
print(f"[{tag}] finalize http={st2} resp={rb2[:200]} outTrackId={mask(out_track_id)}")
if __name__ == "__main__":
main()
+103 -226
View File
@@ -1,280 +1,157 @@
# DobotAI DingTalk 容器部署手册
# DobotAI Docker部署(阿里云RDS)
> 适用范围:DobotAI 测试环境的容器化运行。容器和 Compose 服务名固定为 `dobotai-dingtalk`。
> 本文不授权生产部署;生产必须使用已验收的同一镜像、独立备份和独立变更审批。
## 定位
## 1. 定位
DobotAI 是钉钉渠道的对话编排与交付层,不是知识库或文档解析平台。
它负责:
- 接收钉钉 Stream 消息、富文本和图片;
- 将请求路由到 WeKnora Agent,并维护渠道会话;
- 通过钉钉 AI 卡片或 Markdown 流式交付回答;
- 处理 `/think`、`/setting` 等渠道命令;
- 保存 Turn、会话映射、覆盖判定、Support Case、反馈和审计数据;
- 对“真实检索过但没有引用”的回答创建未覆盖提醒;
- 将 `resource://` 图片按私有 OSS 规则安全外发。
它不负责:
- WeKnora 文档、Chunk、向量、图谱、解析或模型运行;
- WeKnora Agent 的知识检索策略;
- 直接暴露 PostgreSQL、WeKnora 或对象存储到公网;
- 替代钉钉应用、机器人或群权限配置。
## 2. 架构
DobotAI是钉钉渠道编排与交付服务。Docker只运行应用容器`dobotai-dingtalk`;数据库唯一使用已有的阿里云RDS PostgreSQL。
```text
钉钉用户 / 群
│ DingTalk Stream
▼
┌──────────────────────────────────────┐
│ dobotai-dingtalk 容器 │
│ - 消息解析、会话编排、卡片流式交付 │
│ - /think、/setting、首位管理员绑定 │
│ - 覆盖判定、Support Case、审计 │
└───────┬──────────────────┬───────────┘
│ │
│ HTTP Agent Chat │ PostgreSQL
▼ ▼
WeKnora 服务 DobotAI 独立数据库
- Agent/检索 - Turn / 会话 / 覆盖判定
- 文档/Chunk - Support Case / 反馈 / 设置
- 模型与解析
钉钉 Stream
↓
dobotai-dingtalk 容器
├─ 阿里云RDS PostgreSQL:Turn、会话、反馈、管理员、覆盖判定
└─ 既有WeKnora:Agent、检索、知识、模型
```
### 2.1 容器边界
| 资源 | 归属 | 是否由本容器创建/删除 |
| --- | --- | --- |
| `dobotai-dingtalk` | DobotAI应用进程 | 是,容器可重建 |
| DobotAI PostgreSQL | 独立数据服务 | 否,保留现有数据卷 |
| WeKnora | 独立知识服务 | 否 |
| `/var/lib/dobotai-test/secrets` | 宿主机密钥目录 | 仅挂载读写,不迁移、不输出 |
| `/etc/dobotai-test/dobotai.env` | 受保护运行配置 | Docker读取,脚本原子更新 |
测试环境的 PostgreSQL、WeKnora 已通过宿主机回环地址提供服务,因此 Compose 使用 `network_mode: host`。这不是新开公网端口:DobotAI仍按 `DOBOTAI_LISTEN_ADDR` 监听,默认是 `127.0.0.1:8090`。
### 2.2 容器安全属性
Compose 固定启用:
禁止创建或使用:
```text
容器名:dobotai-dingtalk
网络:host(仅兼容现有测试宿主机回环依赖)
运行身份:与原 dobotai-test 服务用户相同的 UID:GID
根文件系统:只读
临时写入:仅 /tmp tmpfs
额外权限:no-new-privileges
重启策略:unless-stopped
dobotai-postgres 容器
dobotai-postgres-data 数据卷
本地PostgreSQL作为DobotAI目标数据库
```
对应的关键 Compose 配置如下;不要为避免端口冲突额外添加 `ports:` 映射:
```yaml
services:
dobotai-dingtalk:
container_name: dobotai-dingtalk
network_mode: host
read_only: true
restart: unless-stopped
```
## 3. 安装前检查
安装操作人员需要具备测试服务器的 `sudo` 和 Docker 权限。开始前确认:
## 架构与目录
```text
1. 已构建或导入经审查的 DobotAI 镜像;
2. 现有 /etc/dobotai-test/dobotai.env 存在且权限为 0600;
3. 现有密钥目录 /var/lib/dobotai-test/secrets 存在;
4. DobotAI PostgreSQL 与 WeKnora 已健康;
5. 已记录当前原生服务二进制SHA和运行状态;
6. 如本版本包含 SQL 迁移,已先完成逻辑备份与迁移演练。
宿主机
├── dobotai-dingtalk 唯一应用容器,host网络
├── /opt/dobotai/container/ Compose、RDS检查器、SQL迁移、启动脚本
├── /opt/dobotai/config/dobotai.env 0600,RDS/WeKnora/钉钉配置
├── /opt/dobotai/secrets/ 0700,DobotAI SecretStore
└── /opt/dobotai/backups/ 发布前RDS备份记录,不保存Docker数据库卷
阿里云RDS
└── DobotAI专用PostgreSQL数据库
```
不要执行:
应用使用`network_mode: host`,因此通过RDS私网地址访问云数据库、通过`127.0.0.1:8080`访问宿主机现有WeKnora。不会开放新的数据库端口;应用HTTP仍由`DOBOTAI_LISTEN_ADDR=127.0.0.1:8090`约束。
```text
docker compose down -v
docker volume rm / prune
删除 PostgreSQL / WeKnora 数据卷
将密钥、数据库连接串或 API Key 写入命令行、日志或 Git
```
## 安装
## 4. 构建和导入镜像
在已审查且工作树干净的源码目录构建。示例中的值均为占位符:
克隆完整仓库后,只执行:
```bash
docker build \
--build-arg VERSION=test-<commit-short> \
--build-arg COMMIT_SHA=<完整提交SHA> \
--build-arg BUILD_TIME=<UTC时间> \
-t dobotai-dingtalk:test-<commit-short> .
sudo bash ./install-docker.sh
```
跨主机传输时先保存镜像归档并记录 SHA-256;测试服务器导入后必须复核归档哈希和镜像标签。生产不得在目标机重新编译,应使用已测试的同一镜像归档或不可变镜像摘要。
## 5. 安装容器资产(不切换服务)
将本目录、候选镜像和已核验的归档放到测试服务器的受控发布目录。确认镜像已经 `docker load` 后执行:
```bash
sudo DOBOTAI_IMAGE=dobotai-dingtalk:test-<commit-short> \
bash ./install-test.sh
```
该脚本会:
根脚本自动:
```text
1. 检查候选镜像、现有 env 与密钥目录;
2. 备份原生 DobotAI 二进制和 systemd Unit,并写入 SHA-256;
3. 安装 Compose、配置器、启动器和绑定脚本;
4. 生成权限 0600 的 /opt/dobotai-test/container/runtime.env;
5. 保持原生 dobotai-test.service 继续运行。
检查或安装Docker Engine和Docker Compose v2(APT系统)
以当前Git提交构建 dobotai-dingtalk:local-<Git短SHA>
创建 /opt/dobotai/{container,config,secrets,backups}
复制应用Compose、RDS检查器和21个SQL迁移
生成权限0600的应用环境模板
安装三条短命令
```
它不会读取或打印环境变量值、不会修改数据库、不会启动容器。
它不会连接、创建或修改RDS,也不会启动未配置的钉钉应用。
## 6. 配置并启动
## 配置、RDS检查与启动
日常只需要执行:
安装后执行:
```bash
sudo dobotai-dingtalk-configure
```
该命令交互更新 `/etc/dobotai-test/dobotai.env`。每项都会显示用途和脱敏格式示例;已配置的值永不回显,直接回车表示保留。
配置器分字段录入阿里云RDS(主机、端口、数据库名、账号、密码),自动组装连接串;WeKnora和钉钉凭证密钥不回显,环境文件原子写入且权限0600。
| 环境变量 | 用途 | 格式示例 |
`DOBOTAI_TENANT_KEY`不是WeKnora Agent租户键。它是DobotAI内部数据隔离键,首次安装时自动生成并固定保存;配置器不会要求填写或改动它。WeKnora侧只需要填写实际的两个Agent ID。
配置器会要求两个不同的WeKnora Agent ID,并自动生成`DOBOTAI_AGENT_BINDINGS={"dingtalk:*":"快速问答Agent ID"}`。`DOBOTAI_INGRESS_TOKEN`是保护本机`POST /v1/qa`内部接口的Bearer Token,不用于钉钉或WeKnora;同样由安装器自动生成。
| 配置 | 示例 | 作用 |
| --- | --- | --- |
| `DATABASE_URL` | DobotAI独立PostgreSQL连接串,不得填写WeKnora数据库 | `postgres://USER:<数据库密码>@127.0.0.1:55432/dobotai_test?sslmode=disable` |
| `WEKNORA_BASE_URL` | 测试WeKnora地址 | `http://127.0.0.1:8080` |
| `WEKNORA_API_KEY` | DobotAI使用的WeKnora API密钥 | `<你的WeKnora API Key>` |
| `DOBOTAI_INGRESS_TOKEN` | `/v1/qa`调用令牌 | `<随机长字符串>` |
| `DOBOTAI_TENANT_KEY` | 与既有DobotAI数据一致的租户键 | `dobotai-test` |
| `DOBOTAI_AGENT_BINDINGS` | 钉钉会话到WeKnora Agent的映射 | `{"dingtalk:*":"你的Agent ID"}` |
| `DINGTALK_CLIENT_ID` | 钉钉应用Client ID/AppKey | `dingxxxxxxxx` |
| `DINGTALK_CLIENT_SECRET` | 钉钉应用Client Secret/AppSecret | `<你的钉钉 AppSecret>` |
| `DATABASE_URL` | `postgres://USER:***@rds-endpoint:5432/dobotai?sslmode=require` | DobotAI专用阿里云RDS PostgreSQL |
| `WEKNORA_BASE_URL` | `http://127.0.0.1:8080` | 宿主机既有WeKnora |
| `WEKNORA_API_KEY` | `<你的WeKnora API Key>` | DobotAI专用Key |
| `DOBOTAI_QUICK_AGENT_ID` | `<快速问答Agent ID>` | 默认快速问答实际调用的Agent |
| `DOBOTAI_REASONING_AGENT_ID` | `<智能推理Agent ID>` | `/think`和持续智能推理实际调用的Agent |
| `DINGTALK_CLIENT_ID` | `dingxxxxxxxx` | 钉钉AppKey |
| `DINGTALK_CLIENT_SECRET` | `<你的钉钉 AppSecret>` | 钉钉AppSecret |
保存成功后,配置器自动调用启动器:
回答图片外发OSS同为分字段录入的可选整组(紧随RDS之后、WeKnora/钉钉凭证之前):
| 配置 | 示例 | 作用 |
| --- | --- | --- |
| `DOBOTAI_OUTBOUND_IMAGE_OSS_ENDPOINT` | `https://oss-cn-shenzhen.aliyuncs.com` | 私有Bucket接入地址 |
| `DOBOTAI_OUTBOUND_IMAGE_OSS_REGION` | `cn-shenzhen` | Bucket所在地域 |
| `DOBOTAI_OUTBOUND_IMAGE_OSS_BUCKET` | `dobot-xxx` | 外发图片专用私有Bucket |
| `DOBOTAI_OUTBOUND_IMAGE_OSS_ACCESS_KEY_ID` | `<AK ID>` | 仅授权该Bucket的AK |
| `DOBOTAI_OUTBOUND_IMAGE_OSS_ACCESS_KEY_SECRET` | `<AK Secret>` | 仅授权该Bucket的SK(不回显) |
| `DOBOTAI_OUTBOUND_IMAGE_PUBLIC_BASE_URL` | `https://host/dingtalk/i` | 钉钉可见的稳定HTTPS入口,不含凭据 |
OSS规则:六项全部留空即整组禁用图片外发(答案含`resource://`图片时按无图降级);填写任一项则六项必填,缺项拒绝保存。运行时仅改写WeKnora `resource://`引用为短签HTTPS,私有Bucket由稳定基础地址代理,`<img>`外链与`data:`剥除。禁止把OSS凭据写入Git或命令行。
保存后脚本自动执行以下RDS门禁:
```text
停止原生 dobotai-test.service
→ 启动 dobotai-dingtalk
→ 检查 http://127.0.0.1:8090/healthz
→ 健康成功后禁用原生服务
1. 使用RDS连接串执行 pg_isready
2. 检查DobotAI核心表
3. 数据库没有任何public表:按000001~000021初始化
4. 所有核心表存在:判定正常,不执行任何DDL
5. 部分表、缺表或非空异常库:拒绝初始化和启动
6. RDS通过后才启动 dobotai-dingtalk 并检查 /healthz
```
首次容器启动失败会恢复原生服务;容器版本升级失败会恢复上一容器镜像。
这保证不会把已有RDS表误当成空库重建,也不会把半成品或错误库静默升级。
## 7. 首位管理员绑定
## 日常命令
仅当该租户不存在有效管理员时执行:
```bash
# 重新检查RDS并启动/升级应用
sudo dobotai-dingtalk-start
# 交互更新RDS、WeKnora、钉钉配置后启动
sudo dobotai-dingtalk-configure
# 仅首位管理员且无有效管理员时使用
sudo dobotai-dingtalk-bind-admin
# 健康与版本
curl --noproxy '*' -fsS http://127.0.0.1:8090/healthz
curl --noproxy '*' -fsS http://127.0.0.1:8090/version
```
## 升级与回滚
应用升级仅替换`dobotai-dingtalk`镜像。启动前再次检查RDS;健康失败时恢复上一应用镜像(如存在)。RDS不是容器资产,不能被`docker compose down`、镜像替换或本机卷操作影响。
RDS有实际数据时,数据库备份、SQL变更与恢复应在阿里云RDS发布流程中完成并验证;不要把应用镜像回滚误当成数据库回滚。
## 首位管理员
应用和钉钉Stream运行后:
```bash
sudo dobotai-dingtalk-bind-admin
```
终端会显示一个10分钟有效的随机六码。候选管理员必须私聊钉钉机器人发送:
终端会显示10分钟有效的六码;候选管理员私聊机器人:
```text
/setting <六码绑定码>
```
验证成功后,终端自动写入 `super_admin/active` 并退出。已有有效管理员时命令拒绝执行,防止借由该入口追加管理员或意外提权。
已有有效管理员时绑定命令会拒绝,避免提权。
## 8. 覆盖判定与未覆盖提醒
覆盖判定基于真实的 `knowledge_searched` 工具事件和结构化引用,不按关键词猜测。
| 场景 | 覆盖结果 | 是否创建 Support Case |
| --- | --- | --- |
| 有引用 | `covered_has_references` | 否 |
| 实际检索过但无引用 | `uncovered_no_references` | 是 |
| 明确无权威知识/无答案 | `uncovered_explicit_no_answer` | 是 |
| 图片未检索、纯视觉识别 | `covered_visual_recognition` | 否 |
| 文本未检索、闲聊或普通非知识响应 | `covered_non_knowledge_response` | 否 |
因此“在吗”“你好”“谢谢”等未触发知识检索的文本不会产生未覆盖提醒;真正检索过但没有权威引用的问题仍会进入人工支持流程。
## 9. 数据库迁移门禁
DobotAI进程只连接数据库,**不会自动发现或执行 SQL 迁移**。包含新迁移的版本必须在容器替换前由发布流程显式完成:
## 不执行的操作
```text
1. 对 DobotAI PostgreSQL 做逻辑备份;
2. 使用 pg_restore --list 验证备份非空;
3. 在事务中执行已审查的 .up.sql;
4. 向 dobotai_schema_migrations 写入对应 version/name/applied_at;
5. 查询约束、列或索引确认生效;
6. 再替换容器镜像并进行健康与业务验收。
不创建本地PostgreSQL容器或数据卷
不迁移、删除或覆盖现有本地 dobotai-test-postgres
不创建、删除或重置阿里云RDS数据库
不把RDS连接串、Token、钉钉凭据写入Git、命令行或日志
不重建WeKnora、Neo4j、Redis、Elasticsearch
```
`000021_coverage_non_knowledge_response_reason` 是一个示例:它只扩展 `coverage_decisions.reason_code` 的 CHECK 约束,不回写历史Turn、覆盖判定或Support Case。
## 10. 日常检查
```bash
# 健康检查
curl --noproxy '*' -fsS http://127.0.0.1:8090/healthz
# 运行版本
curl --noproxy '*' -fsS http://127.0.0.1:8090/version
# 容器状态与重启次数
sudo docker inspect -f '{{.Config.Image}}|{{.State.Running}}|{{.RestartCount}}' dobotai-dingtalk
# 最近日志(不要复制含敏感内容的原始上下文)
sudo docker logs --since 10m dobotai-dingtalk
```
验收至少应同时满足:
```text
容器 running,重启次数为0;
/healthz 返回 ok;
/version 的 commit_sha 与发布记录一致;
钉钉 Stream 仅启动一个消费者;
旧 dobotai-test.service 为 inactive + disabled;
目标业务路径已由真实钉钉消息验证。
```
## 11. 升级与回滚
### 升级
```text
构建/导入经审查的新镜像
→ 备份DobotAI PostgreSQL
→ 应用并记录必要迁移
→ 将 runtime.env 中的 DOBOTAI_IMAGE 更新为新标签
→ 执行 sudo dobotai-dingtalk-start
→ 健康、版本、日志、钉钉真实消息验收
```
注意拼写:数据库和应用名称是 `DobotAI`;上述流程中的镜像名为 `dobotai-dingtalk`。
### 自动回滚
启动器会保存替换前的容器镜像:
```text
首次容器化失败:停止候选容器,恢复原生服务;
后续容器升级失败:停止候选容器,恢复上一容器镜像。
```
数据库迁移不能仅靠容器回滚撤销。只有经过审查且已验证的 `.down.sql` 才可用于数据库回滚;任何数据恢复都必须先确认备份可用。
## 12. 明确非目标
```text
不容器化或重建 WeKnora;
不容器化或删除 DobotAI PostgreSQL;
不重新启用已删除的网页管理面;
不在生产环境复用测试凭据、测试镜像或未验收的迁移;
不把密钥、Token、连接串、钉钉凭据写入 Git、命令行或日志。
```
+101 -63
View File
@@ -1,83 +1,121 @@
#!/usr/bin/env python3
import argparse
import getpass
import json
import os
from pathlib import Path
from urllib.parse import quote, unquote, urlsplit
import subprocess
import sys
FIELDS = (
("DATABASE_URL", "DobotAI PostgreSQL 连接串", "DobotAI独立数据库,不得填写WeKnora数据库。", "postgres://USER:<数据库密码>@127.0.0.1:55432/dobotai_test?sslmode=disable", True),
("WEKNORA_BASE_URL", "WeKnora 地址", "容器通过宿主网络访问测试WeKnora。", "http://127.0.0.1:8080", False),
("WEKNORA_API_KEY", "WeKnora API Key", "在WeKnora中为DobotAI创建的API密钥。", "<你的WeKnora API Key>", True),
("DOBOTAI_INGRESS_TOKEN", "DobotAI Ingress Token", "调用 /v1/qa 时使用的独立长随机令牌。", "<随机长字符串>", True),
("DOBOTAI_TENANT_KEY", "DobotAI 租户键", "必须与现有DobotAI数据和Agent绑定使用同一个租户键。", "dobotai-test", False),
("DOBOTAI_AGENT_BINDINGS", "DobotAI Agent Bindings JSON", "钉钉会话匹配到WeKnora Agent的映射。", '{"dingtalk:*":"你的Agent ID"}', False),
("DINGTALK_CLIENT_ID", "钉钉 Client ID", "钉钉应用的Client ID/AppKey。", "dingxxxxxxxx", False),
("DINGTALK_CLIENT_SECRET", "钉钉 Client Secret", "钉钉应用的Client Secret/AppSecret。", "<你的钉钉 AppSecret>", True),
("WEKNORA_BASE_URL", "WeKnora 地址", "应用容器通过宿主机网络访问既有WeKnora。", "http://127.0.0.1:8080", False),
("WEKNORA_API_KEY", "WeKnora API Key", "DobotAI专用WeKnora API Key。", "<你的WeKnora API Key>", True),
("DOBOTAI_QUICK_AGENT_ID", "快速问答Agent ID", "默认问答实际调用的WeKnora Agent。", "<快速问答Agent ID>", False),
("DOBOTAI_REASONING_AGENT_ID", "智能推理Agent ID", "/think和持续智能推理实际调用的WeKnora Agent。", "<智能推理Agent ID>", False),
("DINGTALK_CLIENT_ID", "钉钉 Client ID", "钉钉应用Client ID/AppKey。", "dingxxxxxxxx", False),
("DINGTALK_CLIENT_SECRET", "钉钉 Client Secret", "钉钉应用Client Secret/AppSecret。", "<你的钉钉 AppSecret>", True),
)
def prompt_rds(values):
"""分字段录入阿里云RDS并组装DATABASE_URL;已有值作为默认。"""
existing = values.get("DATABASE_URL", "").strip()
host = port = database = user = password = ""
if existing:
u = urlsplit(existing)
host = u.hostname or ""
port = str(u.port or "") if u.port else ""
database = u.path.lstrip("/")
user = unquote(u.username or "")
password = unquote(u.password or "")
def parse_env(path: Path):
lines = path.read_text(encoding="utf-8").splitlines()
values = {}
def ask(key, label, default, secret=False):
suffix = "(已配置,直接回车保留)" if default else "(必填)"
reader = getpass.getpass if secret else input
entered = reader(f"{label}{suffix}: ").strip()
return entered or default
print("\n阿里云RDS PostgreSQL(分字段填写)\n用途:DobotAI专用数据库;空库自动初始化,已有表只校验不改结构。")
host = ask("host", "RDS主机地址", host)
port = ask("port", "RDS端口", port or "5432")
database = ask("database", "数据库名", database)
user = ask("user", "数据库账号", user)
password = ask("password", "数据库密码", password, secret=True)
if not all((host, port, database, user, password)):
raise SystemExit("RDS字段不完整,未保存。")
if not database.startswith(("dobotai", "dobotai_")):
print("警告:数据库名不是DobotAI专用库,继续前请确认。")
values["DATABASE_URL"] = f"postgres://{quote(user, safe='')}:{quote(password, safe='')}@{host}:{port}/{database}"
OSS_FIELDS = (
("DOBOTAI_OUTBOUND_IMAGE_OSS_ENDPOINT", "OSS Endpoint", "私有Bucket的接入地址。", "https://oss-cn-shenzhen.aliyuncs.com", False),
("DOBOTAI_OUTBOUND_IMAGE_OSS_REGION", "OSS Region", "Bucket所在地域。", "cn-shenzhen", False),
("DOBOTAI_OUTBOUND_IMAGE_OSS_BUCKET", "OSS Bucket", "外发图片专用私有Bucket。", "dobot-xxx", False),
("DOBOTAI_OUTBOUND_IMAGE_OSS_ACCESS_KEY_ID", "OSS AccessKey ID", "仅授权该Bucket的AK。", "<AK ID>", False),
("DOBOTAI_OUTBOUND_IMAGE_OSS_ACCESS_KEY_SECRET", "OSS AccessKey Secret", "仅授权该Bucket的SK。", "<AK Secret>", True),
("DOBOTAI_OUTBOUND_IMAGE_PUBLIC_BASE_URL", "图片稳定外发基础地址", "钉钉可见的稳定HTTPS入口,不含凭据。", "https://host/dingtalk/i", False),
)
def prompt_oss(values):
"""OSS整组录入:全部留空即禁用;任一项填写则六项必填。"""
print("\n回答图片外发OSS(可选,OSS整组录入;直接回车表示整组禁用)")
entered = {}
for key, label, description, example, secret in OSS_FIELDS:
default = values.get(key, "").strip()
suffix = "(已配置,直接回车保留)" if default else ""
print(f"用途:{description} 格式示例:{example}")
reader = getpass.getpass if secret else input
value = reader(f"{label}{suffix}: ").strip()
entered[key] = value or default
if not any(entered.values()):
for key, *_ in OSS_FIELDS:
values.pop(key, None)
return
missing = [label for key, label, *_ in OSS_FIELDS if not entered[key]]
if missing:
raise SystemExit("OSS为整组可选能力,以下项未填写:" + "、".join(missing))
values.update(entered)
def parse_env(path):
lines = path.read_text(encoding="utf-8").splitlines(); values = {}
for line in lines:
if line and not line.lstrip().startswith("#") and "=" in line:
key, value = line.split("=", 1)
values[key.strip()] = value
key, value = line.split("=", 1); values[key.strip()] = value
return lines, values
def prompt(label: str, description: str, example: str, secret: bool, configured: bool) -> str:
suffix = "(已配置,直接回车保留)" if configured else "(必填)"
print(f"\n{label}\n用途:{description}\n格式示例:{example}")
if secret:
return getpass.getpass(f"{label}{suffix}: ").strip()
return input(f"{label}{suffix}: ").strip()
def write_env(path: Path, lines, values):
rendered = []
written = set()
def write_env(path, lines, values):
rendered=[]; written=set()
for line in lines:
if line and not line.lstrip().startswith("#") and "=" in line:
key = line.split("=", 1)[0].strip()
if key in values:
rendered.append(f"{key}={values[key]}")
written.add(key)
continue
key=line.split("=",1)[0].strip()
if key in values: rendered.append(f"{key}={values[key]}"); written.add(key); continue
rendered.append(line)
for key, value in values.items():
if key not in written:
rendered.append(f"{key}={value}")
temporary = path.with_name(path.name + ".new")
temporary.write_text("\n".join(rendered) + "\n", encoding="utf-8")
os.chmod(temporary,0o600)
os.replace(temporary, path)
rendered.extend(f"{key}={value}" for key,value in values.items() if key not in written)
temp=path.with_name(path.name+".new"); temp.write_text("\n".join(rendered)+"\n",encoding="utf-8"); os.chmod(temp,0o600); os.replace(temp,path)
def main():
parser = argparse.ArgumentParser(description="配置 DobotAI 容器运行环境")
parser.add_argument("--env-file", required=True)
parser.add_argument("--start-script", required=True)
args = parser.parse_args()
env_file = Path(args.env_file)
if not env_file.is_file():
raise SystemExit("DobotAI 环境文件不存在,请先完成部署。")
lines, values = parse_env(env_file)
for key, label, description, example, secret in FIELDS:
entered = prompt(label, description, example, secret, bool(values.get(key, "").strip()))
if entered:
values[key] = entered
missing = [label for key, label, _, _, _ in FIELDS if not values.get(key, "").strip()]
if missing:
raise SystemExit("以下必填项未配置:" + "、".join(missing))
write_env(env_file, lines, values)
os.execv(args.start_script, [args.start_script])
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
sys.exit(130)
parser=argparse.ArgumentParser(description="配置DobotAI阿里云RDS和渠道凭证")
parser.add_argument("--env-file",required=True); parser.add_argument("--schema-script",required=True); parser.add_argument("--start-script",required=True)
args=parser.parse_args(); path=Path(args.env_file)
if not path.is_file(): raise SystemExit("DobotAI环境文件不存在,请先执行根目录install-docker.sh。")
lines,values=parse_env(path)
prompt_rds(values)
prompt_oss(values)
for key,label,description,example,secret in FIELDS:
suffix="(已配置,直接回车保留)" if values.get(key,"").strip() else "(必填)"
print(f"\n{label}\n用途:{description}\n格式示例:{example}")
entered=(getpass.getpass if secret else input)(f"{label}{suffix}: ").strip()
if entered: values[key]=entered
missing=[label for key,label,*_ in FIELDS if not values.get(key,"").strip()]
if missing: raise SystemExit("以下必填项未配置:"+"、".join(missing))
if values["DOBOTAI_QUICK_AGENT_ID"] == values["DOBOTAI_REASONING_AGENT_ID"]:
raise SystemExit("快速问答Agent ID与智能推理Agent ID必须不同。")
values["DOBOTAI_AGENT_BINDINGS"] = json.dumps({"dingtalk:*": values["DOBOTAI_QUICK_AGENT_ID"]}, separators=(",", ":"))
write_env(path,lines,values)
subprocess.run([args.schema_script], check=True)
os.execv(args.start_script,[args.start_script])
if __name__=="__main__":
try: main()
except KeyboardInterrupt: sys.exit(130)
@@ -1,10 +1,6 @@
#!/usr/bin/env bash
set -euo pipefail
container="dobotai-dingtalk"
if [[ "$(docker inspect --format '{{.State.Running}}' "$container" 2>/dev/null || true)" != "true" ]]; then
echo "DobotAI 容器未运行。请先执行:sudo dobotai-dingtalk-configure" >&2
exit 1
fi
home="${DOBOTAI_HOME:-/opt/dobotai}"
container_dir="$home/container"
if [[ "$(docker inspect --format '{{.State.Running}}' dobotai-dingtalk 2>/dev/null || true)" != "true" ]]; then echo "DobotAI容器未运行。请先执行:sudo dobotai-dingtalk-configure" >&2; exit 1; fi
exec docker exec -it dobotai-dingtalk /usr/local/bin/dobotai bootstrap-admin
+11 -65
View File
@@ -1,77 +1,23 @@
#!/usr/bin/env bash
set -euo pipefail
container_dir="/opt/dobotai-test/container"
home="${DOBOTAI_HOME:-/opt/dobotai}"
container_dir="$home/container"
compose_file="$container_dir/compose.yaml"
runtime_env="$container_dir/runtime.env"
legacy_service="dobotai-test.service"
health_url="http://127.0.0.1:8090/healthz"
if [[ ! -r "$compose_file" || ! -r "$runtime_env" ]]; then
echo "容器运行文件未安装。请先部署 dobotai-dingtalk。" >&2
exit 1
fi
legacy_was_active=false
if systemctl is-active --quiet "$legacy_service"; then
legacy_was_active=true
systemctl stop "$legacy_service"
fi
previous_image="$(docker inspect --format '{{.Config.Image}}' dobotai-dingtalk 2>/dev/null || true)"
set_runtime_image() {
local image="$1"
python3 - "$runtime_env" "$image" <<'PY'
from pathlib import Path
import os
import sys
path = Path(sys.argv[1])
image = sys.argv[2]
lines = path.read_text(encoding="utf-8").splitlines()
updated = False
rendered = []
for line in lines:
if line.startswith("DOBOTAI_IMAGE="):
rendered.append("DOBOTAI_IMAGE=" + image)
updated = True
else:
rendered.append(line)
if not updated:
rendered.append("DOBOTAI_IMAGE=" + image)
temporary = path.with_name(path.name + ".new")
temporary.write_text("\n".join(rendered) + "\n", encoding="utf-8")
os.chmod(temporary, 0o600)
os.replace(temporary, path)
PY
}
[[ -r "$compose_file" && -r "$runtime_env" ]] || { echo "容器资产未安装。" >&2; exit 1; }
"$container_dir/rds-schema-init.sh"
compose=(docker compose --env-file "$runtime_env" -f "$compose_file")
previous_image="$(docker inspect -f '{{.Config.Image}}' dobotai-dingtalk 2>/dev/null || true)"
rollback() {
docker compose --env-file "$runtime_env" -f "$compose_file" down --remove-orphans >/dev/null 2>&1 || true
if [[ -n "$previous_image" ]]; then
set_runtime_image "$previous_image"
docker compose --env-file "$runtime_env" -f "$compose_file" up -d --no-build --force-recreate >/dev/null 2>&1 || true
elif [[ "$legacy_was_active" == "true" ]]; then
systemctl start "$legacy_service" || true
fi
"${compose[@]}" rm -sf dobotai-dingtalk >/dev/null 2>&1 || true
if [[ -n "$previous_image" ]]; then "${compose[@]}" up -d --no-build --force-recreate dobotai-dingtalk >/dev/null 2>&1 || true; fi
}
trap rollback ERR
docker compose --env-file "$runtime_env" -f "$compose_file" up -d --no-build --force-recreate
"${compose[@]}" up -d --no-build --force-recreate dobotai-dingtalk
for _ in $(seq 1 30); do
if curl --noproxy '*' --fail --silent --max-time 2 "$health_url" >/dev/null; then
systemctl disable "$legacy_service" >/dev/null
trap - ERR
echo "DobotAI 容器已启动:dobotai-dingtalk"
exit 0
fi
if curl --noproxy '*' --fail --silent --max-time 2 "$health_url" >/dev/null; then trap - ERR; echo "DobotAI容器已启动:dobotai-dingtalk"; exit 0; fi
sleep 1
done
if [[ -n "$previous_image" ]]; then
echo "DobotAI 容器健康检查失败,已回滚到上一容器镜像。" >&2
else
echo "DobotAI 容器健康检查失败,已回滚到原生服务。" >&2
fi
echo "DobotAI健康检查失败,已恢复上一应用容器(如存在)。" >&2
exit 1
+76
View File
@@ -0,0 +1,76 @@
#!/usr/bin/env bash
set -euo pipefail
script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
source_root="${DOBOTAI_SOURCE_ROOT:-$(cd "$script_dir/../.." && pwd)}"
home="${DOBOTAI_HOME:-/opt/dobotai}"
container_dir="$home/container"
config_dir="$home/config"
secret_dir="$home/secrets"
backup_dir="$home/backups"
app_env="$config_dir/dobotai.env"
runtime_env="$container_dir/runtime.env"
image="${DOBOTAI_IMAGE:?root install-docker.sh must provide the local image}"
bin_dir="${DOBOTAI_BIN_DIR:-/usr/local/sbin}"
[[ ${EUID} -eq 0 ]] || { echo "请使用sudo执行安装脚本。" >&2; exit 1; }
docker image inspect "$image" >/dev/null 2>&1 || { echo "应用镜像不存在。" >&2; exit 1; }
install -d -m 0750 "$home" "$container_dir" "$config_dir" "$backup_dir" "$container_dir/migrations"
install -d -m 0700 -o 10001 -g 10001 "$secret_dir"
install -m 0644 "$script_dir/compose.yaml" "$container_dir/compose.yaml"
install -m 0700 "$script_dir/configure_env.py" "$container_dir/configure_env.py"
install -m 0750 "$script_dir/rds-schema-init.sh" "$container_dir/rds-schema-init.sh"
install -m 0750 "$script_dir/dobotai-dingtalk-start" "$container_dir/dobotai-dingtalk-start"
install -m 0750 "$script_dir/dobotai-dingtalk-bind-admin" "$container_dir/dobotai-dingtalk-bind-admin"
install -m 0644 "$source_root"/db/migrations/*.up.sql "$container_dir/migrations/"
if [[ ! -f "$app_env" ]]; then
umask 077
read -r tenant_suffix ingress_token < <(python3 -c 'import secrets; print(secrets.token_hex(12), secrets.token_urlsafe(32))')
cat > "$app_env" <<'EOF'
DOBOTAI_LISTEN_ADDR=127.0.0.1:8090
DATABASE_URL=
WEKNORA_BASE_URL=http://127.0.0.1:8080
WEKNORA_API_KEY=
DOBOTAI_ADMIN_TOKEN=
DOBOTAI_AGENT_BINDINGS=
DOBOTAI_QUICK_AGENT_ID=
DOBOTAI_REASONING_AGENT_ID=
DINGTALK_CLIENT_ID=
DINGTALK_CLIENT_SECRET=
EOF
printf 'DOBOTAI_TENANT_KEY=dobotai-%s\n' "$tenant_suffix" >> "$app_env"
printf 'DOBOTAI_INGRESS_TOKEN=%s\n' "$ingress_token" >> "$app_env"
chmod 0600 "$app_env"
fi
umask 077
cat > "$runtime_env" <<EOF
DOBOTAI_IMAGE=$image
DOBOTAI_RUNTIME_UID=10001
DOBOTAI_RUNTIME_GID=10001
DOBOTAI_ENV_FILE=$app_env
DOBOTAI_SECRET_DIR_HOST=$secret_dir
EOF
chmod 0600 "$runtime_env"
install -d -m 0755 "$bin_dir"
for command in configure start bind-admin; do
target="$bin_dir/dobotai-dingtalk-$command"
cat > "$target" <<EOF
#!/usr/bin/env bash
set -euo pipefail
export DOBOTAI_HOME='$home'
exec '$container_dir/dobotai-dingtalk-$command'
EOF
chmod 0750 "$target"
done
# configure is a Python entrypoint rather than a shell asset.
cat > "$bin_dir/dobotai-dingtalk-configure" <<EOF
#!/usr/bin/env bash
set -euo pipefail
export DOBOTAI_HOME='$home'
exec python3 '$container_dir/configure_env.py' --env-file '$app_env' --schema-script '$container_dir/rds-schema-init.sh' --start-script '$bin_dir/dobotai-dingtalk-start'
EOF
chmod 0750 "$bin_dir/dobotai-dingtalk-configure"
echo "应用容器资产已安装;尚未连接或修改阿里云RDS。"
echo "下一步执行:sudo dobotai-dingtalk-configure"
@@ -0,0 +1,42 @@
#!/usr/bin/env bash
set -euo pipefail
home="${DOBOTAI_HOME:-/opt/dobotai}"
app_env="$home/config/dobotai.env"
migrations_dir="$home/container/migrations"
[[ -r "$app_env" && -d "$migrations_dir" ]] || { echo "RDS检查资产未安装。" >&2; exit 1; }
temp_env="$(mktemp)"; trap 'rm -f "$temp_env"' EXIT
python3 - "$app_env" "$temp_env" <<'PY'
from pathlib import Path
import os,sys
values={}
for line in Path(sys.argv[1]).read_text(encoding='utf-8').splitlines():
if line and not line.startswith('#') and '=' in line:
key,value=line.split('=',1); values[key]=value
url=values.get('DATABASE_URL','').strip()
if not (url.startswith('postgres://') or url.startswith('postgresql://')): raise SystemExit('DATABASE_URL必须是阿里云RDS PostgreSQL连接串。')
Path(sys.argv[2]).write_text('DATABASE_URL='+url+'\n',encoding='utf-8'); os.chmod(sys.argv[2],0o600)
PY
rds_shell() { docker run --rm -i --network host --env-file "$temp_env" postgres:16-alpine sh -ec "$1"; }
rds_shell 'pg_isready -d "$DATABASE_URL" >/dev/null' || true
if ! probe_error="$(rds_shell 'exec psql -At -v ON_ERROR_STOP=1 "$DATABASE_URL" -c "SELECT 1"' 2>&1)"; then
case "$probe_error" in
*"does not exist"*|*"不存在"*)
echo "阿里云RDS目标数据库不存在;请RDS管理员创建DobotAI专用数据库并授予应用账号连接和建表权限。" >&2
;;
*)
echo "无法连接阿里云RDS目标数据库;请管理员检查网络、白名单、TLS、账号权限和连接串。" >&2
;;
esac
exit 1
fi
sql() { rds_shell 'exec psql -v ON_ERROR_STOP=1 "$DATABASE_URL"'; }
sql_at() { rds_shell 'exec psql -At -v ON_ERROR_STOP=1 "$DATABASE_URL"'; }
state=$(printf "%s\n" "WITH expected(name) AS (VALUES ('turns'),('session_mappings'),('messages'),('coverage_decisions'),('support_cases'),('system_config_revisions'),('administrators'),('setting_sessions'),('user_preferences'),('reasoning_mode_states'),('reasoning_mode_turns')) SELECT CASE WHEN to_regclass('public.turns') IS NULL AND (SELECT COUNT(*) FROM information_schema.tables WHERE table_schema='public')=0 THEN 'empty' WHEN (SELECT COUNT(*) FROM expected e JOIN information_schema.tables t ON t.table_schema='public' AND t.table_name=e.name)=11 THEN 'normal' ELSE 'inconsistent' END;" | sql_at)
case "$state" in
normal) echo "阿里云RDS表结构检查通过;不执行初始化。"; exit 0;;
inconsistent) echo "阿里云RDS表结构不完整或不是空库,拒绝初始化和启动。" >&2; exit 1;;
empty) echo "阿里云RDS为空,开始初始化DobotAI表。";;
*) echo "无法判定阿里云RDS表状态。" >&2; exit 1;;
esac
for migration in "$migrations_dir"/*.up.sql; do printf 'BEGIN;\n'; cat "$migration"; printf '\nCOMMIT;\n'; done | sql
echo "阿里云RDS初始化完成。"
+5
View File
@@ -0,0 +1,5 @@
\pset pager off
SELECT table_schema || '.' || table_name AS t
FROM information_schema.tables
WHERE table_schema NOT IN ('pg_catalog', 'information_schema')
ORDER BY 1;
+8
View File
@@ -0,0 +1,8 @@
\pset pager off
\echo == columns ==
SELECT column_name, data_type
FROM information_schema.columns
WHERE table_schema = 'public' AND table_name = 'channel_actors'
ORDER BY ordinal_position;
\echo == rows ==
SELECT * FROM channel_actors ORDER BY 1 LIMIT 6;
@@ -0,0 +1,96 @@
#!/usr/bin/env bash
# Prod-server switch to the combined release:
# reasoning-body boundary refresh + offset-0 body boundary + card structural diagnostics
# + wide-screen (autoLayout) card payload
# RUN ON THE PRODUCTION HOST (120.76.218.137) as root. Do not wrap in ssh.
#
# Usage: bash prod_switch_reasoning_widescreen.sh <new-image-ref>
#
# Safety: requires the new image locally; tags the running image as a rollback tag;
# backs up runtime.env with its original mode; refuses unless DOBOTAI_IMAGE appears
# exactly once before and once after the edit and the line count is unchanged;
# refuses unless the new binary carries every release literal.
# Prints only image refs / variable names / literal names - never env values.
set -euo pipefail
NEW_IMAGE="${1:-}"
[ -n "$NEW_IMAGE" ] || { echo "ABORT: pass the new image ref as \$1" >&2; exit 1; }
RUNTIME_ENV="/opt/dobotai/container/runtime.env"
BACKUP_DIR="/opt/dobotai/container/backups"
COMPOSE_SERVICE="dobotai-dingtalk"
RELEASE_SLUG="reasoning-widescreen"
OLD_IMAGE="$(grep -m1 '^DOBOTAI_IMAGE=' "$RUNTIME_ENV" | cut -d= -f2-)"
OLD_SHA="$(echo "$OLD_IMAGE" | sed 's/.*://')"
ROLLBACK_TAG="dobotai-dingtalk:rollback-${OLD_SHA}-before-${RELEASE_SLUG}"
STAMP="$(date +%Y%m%dT%H%M%S)"
BACKUP="$BACKUP_DIR/runtime.env.before-${RELEASE_SLUG}.${STAMP}"
echo "== preflight =="
docker image inspect "$NEW_IMAGE" >/dev/null || { echo "ABORT: $NEW_IMAGE not present locally (load it first)" >&2; exit 1; }
[ "$OLD_IMAGE" != "$NEW_IMAGE" ] || { echo "ABORT: runtime.env already points at $NEW_IMAGE" >&2; exit 1; }
before_count="$(grep -c '^DOBOTAI_IMAGE=' "$RUNTIME_ENV")"
[ "$before_count" = "1" ] || { echo "ABORT: $before_count DOBOTAI_IMAGE lines before edit" >&2; exit 1; }
for lit in sys_full_json_obj autoLayout body_prefix_runes banner_pos PartialFeed; do
if docker run --rm --entrypoint sh "$NEW_IMAGE" -c "grep -aqF '$lit' /usr/local/bin/dobotai"; then
echo "literal $lit: PRESENT"
else
echo "ABORT: $NEW_IMAGE lacks literal '$lit' (wrong artifact)" >&2
exit 1
fi
done
echo "old_image=$OLD_IMAGE"
echo "new_image=$NEW_IMAGE"
echo "rollback_tag=$ROLLBACK_TAG"
echo "== tag rollback target =="
if docker image inspect "$ROLLBACK_TAG" >/dev/null 2>&1; then
echo "rollback tag already exists, keeping it (no overwrite)"
else
docker tag "$OLD_IMAGE" "$ROLLBACK_TAG"
echo "tagged $OLD_IMAGE -> $ROLLBACK_TAG"
fi
echo "== backup runtime.env =="
mkdir -p "$BACKUP_DIR"
ORIG_MODE="$(stat -c '%a' "$RUNTIME_ENV")"
install -m "$ORIG_MODE" "$RUNTIME_ENV" "$BACKUP"
[ "$ORIG_MODE" = "600" ] || chmod 0600 "$RUNTIME_ENV"
echo "backup=$BACKUP mode=$ORIG_MODE"
echo "== rewrite DOBOTAI_IMAGE =="
tmp="$(mktemp /opt/dobotai/container/.runtime.env.XXXXXX)"
chmod "$ORIG_MODE" "$tmp"
awk -v new="$NEW_IMAGE" '/^DOBOTAI_IMAGE=/ { sub(/^DOBOTAI_IMAGE=.*/, "DOBOTAI_IMAGE=" new) } { print }' "$RUNTIME_ENV" >"$tmp"
after_count="$(grep -c "^DOBOTAI_IMAGE=$NEW_IMAGE\$" "$tmp")"
if [ "$after_count" != "1" ] || [ "$(wc -l <"$tmp")" != "$(wc -l <"$RUNTIME_ENV")" ]; then
rm -f "$tmp"
echo "ABORT: post-edit verification failed (match=$after_count)" >&2
exit 1
fi
mv "$tmp" "$RUNTIME_ENV"
grep -n '^DOBOTAI_IMAGE=' "$RUNTIME_ENV"
echo "== recreate container =="
cd /opt/dobotai/container
docker compose --env-file runtime.env -f compose.yaml up -d --no-deps --force-recreate "$COMPOSE_SERVICE"
echo "== readback =="
sleep 4
curl -fsS -m 5 http://127.0.0.1:8090/healthz; echo
docker inspect "$COMPOSE_SERVICE" --format 'image={{.Config.Image}} status={{.State.Status}} started={{.State.StartedAt}}'
for lit in sys_full_json_obj autoLayout body_prefix_runes banner_pos PartialFeed; do
if docker exec "$COMPOSE_SERVICE" grep -aqF "$lit" /proc/1/root/usr/local/bin/dobotai; then
echo "running literal $lit: PRESENT"
else
echo "running literal $lit: ABSENT"
fi
done
echo "== log scan (since 90s) =="
docker logs "$COMPOSE_SERVICE" --since 90s 2>&1 | grep -icE 'panic|fatal|level=error| ERROR ' || echo "0 error/panic lines"
docker logs "$COMPOSE_SERVICE" --since 90s 2>&1 | tail -6
echo "== rollback (one command, run on this host) =="
echo "sed -i 's|^DOBOTAI_IMAGE=.*|DOBOTAI_IMAGE=$OLD_IMAGE|' $RUNTIME_ENV && cd /opt/dobotai/container && docker compose --env-file runtime.env -f compose.yaml up -d --no-deps --force-recreate $COMPOSE_SERVICE"
echo "or restore: install -m $ORIG_MODE $BACKUP $RUNTIME_ENV (then the compose line above)"
+67
View File
@@ -0,0 +1,67 @@
#!/usr/bin/env bash
# Test-server switch to the build that delivers card images as DingTalk media ids.
# Rollback tag + 0600 env backup first; refuses unless the DOBOTAI_IMAGE
# substitution is exactly 1 before and exactly 1 after. Read-only readback at
# the end. No secrets are printed.
set -euo pipefail
NEW_IMAGE="${NEW_IMAGE:-dobotai-dingtalk:local-dfc7c8dba956}"
RUNTIME_ENV="/opt/dobotai/container/runtime.env"
OLD_IMAGE="$(grep -m1 '^DOBOTAI_IMAGE=' "$RUNTIME_ENV" | cut -d= -f2-)"
ROLLBACK_TAG="dobotai-dingtalk:rollback-$(echo "$OLD_IMAGE" | sed 's/.*://')-before-card-image-media"
STAMP="$(date +%Y%m%dT%H%M%S)"
BACKUP_DIR="/opt/dobotai/container/backups"
BACKUP="$BACKUP_DIR/runtime.env.before-card-image-media.$STAMP"
echo "== preflight =="
docker image inspect "$NEW_IMAGE" >/dev/null
[ "$OLD_IMAGE" != "$NEW_IMAGE" ] || { echo "ABORT: runtime.env already points at $NEW_IMAGE" >&2; exit 1; }
before_count="$(grep -c '^DOBOTAI_IMAGE=' "$RUNTIME_ENV")"
[ "$before_count" = "1" ] || { echo "ABORT: $before_count DOBOTAI_IMAGE lines before edit" >&2; exit 1; }
echo "old_image=$OLD_IMAGE"
echo "new_image=$NEW_IMAGE"
echo "rollback_tag=$ROLLBACK_TAG"
echo "== tag rollback target =="
if docker image inspect "$ROLLBACK_TAG" >/dev/null 2>&1; then
echo "rollback tag already exists, keeping it (no overwrite)"
else
docker tag "$OLD_IMAGE" "$ROLLBACK_TAG"
echo "tagged $OLD_IMAGE -> $ROLLBACK_TAG"
fi
echo "== backup runtime.env =="
mkdir -p "$BACKUP_DIR"
install -m 0600 "$RUNTIME_ENV" "$BACKUP"
echo "backup=$BACKUP"
echo "== rewrite DOBOTAI_IMAGE =="
tmp="$(mktemp /opt/dobotai/container/.runtime.env.XXXXXX)"
awk -v new="$NEW_IMAGE" '/^DOBOTAI_IMAGE=/ { sub(/^DOBOTAI_IMAGE=.*/, "DOBOTAI_IMAGE=" new) } { print }' "$RUNTIME_ENV" >"$tmp"
after_count="$(grep -c "^DOBOTAI_IMAGE=$NEW_IMAGE\$" "$tmp")"
if [ "$after_count" != "1" ] || [ "$(wc -l <"$tmp")" != "$(wc -l <"$RUNTIME_ENV")" ]; then
rm -f "$tmp"
echo "ABORT: post-edit verification failed (match=$after_count)" >&2
exit 1
fi
chmod 0644 "$tmp"
mv "$tmp" "$RUNTIME_ENV"
grep -n '^DOBOTAI_IMAGE=' "$RUNTIME_ENV"
echo "== recreate container =="
cd /opt/dobotai/container
docker compose --env-file runtime.env -f compose.yaml up -d --no-deps --force-recreate dobotai-dingtalk
echo "== readback =="
sleep 3
curl -fsS http://127.0.0.1:8090/healthz; echo
docker inspect dobotai-dingtalk --format 'image={{.Config.Image}} status={{.State.Status}} started={{.State.StartedAt}}'
for lit in oapi.dingtalk.com/media/upload DINGTALK_CARD_IMAGE_MEDIA; do
if docker exec dobotai-dingtalk grep -aqF "$lit" /proc/1/root/usr/local/bin/dobotai; then
echo "literal $lit: PRESENT"
else
echo "literal $lit: ABSENT"
fi
done
docker logs dobotai-dingtalk --since 60s 2>&1 | tail -6
echo "rollback: DOBOTAI_IMAGE=$OLD_IMAGE (+ compose up -d --no-deps --force-recreate dobotai-dingtalk)"
+67
View File
@@ -0,0 +1,67 @@
#!/usr/bin/env bash
# Test-server switch to the wide-screen (autoLayout) card image.
# Rollback tag + 0600 env backup first; refuses unless the DOBOTAI_IMAGE
# substitution is exactly 1 before and exactly 1 after. Read-only readback
# at the end. No secrets are printed.
set -euo pipefail
NEW_IMAGE="dobotai-dingtalk:local-cardfix-ba090d17"
RUNTIME_ENV="/opt/dobotai/container/runtime.env"
OLD_IMAGE="$(grep -m1 '^DOBOTAI_IMAGE=' "$RUNTIME_ENV" | cut -d= -f2-)"
ROLLBACK_TAG="dobotai-dingtalk:rollback-$(echo "$OLD_IMAGE" | sed 's/.*://')-before-card-widescreen"
STAMP="$(date +%Y%m%dT%H%M%S)"
BACKUP_DIR="/opt/dobotai/container/backups"
BACKUP="$BACKUP_DIR/runtime.env.before-card-widescreen.$STAMP"
echo "== preflight =="
docker image inspect "$NEW_IMAGE" >/dev/null
[ "$OLD_IMAGE" != "$NEW_IMAGE" ] || { echo "ABORT: runtime.env already points at $NEW_IMAGE" >&2; exit 1; }
before_count="$(grep -c '^DOBOTAI_IMAGE=' "$RUNTIME_ENV")"
[ "$before_count" = "1" ] || { echo "ABORT: $before_count DOBOTAI_IMAGE lines before edit" >&2; exit 1; }
echo "old_image=$OLD_IMAGE"
echo "new_image=$NEW_IMAGE"
echo "rollback_tag=$ROLLBACK_TAG"
echo "== tag rollback target =="
if docker image inspect "$ROLLBACK_TAG" >/dev/null 2>&1; then
echo "rollback tag already exists, keeping it (no overwrite)"
else
docker tag "$OLD_IMAGE" "$ROLLBACK_TAG"
echo "tagged $OLD_IMAGE -> $ROLLBACK_TAG"
fi
echo "== backup runtime.env =="
mkdir -p "$BACKUP_DIR"
install -m 0600 "$RUNTIME_ENV" "$BACKUP"
echo "backup=$BACKUP"
echo "== rewrite DOBOTAI_IMAGE =="
tmp="$(mktemp /opt/dobotai/container/.runtime.env.XXXXXX)"
awk -v new="$NEW_IMAGE" '/^DOBOTAI_IMAGE=/ { sub(/^DOBOTAI_IMAGE=.*/, "DOBOTAI_IMAGE=" new) } { print }' "$RUNTIME_ENV" >"$tmp"
after_count="$(grep -c "^DOBOTAI_IMAGE=$NEW_IMAGE\$" "$tmp")"
if [ "$after_count" != "1" ] || [ "$(wc -l <"$tmp")" != "$(wc -l <"$RUNTIME_ENV")" ]; then
rm -f "$tmp"
echo "ABORT: post-edit verification failed (match=$after_count)" >&2
exit 1
fi
chmod 0644 "$tmp"
mv "$tmp" "$RUNTIME_ENV"
grep -n '^DOBOTAI_IMAGE=' "$RUNTIME_ENV"
echo "== recreate container =="
cd /opt/dobotai/container
docker compose --env-file runtime.env -f compose.yaml up -d --no-deps --force-recreate dobotai-dingtalk
echo "== readback =="
sleep 3
curl -fsS http://127.0.0.1:8090/healthz; echo
docker inspect dobotai-dingtalk --format 'image={{.Config.Image}} status={{.State.Status}} started={{.State.StartedAt}}'
for lit in sys_full_json_obj autoLayout _CARD_DEBUG_TOOL_ENTRY; do
if docker exec dobotai-dingtalk grep -aqF "$lit" /proc/1/root/usr/local/bin/dobotai; then
echo "literal $lit: PRESENT"
else
echo "literal $lit: ABSENT"
fi
done
docker logs dobotai-dingtalk --since 60s 2>&1 | tail -5
echo "rollback: DOBOTAI_IMAGE=$OLD_IMAGE (+ compose up -d --no-deps --force-recreate dobotai-dingtalk)"
+66
View File
@@ -0,0 +1,66 @@
#!/usr/bin/env bash
# Generic test-server image switch: rollback tag first, 0600 runtime.env backup,
# post-edit verification (exactly 1 DOBOTAI_IMAGE line, unchanged line count),
# container recreate, then read-only readback incl. binary literal presence.
# Usage: bash test_switch_image.sh <image:tag> <label>
set -euo pipefail
NEW_IMAGE="${1:?usage: test_switch_image.sh <image:tag> <label>}"
LABEL="${2:?usage: test_switch_image.sh <image:tag> <label>}"
RUNTIME_ENV=/opt/dobotai/container/runtime.env
OLD_IMAGE="$(grep -m1 '^DOBOTAI_IMAGE=' "$RUNTIME_ENV" | cut -d= -f2-)"
ROLLBACK_TAG="dobotai-dingtalk:rollback-$(echo "$OLD_IMAGE" | sed 's/.*://')-before-$LABEL"
STAMP="$(date +%Y%m%dT%H%M%S)"
BACKUP_DIR=/opt/dobotai/container/backups
BACKUP="$BACKUP_DIR/runtime.env.before-$LABEL.$STAMP"
echo "== preflight =="
docker image inspect "$NEW_IMAGE" >/dev/null
[ "$OLD_IMAGE" != "$NEW_IMAGE" ] || { echo "ABORT: runtime.env already points at $NEW_IMAGE" >&2; exit 1; }
[ "$(grep -c '^DOBOTAI_IMAGE=' "$RUNTIME_ENV")" = "1" ] || { echo "ABORT: DOBOTAI_IMAGE line count != 1" >&2; exit 1; }
echo "old_image=$OLD_IMAGE"
echo "new_image=$NEW_IMAGE"
echo "rollback_tag=$ROLLBACK_TAG"
echo "== tag rollback target =="
if docker image inspect "$ROLLBACK_TAG" >/dev/null 2>&1; then
echo "rollback tag already exists, keeping it (no overwrite)"
else
docker tag "$OLD_IMAGE" "$ROLLBACK_TAG"
echo "tagged $OLD_IMAGE -> $ROLLBACK_TAG"
fi
echo "== backup runtime.env =="
mkdir -p "$BACKUP_DIR"
install -m 0600 "$RUNTIME_ENV" "$BACKUP"
echo "backup=$BACKUP"
echo "== rewrite DOBOTAI_IMAGE =="
tmp="$(mktemp /opt/dobotai/container/.runtime.env.XXXXXX)"
awk -v new="$NEW_IMAGE" '/^DOBOTAI_IMAGE=/ { sub(/^DOBOTAI_IMAGE=.*/, "DOBOTAI_IMAGE=" new) } { print }' "$RUNTIME_ENV" >"$tmp"
after_count="$(grep -c "^DOBOTAI_IMAGE=$NEW_IMAGE\$" "$tmp")"
if [ "$after_count" != "1" ] || [ "$(wc -l <"$tmp")" != "$(wc -l <"$RUNTIME_ENV")" ]; then
rm -f "$tmp"
echo "ABORT: post-edit verification failed (match=$after_count)" >&2
exit 1
fi
chmod 0644 "$tmp"
mv "$tmp" "$RUNTIME_ENV"
grep -n '^DOBOTAI_IMAGE=' "$RUNTIME_ENV"
echo "== recreate container =="
cd /opt/dobotai/container
docker compose --env-file runtime.env -f compose.yaml up -d --no-deps --force-recreate dobotai-dingtalk
echo "== readback =="
sleep 3
curl -fsS http://127.0.0.1:8090/healthz; echo
docker inspect dobotai-dingtalk --format 'image={{.Config.Image}} status={{.State.Status}} started={{.State.StartedAt}}'
for lit in PartialFeed autoLayout 智能推理模式; do
if docker exec dobotai-dingtalk grep -aqF "$lit" /proc/1/root/usr/local/bin/dobotai; then
echo "literal $lit: PRESENT"
else
echo "literal $lit: ABSENT"
fi
done
docker logs dobotai-dingtalk --since 60s 2>&1 | tail -8
echo "rollback: set DOBOTAI_IMAGE=$OLD_IMAGE in $RUNTIME_ENV (backup $BACKUP) then: docker compose --env-file runtime.env -f compose.yaml up -d --no-deps --force-recreate dobotai-dingtalk"
+40
View File
@@ -0,0 +1,40 @@
#!/usr/bin/env bash
set -euo pipefail
if [[ ${EUID} -ne 0 ]]; then
echo "请在仓库根目录执行:sudo bash ./install-docker.sh" >&2
exit 1
fi
repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
cd "$repo_root"
[[ -f Dockerfile && -f deploy/dobotai-dingtalk/install-docker.sh ]] || { echo "必须从完整DobotAI源码仓库根目录运行。" >&2; exit 1; }
command -v git >/dev/null || { echo "未找到git。" >&2; exit 1; }
git rev-parse --is-inside-work-tree >/dev/null || { echo "请从git clone得到的仓库中执行。" >&2; exit 1; }
install_docker_if_missing() {
if command -v docker >/dev/null && docker compose version >/dev/null 2>&1; then return; fi
command -v apt-get >/dev/null || { echo "请先安装Docker Engine和Docker Compose v2。" >&2; exit 1; }
export DEBIAN_FRONTEND=noninteractive
apt-get update
if apt-cache show docker-compose-v2 >/dev/null 2>&1; then
apt-get install -y docker.io docker-compose-v2
elif apt-cache show docker-compose-plugin >/dev/null 2>&1; then
apt-get install -y docker.io docker-compose-plugin
else
echo "APT源缺少Docker Compose v2插件。" >&2; exit 1
fi
systemctl enable --now docker
}
install_docker_if_missing
docker compose version >/dev/null
commit_sha="$(git rev-parse HEAD)"
commit_short="$(git rev-parse --short=12 HEAD)"
image="dobotai-dingtalk:local-${commit_short}"
echo "[1/2] 构建DobotAI应用镜像:$image"
docker build --build-arg "GOPROXY=https://goproxy.cn,direct" --build-arg "VERSION=local-${commit_short}" --build-arg "COMMIT_SHA=${commit_sha}" --build-arg "BUILD_TIME=$(date -u +%Y-%m-%dT%H:%M:%SZ)" --label "org.opencontainers.image.revision=${commit_sha}" --tag "$image" "$repo_root"
echo "[2/2] 安装应用容器资产;阿里云RDS将在配置后检查或初始化"
export DOBOTAI_IMAGE="$image"
export DOBOTAI_SOURCE_ROOT="$repo_root"
exec bash "$repo_root/deploy/dobotai-dingtalk/install-docker.sh"
+19
View File
@@ -21,6 +21,24 @@ const (
cardCreateURL = "https://api.dingtalk.com/v1.0/card/instances/createAndDeliver"
cardStreamingURL = "https://api.dingtalk.com/v1.0/card/streaming"
cardDebugPrivateKey = "_CARD_DEBUG_TOOL_ENTRY"
// cardWideScreenJSONKey is DingTalk's reserved system field inside
// cardData.cardParamMap. Template variables whose value is a complex type
// (object / array of objects) must be passed through this field as a JSON
// string; plain string/number/boolean variables go in cardParamMap
// directly. Putting a complex value anywhere else (for example as a
// sibling of cardParamMap under cardData) leaves the template variable
// unassigned and silently falls back to the default layout.
cardWideScreenJSONKey = "sys_full_json_obj"
// cardWideScreenJSON configures the AI card's root "config": PC clients
// render the card wide (auto layout) when autoLayout is true. The card
// template must declare an object variable "config" with a boolean child
// "autoLayout" (card builder > 数据源 > 编辑变量) and the value must be
// passed when the card is created - setting it statically inside the
// template does not take effect. Mobile clients keep their own adaptive
// layout, so this only widens the card on desktop.
cardWideScreenJSON = `{"config":{"autoLayout":true}}`
)
// CardRequest is the input for sending a final AI card.
@@ -229,6 +247,7 @@ func (s *CardSender) buildCardBody(req CardRequest, outTrackID string) map[strin
"feedbackReasonSelected": "",
"showFeedbackReasons": "false",
"feedbackMessage": "",
cardWideScreenJSONKey: cardWideScreenJSON,
},
},
"userIdType": 1,
+56 -1
View File
@@ -3,8 +3,12 @@ package dingtalk
import (
"fmt"
"strings"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/reasoningmode"
)
const reasoningBannerMarker = "【智能推理模式】"
type cardContentShape struct {
bytes int
lines int
@@ -13,20 +17,63 @@ type cardContentShape struct {
delimiterLines int
confirmedTable bool
confirmedTablePos int
// Structural split state: where the answer body starts inside this exact
// payload. bodyPrefixRunes > 0 means the update still carries narration in
// front of the body (the reasoning text the card is supposed to hide).
bodyOK bool
bodyLine int
bodyPrefixRunes int
bannerPos int
firstHeadingLine int
imageRefs int
externalImageRefs int
mediaImageRefs int
}
func inspectCardContent(content string) cardContentShape {
shape := cardContentShape{
bytes: len(content),
confirmedTablePos: -1,
bodyLine: -1,
firstHeadingLine: -1,
bannerPos: -1,
}
if content == "" {
return shape
}
shape.lines = strings.Count(content, "\n") + 1
shape.imageRefs, shape.externalImageRefs, shape.mediaImageRefs = cardImageRefCounts(content)
shape.bannerPos = strings.Index(content, reasoningBannerMarker)
// The streaming path decorates every frame with the reasoning banner, so the
// prefix in front of the body has to be measured after that banner: only
// what remains is narration the card is meant to hide.
bannerSkip := 0
if shape.bannerPos == 0 {
if idx := strings.Index(content, "\n"); idx >= 0 {
bannerSkip = idx + 1
for bannerSkip < len(content) && content[bannerSkip] == '\n' {
bannerSkip++
}
}
}
if offset, ok := reasoningmode.BodyStartOffset(content); ok {
shape.bodyOK = true
shape.bodyLine = strings.Count(content[:offset], "\n")
prefix := content[:offset]
if bannerSkip > 0 && bannerSkip <= len(prefix) {
prefix = prefix[bannerSkip:]
}
shape.bodyPrefixRunes = len([]rune(prefix))
}
lines := splitMarkdownStreamLines(content)
for i, line := range lines {
trimmed := strings.TrimSpace(line.text)
if shape.firstHeadingLine < 0 && !line.inFence &&
strings.HasPrefix(trimmed, "#") && strings.TrimLeft(trimmed, "#") != "" {
shape.firstHeadingLine = i
}
if strings.Contains(line.text, "|") {
shape.pipeLines++
}
@@ -50,7 +97,7 @@ func inspectCardContent(content string) cardContentShape {
func formatCardUpdateDiagnostic(content string, isFinalize, isError bool) string {
shape := inspectCardContent(content)
return fmt.Sprintf(
"DingTalk card update shape: finalize=%t error=%t bytes=%d lines=%d pipe_lines=%d leading_pipe_lines=%d delimiter_lines=%d confirmed_table=%t table_start=%d",
"DingTalk card update shape: finalize=%t error=%t bytes=%d lines=%d pipe_lines=%d leading_pipe_lines=%d delimiter_lines=%d confirmed_table=%t table_start=%d body_ok=%t body_line=%d body_prefix_runes=%d banner_pos=%d first_heading_line=%d image_refs=%d external_image_refs=%d media_image_refs=%d",
isFinalize,
isError,
shape.bytes,
@@ -60,5 +107,13 @@ func formatCardUpdateDiagnostic(content string, isFinalize, isError bool) string
shape.delimiterLines,
shape.confirmedTable,
shape.confirmedTablePos,
shape.bodyOK,
shape.bodyLine,
shape.bodyPrefixRunes,
shape.bannerPos,
shape.firstHeadingLine,
shape.imageRefs,
shape.externalImageRefs,
shape.mediaImageRefs,
)
}
@@ -0,0 +1,488 @@
package dingtalk
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log"
"mime/multipart"
"net/http"
"net/textproto"
"regexp"
"strings"
"sync"
"time"
)
// DingTalk AI cards only render images that live in DingTalk's own media store:
// an external HTTPS image URL shows up in a normal markdown message but is
// dropped inside a streaming/short-card body (verified on both desktop and
// mobile, 2026-09-17). The card content therefore rewrites our own published
// image URLs into DingTalk media_id values obtained through the legacy
// oapi media upload endpoint, which both clients render.
const (
defaultCardImageTokenURL = "https://oapi.dingtalk.com/gettoken"
defaultCardImageUploadURL = "https://oapi.dingtalk.com/media/upload"
defaultCardImageMediaTTL = 12 * time.Hour
defaultCardImageWaitTimeout = 5 * time.Second
defaultCardImageFetchTimeout = 15 * time.Second
defaultCardImageUploadTimeout = 20 * time.Second
defaultCardImageMaxBytes = 8 << 20
cardImageTokenRefreshMargin = 5 * time.Minute
maxConcurrentCardImageUploads = 3
)
// cardImageStableIDRE matches the deterministic image id produced by
// outboundimage.StableHashPublisher (UUIDv5 shaped, lowercase hex).
var cardImageStableIDRE = regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-5[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$`)
// cardMarkdownImageRE captures the URL of a complete markdown image, including
// the optional title slot. Incomplete image syntax (still streaming) is not
// matched, so it is never rewritten.
var cardMarkdownImageRE = regexp.MustCompile(`!\[[^\]]*\]\(([^\s)\r\n]+)(?:\s+(?:"[^"\r\n]*"|'[^'\r\n]*'|\([^)\r\n]*\)))?\)`)
// CardImageMediaRewriter converts card-body image references into DingTalk
// media ids. Implementations must be safe for concurrent use and must never
// fail a card update: on any error the original reference is kept.
type CardImageMediaRewriter interface {
RewriteForCard(ctx context.Context, content string, wait bool) string
}
// CardImageMediaOptions configures the resolver. ClientID/ClientSecret are the
// same Robot app credentials used for access tokens, and ImageURLPrefix is the
// public prefix of our own published images.
type CardImageMediaOptions struct {
ClientID string
ClientSecret string
ImageURLPrefix string
// FetchURLPrefix optionally replaces ImageURLPrefix when downloading the
// bytes back (for example a loopback address), so the upload never depends
// on the container being able to reach its own public host name.
FetchURLPrefix string
TokenURL string
UploadURL string
HTTPClient *http.Client
TTL time.Duration
WaitTimeout time.Duration
MaxBytes int64
}
type cardImageMediaEntry struct {
mediaID string
expiresAt time.Time
}
type cardImageMediaJob struct {
done chan struct{}
mediaID string
err error
}
type cardImageMediaResolver struct {
clientID string
clientSecret string
imageURLPrefix string
fetchURLPrefix string
tokenURL string
uploadURL string
client *http.Client
ttl time.Duration
waitTimeout time.Duration
maxBytes int64
sem chan struct{}
now func() time.Time
mu sync.Mutex
entries map[string]cardImageMediaEntry
inflight map[string]*cardImageMediaJob
token string
tokenExpiry time.Time
}
// NewCardImageMediaRewriter builds the resolver. It returns a nil rewriter and
// no error when the feature is not configured, which keeps the caller simple.
func NewCardImageMediaRewriter(opts CardImageMediaOptions) (CardImageMediaRewriter, error) {
clientID := strings.TrimSpace(opts.ClientID)
clientSecret := strings.TrimSpace(opts.ClientSecret)
prefix := strings.TrimSpace(opts.ImageURLPrefix)
if clientID == "" || clientSecret == "" || prefix == "" {
return nil, nil
}
if !strings.HasSuffix(prefix, "/") {
prefix += "/"
}
tokenURL := strings.TrimSpace(opts.TokenURL)
if tokenURL == "" {
tokenURL = defaultCardImageTokenURL
}
uploadURL := strings.TrimSpace(opts.UploadURL)
if uploadURL == "" {
uploadURL = defaultCardImageUploadURL
}
ttl := opts.TTL
if ttl <= 0 {
ttl = defaultCardImageMediaTTL
}
waitTimeout := opts.WaitTimeout
if waitTimeout <= 0 {
waitTimeout = defaultCardImageWaitTimeout
}
maxBytes := opts.MaxBytes
if maxBytes <= 0 {
maxBytes = defaultCardImageMaxBytes
}
client := opts.HTTPClient
if client == nil {
transport := http.DefaultTransport.(*http.Transport).Clone()
client = &http.Client{Timeout: defaultCardImageFetchTimeout, Transport: transport}
}
return &cardImageMediaResolver{
clientID: clientID,
clientSecret: clientSecret,
imageURLPrefix: prefix,
fetchURLPrefix: strings.TrimSpace(opts.FetchURLPrefix),
tokenURL: tokenURL,
uploadURL: uploadURL,
client: client,
ttl: ttl,
waitTimeout: waitTimeout,
maxBytes: maxBytes,
sem: make(chan struct{}, maxConcurrentCardImageUploads),
now: time.Now,
entries: make(map[string]cardImageMediaEntry),
inflight: make(map[string]*cardImageMediaJob),
}, nil
}
// RewriteForCard replaces complete markdown image references that point at our
// own published images with a DingTalk media id. With wait=false the rewrite
// only uses media ids that are already known and kicks off the upload in the
// background, so streaming frames never block on DingTalk; with wait=true it
// waits (bounded) for in-flight uploads, which is what the final card update
// uses.
func (r *cardImageMediaResolver) RewriteForCard(ctx context.Context, content string, wait bool) string {
if r == nil || strings.TrimSpace(content) == "" {
return content
}
var waitCtx context.Context
if wait {
bound := r.waitTimeout
if bound <= 0 {
bound = defaultCardImageWaitTimeout
}
var cancel context.CancelFunc
waitCtx, cancel = context.WithTimeout(context.WithoutCancel(ctx), bound)
defer cancel()
}
rewritten := cardMarkdownImageRE.ReplaceAllStringFunc(content, func(image string) string {
match := cardMarkdownImageRE.FindStringSubmatch(image)
if len(match) < 2 {
return image
}
rawURL := strings.TrimSpace(match[1])
if rawURL == "" || !r.isOwnImageURL(rawURL) {
return image
}
if mediaID, ok := r.lookup(rawURL); ok {
return replaceCardImageURL(image, match[1], mediaID)
}
job := r.startUpload(rawURL)
if !wait {
return image
}
select {
case <-job.done:
case <-waitCtx.Done():
}
if job.mediaID != "" {
return replaceCardImageURL(image, match[1], job.mediaID)
}
if job.err != nil {
log.Printf("DingTalk card image media upload failed, keeping external reference")
}
return image
})
return rewritten
}
func replaceCardImageURL(image, rawURL, mediaID string) string {
if mediaID == "" || rawURL == "" {
return image
}
return strings.Replace(image, rawURL, mediaID, 1)
}
func (r *cardImageMediaResolver) isOwnImageURL(rawURL string) bool {
if r.imageURLPrefix == "" || !strings.HasPrefix(rawURL, r.imageURLPrefix) {
return false
}
id := strings.TrimPrefix(rawURL, r.imageURLPrefix)
if id == "" || strings.ContainsAny(id, "/?#") {
return false
}
return cardImageStableIDRE.MatchString(id)
}
func (r *cardImageMediaResolver) lookup(rawURL string) (string, bool) {
r.mu.Lock()
defer r.mu.Unlock()
entry, ok := r.entries[rawURL]
if !ok || r.now().After(entry.expiresAt) {
return "", false
}
return entry.mediaID, true
}
// startUpload returns a job for the image, reusing an in-flight upload or a
// cached media id when available. The upload itself runs on a background
// context so a streaming frame's cancellation cannot abort it half way.
func (r *cardImageMediaResolver) startUpload(rawURL string) *cardImageMediaJob {
r.mu.Lock()
if entry, ok := r.entries[rawURL]; ok && r.now().Before(entry.expiresAt) {
job := &cardImageMediaJob{done: make(chan struct{}), mediaID: entry.mediaID}
close(job.done)
r.mu.Unlock()
return job
}
if job, ok := r.inflight[rawURL]; ok {
r.mu.Unlock()
return job
}
job := &cardImageMediaJob{done: make(chan struct{})}
r.inflight[rawURL] = job
r.mu.Unlock()
go func() {
uploadCtx, cancel := context.WithTimeout(context.Background(), defaultCardImageUploadTimeout)
defer cancel()
mediaID, err := r.upload(uploadCtx, rawURL)
job.mediaID = mediaID
job.err = err
r.mu.Lock()
delete(r.inflight, rawURL)
if err == nil && mediaID != "" {
r.entries[rawURL] = cardImageMediaEntry{mediaID: mediaID, expiresAt: r.now().Add(r.ttl)}
}
r.mu.Unlock()
close(job.done)
}()
return job
}
func (r *cardImageMediaResolver) upload(ctx context.Context, rawURL string) (string, error) {
select {
case r.sem <- struct{}{}:
case <-ctx.Done():
return "", ctx.Err()
}
defer func() { <-r.sem }()
data, contentType, err := r.fetch(ctx, rawURL)
if err != nil {
return "", err
}
token, err := r.accessToken(ctx)
if err != nil {
return "", err
}
return r.uploadMedia(ctx, token, data, contentType)
}
func (r *cardImageMediaResolver) fetchURL(rawURL string) string {
targets := r.fetchURLs(rawURL)
if len(targets) == 0 {
return rawURL
}
return targets[0]
}
// fetchURLs lists the download candidates for one of our published images: the
// loopback prefix first (cheap, stays inside the host) and the public URL as a
// fallback so a missing local route never turns into a missing card image.
func (r *cardImageMediaResolver) fetchURLs(rawURL string) []string {
relative := ""
if strings.HasPrefix(rawURL, r.imageURLPrefix) {
relative = strings.TrimPrefix(rawURL, r.imageURLPrefix)
}
targets := make([]string, 0, 2)
if r.fetchURLPrefix != "" && relative != "" {
targets = append(targets, strings.TrimRight(r.fetchURLPrefix, "/")+"/"+relative)
}
if relative == "" || r.fetchURLPrefix == "" {
targets = append(targets, rawURL)
} else {
targets = append(targets, strings.TrimRight(r.imageURLPrefix, "/")+"/"+relative)
}
return targets
}
func (r *cardImageMediaResolver) fetch(ctx context.Context, rawURL string) ([]byte, string, error) {
var firstErr error
for _, target := range r.fetchURLs(rawURL) {
data, contentType, err := r.fetchOnce(ctx, target)
if err == nil {
return data, contentType, nil
}
if firstErr == nil {
firstErr = err
}
if ctx.Err() != nil {
break
}
}
if firstErr == nil {
firstErr = errors.New("card image fetch: no candidate URL")
}
return nil, "", firstErr
}
func (r *cardImageMediaResolver) fetchOnce(ctx context.Context, target string) ([]byte, string, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
if err != nil {
return nil, "", err
}
resp, err := r.client.Do(req)
if err != nil {
return nil, "", err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return nil, "", fmt.Errorf("card image fetch: unexpected status %d", resp.StatusCode)
}
contentType := strings.TrimSpace(strings.Split(resp.Header.Get("Content-Type"), ";")[0])
if !strings.HasPrefix(strings.ToLower(contentType), "image/") {
return nil, "", fmt.Errorf("card image fetch: unexpected content type")
}
data, err := io.ReadAll(io.LimitReader(resp.Body, r.maxBytes+1))
if err != nil {
return nil, "", err
}
if int64(len(data)) > r.maxBytes {
return nil, "", errors.New("card image fetch: image exceeds upload size limit")
}
if len(data) == 0 {
return nil, "", errors.New("card image fetch: empty image")
}
return data, contentType, nil
}
// cardImageRefCounts splits the markdown image references of a card body into
// externally hosted URLs and DingTalk media ids, so operators can tell from the
// logs whether a card body can actually render its images.
func cardImageRefCounts(content string) (total, external, media int) {
for _, match := range cardMarkdownImageRE.FindAllStringSubmatch(content, -1) {
if len(match) < 2 {
continue
}
total++
if strings.Contains(strings.TrimSpace(match[1]), "://") {
external++
} else {
media++
}
}
return total, external, media
}
func (r *cardImageMediaResolver) accessToken(ctx context.Context) (string, error) {
r.mu.Lock()
if r.token != "" && r.now().Before(r.tokenExpiry) {
token := r.token
r.mu.Unlock()
return token, nil
}
r.mu.Unlock()
values := map[string]string{"appkey": r.clientID, "appsecret": r.clientSecret}
payload, err := json.Marshal(values)
if err != nil {
return "", err
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, r.tokenURL, bytes.NewReader(payload))
if err != nil {
return "", err
}
req.Header.Set("Content-Type", "application/json")
resp, err := r.client.Do(req)
if err != nil {
return "", err
}
defer resp.Body.Close()
body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
if err != nil {
return "", err
}
var parsed struct {
ErrCode int `json:"errcode"`
ErrMsg string `json:"errmsg"`
AccessToken string `json:"access_token"`
ExpiresIn int `json:"expires_in"`
}
if err := json.Unmarshal(body, &parsed); err != nil {
return "", fmt.Errorf("card image token: invalid response")
}
if parsed.ErrCode != 0 || strings.TrimSpace(parsed.AccessToken) == "" {
return "", fmt.Errorf("card image token: rejected with errcode %d", parsed.ErrCode)
}
expiresIn := parsed.ExpiresIn
if expiresIn <= 0 {
expiresIn = 7200
}
r.mu.Lock()
r.token = parsed.AccessToken
r.tokenExpiry = r.now().Add(time.Duration(expiresIn)*time.Second - cardImageTokenRefreshMargin)
r.mu.Unlock()
return parsed.AccessToken, nil
}
func (r *cardImageMediaResolver) uploadMedia(ctx context.Context, token string, data []byte, contentType string) (string, error) {
var buffer bytes.Buffer
writer := multipart.NewWriter(&buffer)
header := make(textproto.MIMEHeader)
header.Set("Content-Disposition", `form-data; name="media"; filename="card-image"`)
if contentType == "" {
contentType = "image/png"
}
header.Set("Content-Type", contentType)
part, err := writer.CreatePart(header)
if err != nil {
return "", err
}
if _, err := part.Write(data); err != nil {
return "", err
}
if err := writer.Close(); err != nil {
return "", err
}
uploadURL := r.uploadURL + "?access_token=" + token + "&type=image"
req, err := http.NewRequestWithContext(ctx, http.MethodPost, uploadURL, &buffer)
if err != nil {
return "", err
}
req.Header.Set("Content-Type", writer.FormDataContentType())
resp, err := r.client.Do(req)
if err != nil {
return "", err
}
defer resp.Body.Close()
body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
if err != nil {
return "", err
}
var parsed struct {
ErrCode int `json:"errcode"`
ErrMsg string `json:"errmsg"`
MediaID string `json:"media_id"`
}
if err := json.Unmarshal(body, &parsed); err != nil {
return "", errors.New("card image upload: invalid response")
}
if parsed.ErrCode != 0 || strings.TrimSpace(parsed.MediaID) == "" {
return "", fmt.Errorf("card image upload: rejected with errcode %d", parsed.ErrCode)
}
return parsed.MediaID, nil
}
@@ -0,0 +1,398 @@
package dingtalk
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
)
const cardImageTestID = "5f0a1c2e-3b4d-5e6f-8a9b-0c1d2e3f4a5b"
const cardImageTestMediaID = "@lADP-test-media-id"
type cardImageTestServer struct {
server *httptest.Server
imageBytes []byte
oversizeBytes []byte
uploadErrcode int
tokenCalls int
uploadCalls int
uploadedBytes int
fetched []string
mu sync.Mutex
uploadedC chan struct{}
}
func newCardImageTestServer(t *testing.T) *cardImageTestServer {
t.Helper()
fixture := &cardImageTestServer{
imageBytes: []byte("\x89PNG\r\n\x1a\nfixture-bytes"),
oversizeBytes: make([]byte, 512),
uploadedC: make(chan struct{}, 4),
}
mux := http.NewServeMux()
serveImage := func(w http.ResponseWriter, r *http.Request) {
fixture.mu.Lock()
fixture.fetched = append(fixture.fetched, r.URL.Path)
fixture.mu.Unlock()
body := fixture.imageBytes
if strings.HasSuffix(r.URL.Path, "big") {
body = fixture.oversizeBytes
}
w.Header().Set("Content-Type", "image/png")
_, _ = w.Write(body)
}
mux.HandleFunc("/dingtalk/i/", serveImage)
mux.HandleFunc("/images/", serveImage)
mux.HandleFunc("/gettoken", func(w http.ResponseWriter, r *http.Request) {
fixture.mu.Lock()
fixture.tokenCalls++
fixture.mu.Unlock()
_ = json.NewEncoder(w).Encode(map[string]any{
"errcode": 0, "errmsg": "ok", "access_token": "test-token", "expires_in": 7200,
})
})
mux.HandleFunc("/media/upload", func(w http.ResponseWriter, r *http.Request) {
if r.URL.Query().Get("access_token") != "test-token" || r.URL.Query().Get("type") != "image" {
w.WriteHeader(http.StatusBadRequest)
_ = json.NewEncoder(w).Encode(map[string]any{"errcode": 40001, "errmsg": "bad token"})
return
}
if err := r.ParseMultipartForm(1 << 20); err != nil {
w.WriteHeader(http.StatusBadRequest)
return
}
file, _, err := r.FormFile("media")
if err == nil {
data, _ := io.ReadAll(file)
_ = file.Close()
fixture.mu.Lock()
fixture.uploadedBytes += len(data)
fixture.mu.Unlock()
}
fixture.mu.Lock()
fixture.uploadCalls++
code := fixture.uploadErrcode
fixture.mu.Unlock()
select {
case fixture.uploadedC <- struct{}{}:
default:
}
_ = json.NewEncoder(w).Encode(map[string]any{
"errcode": code, "errmsg": "ok", "media_id": cardImageTestMediaID,
})
})
fixture.server = httptest.NewServer(mux)
t.Cleanup(fixture.server.Close)
return fixture
}
func (f *cardImageTestServer) resolver(t *testing.T, mutate func(*CardImageMediaOptions)) CardImageMediaRewriter {
t.Helper()
opts := CardImageMediaOptions{
ClientID: "robot-client-id",
ClientSecret: "robot-client-secret",
ImageURLPrefix: f.server.URL + "/dingtalk/i/",
TokenURL: f.server.URL + "/gettoken",
UploadURL: f.server.URL + "/media/upload",
HTTPClient: f.server.Client(),
}
if mutate != nil {
mutate(&opts)
}
rewriter, err := NewCardImageMediaRewriter(opts)
if err != nil {
t.Fatalf("NewCardImageMediaRewriter: %v", err)
}
if rewriter == nil {
t.Fatal("expected a configured rewriter")
}
return rewriter
}
func (f *cardImageTestServer) counts() (token, upload int) {
f.mu.Lock()
defer f.mu.Unlock()
return f.tokenCalls, f.uploadCalls
}
func (f *cardImageTestServer) uploadedByteCount() int {
f.mu.Lock()
defer f.mu.Unlock()
return f.uploadedBytes
}
func (f *cardImageTestServer) imageURL(id string) string {
return f.server.URL + "/dingtalk/i/" + id
}
func (f *cardImageTestServer) waitForUpload(t *testing.T) {
t.Helper()
select {
case <-f.uploadedC:
case <-time.After(3 * time.Second):
t.Fatal("timed out waiting for the media upload")
}
}
func TestCardImageMediaRewritesOwnImageAndCachesMediaID(t *testing.T) {
fixture := newCardImageTestServer(t)
rewriter := fixture.resolver(t, nil)
content := fmt.Sprintf("看图:![设备图](%s)", fixture.imageURL(cardImageTestID))
got := rewriter.RewriteForCard(context.Background(), content, true)
want := fmt.Sprintf("看图:![设备图](%s)", cardImageTestMediaID)
if got != want {
t.Fatalf("got=%q want=%q", got, want)
}
if _, uploads := fixture.counts(); uploads != 1 {
t.Fatalf("uploads=%d want 1", uploads)
}
if got := fixture.uploadedByteCount(); got != len(fixture.imageBytes) {
t.Fatalf("uploaded bytes=%d want %d", got, len(fixture.imageBytes))
}
// A second rewrite must reuse the cached media id without another upload,
// and the token must only be fetched once.
again := rewriter.RewriteForCard(context.Background(), content, true)
if again != want {
t.Fatalf("cached rewrite got=%q want=%q", again, want)
}
tokens, uploads := fixture.counts()
if uploads != 1 || tokens != 1 {
t.Fatalf("tokens=%d uploads=%d want 1/1", tokens, uploads)
}
}
func TestCardImageMediaPartialFramesNeverBlockAndPickUpLater(t *testing.T) {
fixture := newCardImageTestServer(t)
rewriter := fixture.resolver(t, nil)
content := fmt.Sprintf("![设备图](%s)", fixture.imageURL(cardImageTestID))
// A streaming frame keeps the external URL but starts the upload.
if got := rewriter.RewriteForCard(context.Background(), content, false); got != content {
t.Fatalf("streaming frame must stay unchanged, got=%q", got)
}
fixture.waitForUpload(t)
deadline := time.Now().Add(2 * time.Second)
for {
got := rewriter.RewriteForCard(context.Background(), content, false)
if strings.Contains(got, cardImageTestMediaID) {
break
}
if time.Now().After(deadline) {
t.Fatalf("media id never became available, got=%q", got)
}
time.Sleep(10 * time.Millisecond)
}
if _, uploads := fixture.counts(); uploads != 1 {
t.Fatalf("uploads=%d want 1", uploads)
}
}
func TestCardImageMediaIgnoresForeignAndIncompleteReferences(t *testing.T) {
fixture := newCardImageTestServer(t)
rewriter := fixture.resolver(t, nil)
content := strings.Join([]string{
fmt.Sprintf("![外部图](https://example.com/dingtalk/i/%s)", cardImageTestID),
fmt.Sprintf("![流水线](%s)", fixture.imageURL("not-a-stable-id")),
fmt.Sprintf("![未闭合](%s", fixture.imageURL(cardImageTestID)),
fmt.Sprintf("纯链接 %s", fixture.imageURL(cardImageTestID)),
}, "\n")
if got := rewriter.RewriteForCard(context.Background(), content, true); got != content {
t.Fatalf("content must be untouched, got=%q", got)
}
tokens, uploads := fixture.counts()
if tokens != 0 || uploads != 0 {
t.Fatalf("tokens=%d uploads=%d want 0/0", tokens, uploads)
}
}
func TestCardImageMediaFallsBackWhenUploadFails(t *testing.T) {
fixture := newCardImageTestServer(t)
fixture.uploadErrcode = 40035
rewriter := fixture.resolver(t, nil)
content := fmt.Sprintf("![设备图](%s)", fixture.imageURL(cardImageTestID))
if got := rewriter.RewriteForCard(context.Background(), content, true); got != content {
t.Fatalf("failed upload must keep the external reference, got=%q", got)
}
}
func TestCardImageMediaFallsBackForOversizeImages(t *testing.T) {
fixture := newCardImageTestServer(t)
rewriter := fixture.resolver(t, func(opts *CardImageMediaOptions) {
opts.MaxBytes = 128
})
content := fmt.Sprintf("![大图](%s/big)", fixture.imageURL(cardImageTestID))
if got := rewriter.RewriteForCard(context.Background(), content, true); got != content {
t.Fatalf("oversize image must keep the external reference, got=%q", got)
}
if _, uploads := fixture.counts(); uploads != 0 {
t.Fatalf("uploads=%d want 0", uploads)
}
}
func TestCardImageMediaFetchUsesLoopbackPrefix(t *testing.T) {
fixture := newCardImageTestServer(t)
rewriter := fixture.resolver(t, func(opts *CardImageMediaOptions) {
opts.ImageURLPrefix = "https://weknora-api.invalid/dingtalk/i/"
opts.FetchURLPrefix = fixture.server.URL + "/images/"
})
content := fmt.Sprintf("![设备图](https://weknora-api.invalid/dingtalk/i/%s)", cardImageTestID)
got := rewriter.RewriteForCard(context.Background(), content, true)
if !strings.Contains(got, cardImageTestMediaID) {
t.Fatalf("loopback fetch must still produce a media id, got=%q", got)
}
fixture.mu.Lock()
defer fixture.mu.Unlock()
if len(fixture.fetched) != 1 || !strings.HasPrefix(fixture.fetched[0], "/images/") {
t.Fatalf("fetched=%v want one loopback path", fixture.fetched)
}
}
func TestCardImageMediaDeduplicatesConcurrentUploads(t *testing.T) {
fixture := newCardImageTestServer(t)
rewriter := fixture.resolver(t, nil)
content := fmt.Sprintf("![设备图](%s)", fixture.imageURL(cardImageTestID))
var wg sync.WaitGroup
for i := 0; i < 4; i++ {
wg.Add(1)
go func() {
defer wg.Done()
rewriter.RewriteForCard(context.Background(), content, true)
}()
}
wg.Wait()
if _, uploads := fixture.counts(); uploads != 1 {
t.Fatalf("uploads=%d want 1 for concurrent callers", uploads)
}
}
func TestCardImageMediaRewriterRequiresFullConfiguration(t *testing.T) {
for name, opts := range map[string]CardImageMediaOptions{
"missing client id": {ClientSecret: "s", ImageURLPrefix: "https://img.example/dingtalk/i/"},
"missing secret": {ClientID: "id", ImageURLPrefix: "https://img.example/dingtalk/i/"},
"missing prefix": {ClientID: "id", ClientSecret: "s"},
} {
rewriter, err := NewCardImageMediaRewriter(opts)
if err != nil || rewriter != nil {
t.Fatalf("%s: rewriter=%v err=%v want nil/nil", name, rewriter, err)
}
}
}
func TestStreamingCardDeliverySendsMediaIDAndKeepsMarkdownExternal(t *testing.T) {
fixture := newCardImageTestServer(t)
rewriter := fixture.resolver(t, nil)
answer := fmt.Sprintf("CR20V 接线如下:\n\n![工控柜接线图](%s)", fixture.imageURL(cardImageTestID))
card := &fakeStreamingCard{startID: "track-image-card"}
qa := &fakeStreamingQA{chunks: []string{answer}, result: core.QAResult{Answer: answer}}
reply := &fakeReply{}
h := newTestHandlerWithStreaming(t, qa, reply, card)
h.cardImageMedia = rewriter
if err := h.Handle(context.Background(), directTextCallback()); err != nil {
t.Fatalf("Handle: %v", err)
}
if len(card.updates) == 0 {
t.Fatal("no card update was recorded")
}
final := card.updates[len(card.updates)-1]
if !final.isFinalize {
t.Fatalf("last update must finalize the card, got %+v", final)
}
if !strings.Contains(final.content, cardImageTestMediaID) {
t.Fatalf("final card body must carry the media id, got=%q", final.content)
}
if strings.Contains(final.content, fixture.server.URL) {
t.Fatalf("final card body must not keep the external URL, got=%q", final.content)
}
}
func TestCardFinalizeFallbackKeepsExternalImageURL(t *testing.T) {
fixture := newCardImageTestServer(t)
rewriter := fixture.resolver(t, nil)
answer := fmt.Sprintf("![工控柜接线图](%s)", fixture.imageURL(cardImageTestID))
card := &fakeStreamingCard{startID: "track-image-fallback", updateErr: errors.New("card unavailable")}
qa := &fakeStreamingQA{chunks: []string{answer}, result: core.QAResult{Answer: answer}}
reply := &fakeReply{}
h := newTestHandlerWithStreaming(t, qa, reply, card)
h.cardImageMedia = rewriter
if err := h.Handle(context.Background(), directTextCallback()); err != nil {
t.Fatalf("Handle: %v", err)
}
if !reply.called {
t.Fatal("markdown fallback was not sent")
}
if !strings.Contains(reply.lastText, fixture.server.URL) {
t.Fatalf("markdown fallback must keep the external URL, got=%q", reply.lastText)
}
if strings.Contains(reply.lastText, cardImageTestMediaID) {
t.Fatalf("markdown fallback must not carry a media id, got=%q", reply.lastText)
}
}
func TestCardImageMediaFallsBackToPublicURLWhenLoopbackMisses(t *testing.T) {
fixture := newCardImageTestServer(t)
rewriter := fixture.resolver(t, func(opts *CardImageMediaOptions) {
// The loopback route exists but does not serve this object.
opts.FetchURLPrefix = fixture.server.URL + "/missing/"
})
content := fmt.Sprintf("![设备图](%s)", fixture.imageURL(cardImageTestID))
got := rewriter.RewriteForCard(context.Background(), content, true)
if !strings.Contains(got, cardImageTestMediaID) {
t.Fatalf("public fallback must still produce a media id, got=%q", got)
}
fixture.mu.Lock()
defer fixture.mu.Unlock()
if len(fixture.fetched) != 1 || !strings.HasPrefix(fixture.fetched[0], "/dingtalk/i/") {
t.Fatalf("fetched=%v want a single public fetch after the loopback miss", fixture.fetched)
}
}
func TestCardImageRefCountsSplitMediaAndExternalReferences(t *testing.T) {
content := strings.Join([]string{
fmt.Sprintf("![图一](%s)", cardImageTestMediaID),
fmt.Sprintf("![图二](https://weknora-api.dobot.cn/dingtalk/i/%s)", cardImageTestID),
fmt.Sprintf("![图三](%s)", cardImageTestMediaID),
}, "\n")
total, external, media := cardImageRefCounts(content)
if total != 3 || external != 1 || media != 2 {
t.Fatalf("total=%d external=%d media=%d want 3/1/2", total, external, media)
}
if total, external, media := cardImageRefCounts("无图回答"); total != 0 || external != 0 || media != 0 {
t.Fatalf("plain text must report no image refs, got %d/%d/%d", total, external, media)
}
}
func TestCardUpdateDiagnosticReportsImageReferenceShape(t *testing.T) {
content := fmt.Sprintf("正文\n\n![图](%s)", cardImageTestMediaID)
got := formatCardUpdateDiagnostic(content, true, false)
if !strings.Contains(got, "image_refs=1") || !strings.Contains(got, "media_image_refs=1") || !strings.Contains(got, "external_image_refs=0") {
t.Fatalf("diagnostic must report image reference shape: %q", got)
}
if strings.Contains(got, cardImageTestMediaID) {
t.Fatalf("diagnostic must not leak the media id: %q", got)
}
}
+111
View File
@@ -1289,6 +1289,117 @@ func TestOAuthAcrossStartAndUpdates(t *testing.T) {
}
}
// ---------------------------------------------------------------------------
// Test: PC wide-screen layout config is delivered through the reserved
// cardParamMap system field, never as a sibling of cardParamMap
// ---------------------------------------------------------------------------
func TestCardWideScreenAutoLayoutParam(t *testing.T) {
cases := []struct {
name string
req CardRequest
}{
{
name: "direct",
req: CardRequest{
ConversationType: core.ConversationDirect,
ExternalUserID: "user-456",
Content: "Hello from AI",
},
},
{
name: "group",
req: CardRequest{
ConversationType: core.ConversationGroup,
ExternalChatID: "chat-789",
ExternalUserID: "user-456",
Content: "Hello from AI",
},
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
ts := newCardTestServer()
defer ts.Close()
sender, err := newCardSenderForTest("id", "secret", "tpl", ts.Client(), ts.URL()+"/token", ts.URL()+"/create")
if err != nil {
t.Fatal(err)
}
if _, err := sender.StartCard(context.Background(), tc.req); err != nil {
t.Fatalf("StartCard: %v", err)
}
cardData, ok := ts.lastCreateBody["cardData"].(map[string]any)
if !ok {
t.Fatalf("cardData missing or wrong type: %#v", ts.lastCreateBody["cardData"])
}
// Regression guard: a complex-typed value placed next to
// cardParamMap leaves the template variable unassigned and the
// card silently renders in the default narrow layout.
for _, badKey := range []string{"config", cardWideScreenJSONKey} {
if _, present := cardData[badKey]; present {
t.Fatalf("%q must not be a sibling of cardParamMap: %#v", badKey, cardData)
}
}
paramMap, ok := cardData["cardParamMap"].(map[string]any)
if !ok {
t.Fatalf("cardParamMap missing or wrong type: %#v", cardData["cardParamMap"])
}
raw, ok := paramMap[cardWideScreenJSONKey].(string)
if !ok {
t.Fatalf("%s must be a JSON string, got %#v", cardWideScreenJSONKey, paramMap[cardWideScreenJSONKey])
}
var decoded struct {
Config struct {
AutoLayout bool `json:"autoLayout"`
} `json:"config"`
}
if err := json.Unmarshal([]byte(raw), &decoded); err != nil {
t.Fatalf("%s is not valid JSON (%q): %v", cardWideScreenJSONKey, raw, err)
}
if !decoded.Config.AutoLayout {
t.Fatalf("config.autoLayout=false in %q, want true", raw)
}
// Plain string variables must keep going through cardParamMap directly.
if paramMap["content"] != tc.req.Content {
t.Errorf("content=%v, want %v", paramMap["content"], tc.req.Content)
}
})
}
}
// ---------------------------------------------------------------------------
// Test: streaming updates only carry content (no layout config)
// ---------------------------------------------------------------------------
func TestCardWideScreenNotSentOnStreamingUpdate(t *testing.T) {
ts := newCardTestServer()
defer ts.Close()
sender, err := newCardSenderForTest("id", "secret", "tpl", ts.Client(), ts.URL()+"/token", ts.URL()+"/create")
if err != nil {
t.Fatal(err)
}
outTrackID, err := sender.StartCard(context.Background(), CardRequest{
ConversationType: core.ConversationDirect,
ExternalUserID: "user-456",
Content: "placeholder",
})
if err != nil {
t.Fatalf("StartCard: %v", err)
}
if err := sender.UpdateCard(context.Background(), outTrackID, "streamed chunk", false, false); err != nil {
t.Fatalf("UpdateCard: %v", err)
}
if _, present := ts.lastStreamBody[cardWideScreenJSONKey]; present {
t.Fatalf("streaming body must not carry %s: %#v", cardWideScreenJSONKey, ts.lastStreamBody)
}
if ts.lastStreamBody["key"] != "content" {
t.Errorf("stream key=%v, want content", ts.lastStreamBody["key"])
}
}
// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------
@@ -8,7 +8,11 @@ import (
const displayReferenceMarkerPattern = `\[\[ref[\t ]+id[\t ]*=[\t ]*["']c[0-9]+["'][\t ]*\]\]`
var (
displayReferenceMarkerRE = regexp.MustCompile(displayReferenceMarkerPattern)
displayReferenceMarkerRE = regexp.MustCompile(displayReferenceMarkerPattern)
// bareInternalSourceHandleRE matches WeKnora request-local source aliases
// leaked by a model (for example [d21], [c12], or [w3]). It intentionally
// does not match ordinary Markdown links or arbitrary bracketed content.
bareInternalSourceHandleRE = regexp.MustCompile(`(?i)\[(?:[dcw][0-9]+)(?:[ ]*[,,][ ]*(?:[dcw][0-9]+))*\]`)
leadingReferenceGroupColonRE = regexp.MustCompile(`(?m)^[ ]*(?:` + displayReferenceMarkerPattern + `[ ]*)+[::][ ]*`)
spacedReferenceGroupRE = regexp.MustCompile(`[ ]+(?:` + displayReferenceMarkerPattern + `[ ]*)+`)
spaceBeforeDisplayPunctuation = regexp.MustCompile(`[ ]+([,。;:!?、,.;:!?])`)
@@ -19,12 +23,13 @@ func sanitizeDisplayAnswer(content string) string {
if index := strings.LastIndex(content, "[[ref"); index >= 0 && !strings.Contains(content[index:], "]]") {
content = content[:index]
}
if !displayReferenceMarkerRE.MatchString(content) {
if !displayReferenceMarkerRE.MatchString(content) && !bareInternalSourceHandleRE.MatchString(content) {
return content
}
content = leadingReferenceGroupColonRE.ReplaceAllString(content, "")
content = spacedReferenceGroupRE.ReplaceAllString(content, " ")
content = displayReferenceMarkerRE.ReplaceAllString(content, "")
content = bareInternalSourceHandleRE.ReplaceAllString(content, "")
content = spaceBeforeDisplayPunctuation.ReplaceAllString(content, "$1")
content = spaceAfterDisplayColon.ReplaceAllString(content, "$1")
return content
@@ -34,10 +34,20 @@ func TestSanitizeDisplayAnswerRemovesInternalReferenceMarkers(t *testing.T) {
want: "Feature A remains supported.",
},
{
name: "unrelated double bracket syntax is preserved",
name: "unrelated double bracket syntax and ordinary link are preserved",
input: "请保留 [[wiki|页面名称]] 和 [普通链接](https://example.com)。",
want: "请保留 [[wiki|页面名称]] 和 [普通链接](https://example.com)。",
},
{
name: "bare internal source handles",
input: "Nova支持虚拟控制器离线仿真[d21],并可搭配VX500[w3]。",
want: "Nova支持虚拟控制器离线仿真,并可搭配VX500。",
},
{
name: "grouped bare internal source handles",
input: "该结论由多个资料支持 [d21, c12,w3]。",
want: "该结论由多个资料支持。",
},
{
name: "content without references remains byte for byte unchanged",
input: "Protocol: value ; keep spacing\n indented",
@@ -30,4 +30,4 @@ func TestEffectiveDownloadCodesEmpty(t *testing.T) {
if codes := effectiveDownloadCodes(BotCallback{}); len(codes) != 0 {
t.Fatalf("codes=%v, want empty", codes)
}
}
}
+86 -16
View File
@@ -11,6 +11,7 @@ import (
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/directory"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/outboundtext"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/reasoningmode"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/weknora"
)
@@ -157,9 +158,22 @@ type HandlerOptions struct {
CardSender FinalCardSender
Directory UserDirectory
OutboundRewriter OutboundContentRewriter
// CardImageMedia rewrites our own published image URLs in card bodies into
// DingTalk media ids; nil keeps external URLs as-is.
CardImageMedia CardImageMediaRewriter
SettingsRouter SettingsRouter
ReasoningRouter ReasoningRouter
LanguageResolver LanguagePreferenceResolver
// OutboundRewriteTimeout bounds the outbound image rewrite stage (fetch +
// publish). Answers carrying many large screenshots need more than the
// delivery timeout, otherwise the remaining images are dropped. Zero keeps
// finalDeliveryTimeout.
OutboundRewriteTimeout time.Duration
// WideTableGrouping selects how over-wide answer tables are adapted, per
// frame: "" or "on" groups on every frame (historical), "off" never groups,
// "stream" groups only while the answer is still streaming and keeps the
// real table on the final frame. See tableGroupingForMode.
WideTableGrouping string
}
type Handler struct {
@@ -171,12 +185,22 @@ type Handler struct {
card FinalCardSender
directory UserDirectory
outboundRewriter OutboundContentRewriter
cardImageMedia CardImageMediaRewriter
settingsRouter SettingsRouter
reasoningRouter ReasoningRouter
languageResolver LanguagePreferenceResolver
now func() time.Time
cardUpdateInterval time.Duration
finalDeliveryTimeout time.Duration
// outboundRewriteTimeout applies to the image rewrite stage only; zero
// falls back to finalDeliveryTimeout.
outboundRewriteTimeout time.Duration
// groupTablesPartial / groupTablesFinal decide per frame whether an
// over-wide table becomes a grouped list. Grouped lists survive the
// streaming frames where a table is still incomplete; the final frame can
// keep the real table, which the PC wide-screen card lays out itself.
groupTablesPartial bool
groupTablesFinal bool
}
// NewHandler accepts at most one image downloader. A nil/omitted downloader
@@ -208,6 +232,8 @@ func NewHandlerWithOptions(cfg TenantConfig, qa QAAsker, reply MarkdownSender, o
if reply == nil {
return nil, errors.New("dingtalk handler: markdown sender is required")
}
groupPartial, groupFinal := tableGroupingForMode(opts.WideTableGrouping)
return &Handler{
cfg: cfg,
qa: qa,
@@ -217,12 +243,17 @@ func NewHandlerWithOptions(cfg TenantConfig, qa QAAsker, reply MarkdownSender, o
card: opts.CardSender,
directory: opts.Directory,
outboundRewriter: opts.OutboundRewriter,
cardImageMedia: opts.CardImageMedia,
settingsRouter: opts.SettingsRouter,
reasoningRouter: opts.ReasoningRouter,
languageResolver: opts.LanguageResolver,
now: time.Now,
cardUpdateInterval: 500 * time.Millisecond,
finalDeliveryTimeout: defaultHTTPTimeout,
outboundRewriteTimeout: opts.OutboundRewriteTimeout,
groupTablesPartial: groupPartial,
groupTablesFinal: groupFinal,
}, nil
}
@@ -440,7 +471,7 @@ func (h *Handler) handleNonStreaming(ctx context.Context, webhookURL string, req
ConversationType: req.ConversationType,
ExternalUserID: req.ExternalUserID,
ExternalChatID: req.ExternalChatID,
Content: content,
Content: h.rewriteCardImages(ctx, content, true),
}
if outTrackID, cardErr := h.card.SendFinalCard(ctx, cardReq); cardErr == nil {
h.recordDelivery(ctx, result, core.DeliverySent, core.DeliveryMethodCard, outTrackID, "")
@@ -490,14 +521,16 @@ func (h *Handler) handleStreaming(ctx context.Context, webhookURL string, req co
// Streaming observer with the configured update throttle (500ms by default).
var lastUpdate time.Time
var lastPartialContent string
feed := reasoningmode.NewPartialFeed(req.ReasoningExecution)
observer := func(fullContent string) {
if strings.TrimSpace(fullContent) == "" {
visible := feed.OnAnswer(fullContent)
if strings.TrimSpace(visible) == "" {
return
}
fullContent = decorateReasoningPartial(fullContent, req.ReasoningExecution)
fullContent = decorateReasoningPartial(visible, req.ReasoningExecution)
now := h.now()
if lastUpdate.IsZero() || now.Sub(lastUpdate) >= h.cardUpdateInterval {
partialContent := h.sanitizeOutboundPartial(fullContent)
partialContent := h.sanitizeOutboundPartial(ctx, fullContent)
if strings.TrimSpace(partialContent) == "" || partialContent == lastPartialContent {
return
}
@@ -551,17 +584,26 @@ func (h *Handler) handleEventStreaming(ctx context.Context, webhookURL string, r
pump := newCardUpdatePump(ctx, h.cardUpdateInterval, func(updateCtx context.Context, content string) error {
return streamingCard.UpdateCard(updateCtx, outTrackID, content, false, false)
})
answerStarted := false
feed := reasoningmode.NewPartialFeed(req.ReasoningExecution)
result, err := eventQA.AskEvents(ctx, req, func(event weknora.StreamEvent) {
if event.ResponseType == "answer" && strings.TrimSpace(event.FullAnswer) != "" {
answerStarted = true
visibleAnswer := decorateReasoningPartial(event.FullAnswer, req.ReasoningExecution)
if partialContent := h.sanitizeOutboundPartial(visibleAnswer); strings.TrimSpace(partialContent) != "" {
if event.ResponseType == "answer" {
visible := feed.OnAnswer(event.FullAnswer)
if strings.TrimSpace(visible) == "" {
return
}
visibleAnswer := decorateReasoningPartial(visible, req.ReasoningExecution)
if partialContent := h.sanitizeOutboundPartial(ctx, visibleAnswer); strings.TrimSpace(partialContent) != "" {
pump.Offer(partialContent)
}
return
}
if answerStarted {
switch event.ResponseType {
case "tool_call", "tool_result", "thinking", "reflection":
// A reasoning step ended: the next narrative segment starts a new
// line so the card shows the reasoning process line by line.
feed.StepBoundary()
}
if feed.Started() {
return
}
if content := renderProgressEvent(event); content != "" {
@@ -602,7 +644,13 @@ func (h *Handler) finalizeCardOrMarkdown(ctx context.Context, webhookURL string,
// Final card convergence must still be attempted when the QA context was
// canceled or timed out; otherwise the card can remain "processing".
finalCtx, cancelFinal := context.WithTimeout(context.WithoutCancel(ctx), timeout)
updateErr := streamingCard.UpdateCard(finalCtx, outTrackID, content, true, isError)
cardContent := content
if !isError {
// Card bodies render DingTalk media ids only, so the images are
// converted here while the markdown fallback keeps the external URLs.
cardContent = h.rewriteCardImages(finalCtx, content, true)
}
updateErr := streamingCard.UpdateCard(finalCtx, outTrackID, cardContent, true, isError)
cancelFinal()
if updateErr == nil {
if !isError {
@@ -628,24 +676,46 @@ func (h *Handler) finalizeCardOrMarkdown(ctx context.Context, webhookURL string,
}
func (h *Handler) rewriteOutboundContent(ctx context.Context, content string) string {
content = outboundtext.NormalizeMath(content)
options := tableNormalizeOptions{groupWideTables: h.groupTablesFinal}
if h.outboundRewriter == nil {
return sanitizeDisplayAnswer(content)
return normalizeMarkdownTables(sanitizeDisplayAnswer(content), options)
}
timeout := h.outboundRewriteTimeout
if timeout <= 0 {
timeout = h.finalDeliveryTimeout
}
timeout := h.finalDeliveryTimeout
if timeout <= 0 {
timeout = defaultHTTPTimeout
}
rewriteCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), timeout)
defer cancel()
return sanitizeDisplayAnswer(h.outboundRewriter.Rewrite(rewriteCtx, content))
return normalizeMarkdownTables(sanitizeDisplayAnswer(h.outboundRewriter.Rewrite(rewriteCtx, content)), options)
}
func (h *Handler) sanitizeOutboundPartial(content string) string {
func (h *Handler) sanitizeOutboundPartial(ctx context.Context, content string) string {
content = outboundtext.NormalizeMath(content)
sanitizer, ok := h.outboundRewriter.(OutboundPartialSanitizer)
if ok {
content = sanitizer.SanitizePartial(content)
}
return stableMarkdownPrefix(sanitizeDisplayAnswer(content))
options := tableNormalizeOptions{groupWideTables: h.groupTablesPartial}
content = normalizeMarkdownTables(sanitizeDisplayAnswer(content), options)
return h.rewriteCardImages(ctx, stableMarkdownPrefix(content), false)
}
// rewriteCardImages converts image references in card content into DingTalk
// media ids. It is a no-op when the feature is not configured and never fails
// the caller: unuploadable images keep their original external reference.
func (h *Handler) rewriteCardImages(ctx context.Context, content string, wait bool) string {
if h.cardImageMedia == nil || strings.TrimSpace(content) == "" {
return content
}
rewritten := h.cardImageMedia.RewriteForCard(ctx, content, wait)
if strings.TrimSpace(rewritten) == "" {
return content
}
return rewritten
}
func (h *Handler) recordDelivery(ctx context.Context, result core.QAResult, status, method, providerMessageID, failureCode string) {
@@ -77,4 +77,4 @@ func TestPublishQuestionImageFailsClosedWithoutURL(t *testing.T) {
if got := h2.publishQuestionImage(context.Background(), image); got != "" {
t.Fatalf("expected empty URL on publish error, got %q", got)
}
}
}
@@ -0,0 +1,40 @@
package dingtalk
import (
"context"
"strings"
"testing"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/outboundimage"
)
type diagFetcher struct{}
func (diagFetcher) Fetch(ctx context.Context, ref string) ([]byte, string, error) {
return []byte("PNGDATA"), "image/png", nil
}
type diagPublisher struct{}
func (diagPublisher) Publish(ctx context.Context, data []byte, contentType string) (string, error) {
return "https://weknora-api-test.dobot.cn/dingtalk/i/a87f73b8-15cb-546e-ba53-bf93dade916f", nil
}
// The outbound rewrite must publish a resource image and keep it in the card
// body. A reference whose handle contains "_x" (…ZM_pltQ) used to be mangled by
// the math normalizer, which made the image unpublishable and the sanitizer
// drop the whole line.
func TestRewriteOutboundContentKeepsPublishedImage(t *testing.T) {
content := "![CR10A 末端负载曲线](resource://efVTXx0gY9Lk2vuZM_pltQ)\n\n公式:$r = \\sqrt{X^2 + Y^2}$\n"
h := &Handler{
outboundRewriter: outboundimage.NewRewriter(diagFetcher{}, diagPublisher{}, 50),
groupTablesFinal: true,
}
out := h.rewriteOutboundContent(context.Background(), content)
if !strings.Contains(out, "![CR10A 末端负载曲线](https://weknora-api-test.dobot.cn/dingtalk/i/") {
t.Fatalf("published image missing from card content: %q", out)
}
if !strings.Contains(out, "r = √(X² + Y²)") {
t.Fatalf("formula not converted: %q", out)
}
}
@@ -1,6 +1,7 @@
package dingtalk
import (
"strings"
"testing"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/reasoningmode"
@@ -19,3 +20,38 @@ func TestDecorateReasoningPartialLeavesQuickAnswerUnchanged(t *testing.T) {
t.Fatalf("got=%q", got)
}
}
func TestReasoningStreamingShowsStepsLineByLineThenDropsThemForBody(t *testing.T) {
execution := string(reasoningmode.ExecutionReasoning)
first := "我先确认一下设备的型号和手册。"
second := "手册里有关节扭矩表,让我核对一下。"
body := "## CR20V 关节电机最大旋转扭矩\n\n| 关节 | 扭矩 |"
feed := reasoningmode.NewPartialFeed(execution)
if got, want := decorateReasoningPartial(feed.OnAnswer(first), execution), "【智能推理模式】\n\n"+first; got != want {
t.Fatalf("step1 got=%q want=%q", got, want)
}
feed.StepBoundary()
if got, want := decorateReasoningPartial(feed.OnAnswer(first+second), execution), "【智能推理模式】\n\n"+first+"\n\n"+second; got != want {
t.Fatalf("step2 got=%q want=%q", got, want)
}
answer := first + second + "\n\n" + body
if got, want := decorateReasoningPartial(feed.OnAnswer(answer), execution), "【智能推理模式】\n\n"+body; got != want {
t.Fatalf("refresh got=%q want=%q", got, want)
}
got := feed.OnAnswer(answer + "\n| J1 | 98 N·m |")
if strings.Contains(got, first) || !strings.Contains(got, "J1") {
t.Fatalf("reasoning must stay hidden while the body streams, got=%q", got)
}
}
func TestQuickStreamingKeepsAccumulatedAnswer(t *testing.T) {
feed := reasoningmode.NewPartialFeed(string(reasoningmode.ExecutionQuick))
if got := feed.OnAnswer("快速答案"); got != "快速答案" {
t.Fatalf("got=%q", got)
}
feed.StepBoundary()
if got := feed.OnAnswer("快速答案追加"); got != "快速答案追加" {
t.Fatalf("got=%q", got)
}
}
@@ -0,0 +1,255 @@
package dingtalk
import (
"strings"
"unicode"
)
// maxTableCellVisualWidth is the DingTalk-safe cell budget. CJK counts as 2
// display columns; a cell wider than this tends to collapse the column layout
// (one character per line) in the streaming card renderer.
const maxTableCellVisualWidth = 32
// tableNormalizeOptions controls how a final answer is adapted to the card
// renderer. The zero value keeps the historical behaviour.
type tableNormalizeOptions struct {
// groupWideTables rewrites tables containing over-wide cells into grouped
// lists (bold row heading + "- header:value" lines). Disable it when the
// client lays out wide tables itself, so the answer keeps its real table.
groupWideTables bool
}
func defaultTableNormalizeOptions() tableNormalizeOptions {
return tableNormalizeOptions{groupWideTables: true}
}
// tableGroupingForMode maps the configured mode to the two frames the card
// renders. An unknown or empty mode keeps the historical behaviour (group on
// every frame), so an unset environment never changes rendering.
//
// ""/"on" group on every frame
// "off" never group: real tables everywhere
// "stream" group while streaming, real table on the final frame
func tableGroupingForMode(mode string) (partial, final bool) {
switch strings.ToLower(strings.TrimSpace(mode)) {
case "off", "0", "false", "no":
return false, false
case "stream", "streaming", "partial":
return true, false
default:
return true, true
}
}
// normalizeWideMarkdownTables makes final answers render reliably on DingTalk
// cards without changing the knowledge itself. It (1) replaces <br> inside
// table cells with a full-width separator, and (2) rewrites any table that
// still contains an over-wide cell into a grouped list — a bold row heading
// followed by one "- header:value" line per column. Compact tables pass
// through unchanged. Tables inside code fences are never touched.
func normalizeWideMarkdownTables(content string) string {
return normalizeMarkdownTables(content, defaultTableNormalizeOptions())
}
func normalizeMarkdownTables(content string, options tableNormalizeOptions) string {
if !strings.Contains(content, "|") {
return content
}
lines := splitMarkdownStreamLines(content)
out := make([]string, 0, len(lines))
for i := 0; i < len(lines); {
line := lines[i]
if line.inFence {
out = append(out, line.text)
i++
continue
}
if isTableHeaderCandidate(line.text) && i+1 < len(lines) &&
isCompleteTableDelimiter(lines[i+1].text) {
end := i + 1
for end+1 < len(lines) && !lines[end+1].inFence &&
strings.HasPrefix(strings.TrimSpace(lines[end+1].text), "|") {
end++
}
out = append(out, normalizeTableBlock(lines[i:end+1], options)...)
i = end + 1
continue
}
out = append(out, line.text)
i++
}
return strings.Join(out, "\n")
}
func normalizeTableBlock(rows []markdownStreamLine, options tableNormalizeOptions) []string {
cells := make([][]string, 0, len(rows))
for idx, row := range rows {
if idx == 1 { // delimiter row
continue
}
cells = append(cells, splitTableRow(row.text))
}
maxWidth := 0
for _, row := range cells {
for _, cell := range row {
if w := visualWidth(cell); w > maxWidth {
maxWidth = w
}
}
}
header := cells[0]
if maxWidth <= maxTableCellVisualWidth || !options.groupWideTables {
// Compact tables keep their exact original bytes unless a cell carries
// <br>, which the card renderer mishandles inside cells. The same path
// serves over-wide tables when grouping is disabled: the table stays a
// table and only <br> is replaced.
if !blockHasBreak(rows) {
rebuilt := make([]string, len(rows))
for i, row := range rows {
rebuilt[i] = row.text
}
return rebuilt
}
var rebuilt []string
for idx, row := range cells {
joined := "| " + strings.Join(row, " | ") + " |"
if idx == 0 {
rebuilt = append(rebuilt, joined)
delims := make([]string, len(row))
for c := range row {
delims[c] = "---"
}
rebuilt = append(rebuilt, "| "+strings.Join(delims, " | ")+" |")
} else {
rebuilt = append(rebuilt, joined)
}
}
return rebuilt
}
var out []string
for gi, row := range cells[1:] {
if gi > 0 {
out = append(out, "")
}
var title string
var rest []string
for c, cell := range row {
if c == 0 {
title = cell
continue
}
label := strings.TrimSpace(header[min(c, len(header)-1)])
if label == "" {
label = "第" + itoa(c+1) + "项"
}
if cell == "" {
continue
}
rest = append(rest, "- "+label+":"+cell)
}
if title == "" {
title = label2(row[0])
}
out = append(out, "**"+strings.Trim(title, "* ")+"**")
out = append(out, rest...)
}
return out
}
func blockHasBreak(rows []markdownStreamLine) bool {
for _, row := range rows {
lower := strings.ToLower(row.text)
if strings.Contains(lower, "<br") {
return true
}
}
return false
}
func splitTableRow(line string) []string {
trimmed := strings.TrimSpace(line)
trimmed = strings.TrimPrefix(trimmed, "|")
trimmed = strings.TrimSuffix(trimmed, "|")
parts := splitEscaped(trimmed)
for i := range parts {
cell := strings.TrimSpace(parts[i])
cell = strings.ReplaceAll(cell, "<br>", ";")
cell = strings.ReplaceAll(cell, "<br/>", ";")
cell = strings.ReplaceAll(cell, "<br />", ";")
cell = strings.ReplaceAll(cell, "\\|", "|")
parts[i] = strings.TrimSpace(cell)
}
return parts
}
// splitEscaped splits on unescaped pipe separators.
func splitEscaped(s string) []string {
var parts []string
var b strings.Builder
for i := 0; i < len(s); i++ {
switch s[i] {
case '\\':
if i+1 < len(s) && s[i+1] == '|' {
b.WriteString("\\|")
i++
continue
}
b.WriteByte(s[i])
case '|':
parts = append(parts, b.String())
b.Reset()
default:
b.WriteByte(s[i])
}
}
parts = append(parts, b.String())
return parts
}
func label2(cell string) string { return strings.Trim(cell, "* ") }
func min(a, b int) int {
if a < b {
return a
}
return b
}
func itoa(n int) string {
if n == 0 {
return "0"
}
var d []byte
for n > 0 {
d = append([]byte{byte('0' + n%10)}, d...)
n /= 10
}
return string(d)
}
func visualWidth(s string) int {
w := 0
for _, r := range s {
if r == '\n' {
if w > 0 {
w++ // separator counted conservatively
}
continue
}
if unicode.Is(unicode.Han, r) || unicode.Is(unicode.Katakana, r) ||
unicode.Is(unicode.Hiragana, r) || fullwidthPunct(r) {
w += 2
continue
}
w++
}
return w
}
func fullwidthPunct(r rune) bool {
switch r {
case ',', '。', ':', ';', '(', ')', '、', '【', '】', '《', '》', '“', '”', '‘', '’', ' ':
return true
}
return false
}
@@ -0,0 +1,167 @@
package dingtalk
import (
"context"
"strings"
"testing"
)
func TestNormalizeKeepsCompactTableUnchanged(t *testing.T) {
content := "重量如下:\n\n| 产品型号 | 净重 (本体重量) | 毛重 (含包装) |\n| :--- | :--- | :--- |\n| CR3A | 16.5 kg / 17.5 kg | 28.5 kg |\n| CR5A | 25 kg | 37 kg |\n\n结束"
got := normalizeWideMarkdownTables(content)
if got != content {
t.Fatalf("compact table must pass through untouched, got %q", got)
}
}
func TestNormalizeConvertsWideTableToGroupedList(t *testing.T) {
longCell := "J1: 180°/s (CR3A/5A/7A), 150°/s (CR10A/12A/16A)<br>J2: 180°/s (CR3A/5A/7A)"
content := "差异如下:\n\n| 对比项 | CR 系列 | CRA 系列 |\n| :--- | :--- | :--- |\n| **关节最大速度** | " + longCell + " | J1: 180°/s (CR3A/5A/7A), 150°/s (CR10A/12A/16A)<br>J2: 更多参数说明 |\n| 控制柜 | CC162 | CC262 |\n\n结论"
got := normalizeWideMarkdownTables(content)
if strings.Contains(got, "| 对比项") || strings.Contains(got, "<br>") {
t.Fatalf("wide table must be converted and <br> removed, got %q", got)
}
if !strings.HasPrefix(got, "差异如下:\n\n") || !strings.HasSuffix(got, "\n\n结论") {
t.Fatalf("surrounding text must be preserved, got %q", got)
}
// compact row (控制柜) from the same table is now a list entry too
if !strings.Contains(got, "**关节最大速度**\n- CR 系列:") || !strings.Contains(got, "- CRA 系列:") {
t.Fatalf("row key must become bold heading with header-labelled values, got %q", got)
}
if !strings.Contains(got, "**控制柜**\n- CR 系列:CC162\n- CRA 系列:CC262") {
t.Fatalf("every row must become a group, got %q", got)
}
if strings.Contains(got, "J1: 180°/s") == false {
t.Fatalf("cell content must be preserved, got %q", got)
}
}
func TestNormalizeReplacesBrKeepsTableWhenCellsShort(t *testing.T) {
content := "| A | B |\n| --- | --- |\n| 短 | 行一<br>行二 |"
got := normalizeWideMarkdownTables(content)
if strings.Contains(got, "<br>") {
t.Fatalf("<br> must be replaced, got %q", got)
}
if !strings.Contains(got, "| A | B |") || !strings.Contains(got, "行一;行二") {
t.Fatalf("short table must stay a table with ; separator, got %q", got)
}
}
func TestNormalizeSkipsFencedTables(t *testing.T) {
long := strings.Repeat("长", 30)
content := "```\n| 对比 | " + long + " |\n| --- | --- |\n| a | b |\n```\n正文"
got := normalizeWideMarkdownTables(content)
if got != content {
t.Fatalf("fenced content must never be rewritten, got %q", got)
}
}
func TestNormalizeKeepsWideTableWhenGroupingDisabled(t *testing.T) {
longCell := "J1: 180°/s (CR3A/5A/7A), 150°/s (CR10A/12A/16A)<br>J2: 180°/s (CR3A/5A/7A)"
content := "差异如下:\n\n| 对比项 | CR 系列 | CRA 系列 |\n| :--- | :--- | :--- |\n| **关节最大速度** | " + longCell + " | J1: 180°/s (CR3A/5A/7A)<br>J2: 更多参数说明 |\n| 控制柜 | CC162 | CC262 |\n\n结论"
options := defaultTableNormalizeOptions()
options.groupWideTables = false
got := normalizeMarkdownTables(content, options)
if !strings.Contains(got, "| 对比项 | CR 系列 | CRA 系列 |") || !strings.Contains(got, "| **关节最大速度** |") {
t.Fatalf("wide table must stay a table when grouping is disabled, got %q", got)
}
if strings.Contains(got, "**关节最大速度**\n- CR 系列:") {
t.Fatalf("grouped list must not be produced when grouping is disabled, got %q", got)
}
if strings.Contains(got, "<br>") {
t.Fatalf("<br> must still be replaced inside cells, got %q", got)
}
if !strings.Contains(got, ";") {
t.Fatalf("cell line breaks must become the full-width separator, got %q", got)
}
if !strings.HasPrefix(got, "差异如下:\n\n") || !strings.HasSuffix(got, "\n\n结论") {
t.Fatalf("surrounding text must be preserved, got %q", got)
}
}
func TestTableGroupingForMode(t *testing.T) {
cases := []struct {
mode string
wantPartial bool
wantFinal bool
}{
{mode: "", wantPartial: true, wantFinal: true},
{mode: "on", wantPartial: true, wantFinal: true},
{mode: "weird", wantPartial: true, wantFinal: true},
{mode: "off", wantPartial: false, wantFinal: false},
{mode: "NO", wantPartial: false, wantFinal: false},
{mode: "stream", wantPartial: true, wantFinal: false},
{mode: " Stream ", wantPartial: true, wantFinal: false},
}
for _, tc := range cases {
partial, final := tableGroupingForMode(tc.mode)
if partial != tc.wantPartial || final != tc.wantFinal {
t.Fatalf("tableGroupingForMode(%q) = (%v,%v), want (%v,%v)", tc.mode, partial, final, tc.wantPartial, tc.wantFinal)
}
}
}
// TestWideTableGroupingPerFrame proves the switch reaches both frames the card
// renders: the streaming frames and the final one. The streaming path drops
// unstable table blocks, so with grouping off a partial frame simply carries no
// table at all.
func TestWideTableGroupingPerFrame(t *testing.T) {
wide := "| 对比项 | CR 系列 | CRA 系列 |\n| :--- | :--- | :--- |\n| **关节最大速度** | " +
strings.Repeat("J1: 180°/s (CR3A/5A/7A) ", 6) + " | CC262 |"
const tableMarker, groupMarker = "| 对比项 |", "- CR 系列:"
cases := []struct {
name string
mode string
partialWant [2]bool // [grouped, table]
finalWant [2]bool // [grouped, table]
}{
{name: "default groups every frame", mode: "", partialWant: [2]bool{true, false}, finalWant: [2]bool{true, false}},
{name: "off keeps the table on the final frame", mode: "off", partialWant: [2]bool{false, false}, finalWant: [2]bool{false, true}},
{name: "stream groups partial and keeps the final table", mode: "stream", partialWant: [2]bool{true, false}, finalWant: [2]bool{false, true}},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
h, err := NewHandlerWithOptions(
TenantConfig{TenantKey: "tenant-a", Channel: "dingtalk"},
&fakeQA{},
&fakeReply{},
HandlerOptions{WideTableGrouping: tc.mode},
)
if err != nil {
t.Fatalf("NewHandlerWithOptions: %v", err)
}
frames := map[string]struct {
got string
want [2]bool
}{
"partial": {got: h.sanitizeOutboundPartial(context.Background(), wide), want: tc.partialWant},
"final": {got: h.rewriteOutboundContent(context.Background(), wide), want: tc.finalWant},
}
for name, frame := range frames {
grouped := strings.Contains(frame.got, groupMarker)
table := strings.Contains(frame.got, tableMarker)
if grouped != frame.want[0] || table != frame.want[1] {
t.Fatalf("%s frame: grouped=%v table=%v, want grouped=%v table=%v, got %q",
name, grouped, table, frame.want[0], frame.want[1], frame.got)
}
}
})
}
}
func TestNormalizeHandlesEscapedPipesAndEmptyHeaders(t *testing.T) {
long := strings.Repeat("x", 40)
content := "| 项 | | \n| --- | --- |\n| a\\|b | " + long + " |"
got := normalizeWideMarkdownTables(content)
if strings.Contains(got, "| a\\|b") {
t.Fatalf("escaped pipe cell should be converted out of table form, got %q", got)
}
if !strings.Contains(got, "a|b") {
t.Fatalf("escaped pipe must be restored as literal, got %q", got)
}
if !strings.Contains(got, "第2项:") {
t.Fatalf("missing header should fall back to ordinal label, got %q", got)
}
}
+35 -59
View File
@@ -6,36 +6,11 @@ import (
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
)
// noAnswerPhrases lists explicit "no answer" phrases that the knowledge base
// returns when it cannot find relevant information. These are matched as
// substrings after trimming whitespace.
var noAnswerPhrases = []string{
"未找到",
"没有找到",
"无法确定",
"暂无相关",
"暂无已发布的权威知识",
"知识库中没有",
"产品不存在于资料中",
}
// retrievalGapPhrases are only considered a no-answer signal when the answer
// also directs the user to a human channel. Requiring both signals avoids
// misclassifying valid negative results such as "未查询到异常记录".
var retrievalGapPhrases = []string{
"未查询到",
"未检索到",
"未能查询到",
"未能检索到",
}
var humanEscalationPhrases = []string{
"建议您联系",
"建议联系",
"请联系",
"联系技术支持",
"联系官方销售",
}
// canonicalNoAnswer is the only response that carries the explicit-no-answer
// machine contract. It must be the complete answer, never a substring match:
// an otherwise supported answer may legitimately say that one specific detail
// was not found.
const canonicalNoAnswer = "当前暂无已发布的权威知识,暂不建议操作。"
// Evaluator classifies a successful turn into a coverage verdict.
type Evaluator struct{}
@@ -53,20 +28,20 @@ func (e *Evaluator) Version() string {
// Evaluate classifies a turn. It rejects non-succeeded turns. The evaluation
// rules (in priority order):
// 1. Failed/processing turns → ErrTurnNotSucceeded
// 2. Answer contains an explicit "no answer" phrase → uncovered_explicit_no_answer
// 3. Image turns (image media present) are classified by whether the question
// 2. Has references → covered_has_references (references are the coverage
// evidence; wording in a supported answer never overrides them)
// 3. Complete canonical no-answer response with no references
// → uncovered_explicit_no_answer
// 4. Image turns (image media present) are classified by whether the question
// actually triggered a knowledge-base search:
// - has references → covered_has_references
// - no references + knowledge searched → uncovered_no_references
// - no references + no knowledge search → covered_visual_recognition
// (pure visual recognition is not a coverage miss)
// 4. Text turns with zero references are classified by whether the question
// 5. Text turns with zero references are classified by whether the question
// actually triggered a knowledge-base search:
// - knowledge searched → uncovered_no_references
// - no knowledge search → covered_non_knowledge_response
// 5. Has references → covered_has_references
//
// v1 never uses a score threshold. TopScore captures the highest reference score
// v5 never uses a score threshold. TopScore captures the highest reference score
// (0 if no references).
func (e *Evaluator) Evaluate(turn core.Turn) (Decision, error) {
if turn.Status != "succeeded" {
@@ -76,9 +51,25 @@ func (e *Evaluator) Evaluate(turn core.Turn) (Decision, error) {
// Normalize answer for no-answer phrase detection.
trimmed := strings.TrimSpace(turn.Answer)
// Check for explicit no-answer phrases first — this takes priority
// over reference presence.
if isNoAnswer(trimmed) {
// References are the durable coverage evidence. A supported answer may
// correctly note that one sub-detail is absent, so its wording cannot turn
// the entire turn into a knowledge gap.
if len(turn.References) > 0 {
return Decision{
TurnID: turn.ID,
TenantKey: turn.TenantKey,
Channel: turn.Channel,
EvaluatorVersion: EvaluatorVersion,
Verdict: VerdictCovered,
ReasonCode: ReasonCoveredHasReferences,
ReferenceCount: len(turn.References),
TopScore: maxScore(turn.References),
}, nil
}
// An explicit abstention is a stable machine contract only when it is the
// complete answer. Do not use broad natural-language substring matching.
if isCanonicalNoAnswer(trimmed) {
return Decision{
TurnID: turn.ID,
TenantKey: turn.TenantKey,
@@ -86,8 +77,8 @@ func (e *Evaluator) Evaluate(turn core.Turn) (Decision, error) {
EvaluatorVersion: EvaluatorVersion,
Verdict: VerdictUncovered,
ReasonCode: ReasonUncoveredExplicitNoAnswer,
ReferenceCount: len(turn.References),
TopScore: maxScore(turn.References),
ReferenceCount: 0,
TopScore: 0,
}, nil
}
@@ -180,23 +171,8 @@ func hasImageMedia(media []core.MediaAudit) bool {
return false
}
func isNoAnswer(answer string) bool {
lower := strings.ToLower(answer)
for _, phrase := range noAnswerPhrases {
if strings.Contains(lower, strings.ToLower(phrase)) {
return true
}
}
return containsAny(lower, retrievalGapPhrases) && containsAny(lower, humanEscalationPhrases)
}
func containsAny(value string, phrases []string) bool {
for _, phrase := range phrases {
if strings.Contains(value, strings.ToLower(phrase)) {
return true
}
}
return false
func isCanonicalNoAnswer(answer string) bool {
return strings.TrimSpace(answer) == canonicalNoAnswer
}
func maxScore(refs []core.Reference) float64 {
+45 -60
View File
@@ -8,8 +8,8 @@ import (
func TestEvaluatorVersion(t *testing.T) {
e := NewEvaluator()
if e.Version() != "v4" {
t.Errorf("Evaluator version = %q, want v4", e.Version())
if e.Version() != "v5" {
t.Errorf("Evaluator version = %q, want v5", e.Version())
}
}
@@ -30,55 +30,40 @@ func TestEvaluator_RejectsProcessingTurn(t *testing.T) {
}
}
func TestEvaluator_UncoveredExplicitNoAnswer(t *testing.T) {
e := NewEvaluator()
noAnswerPhrases := []string{
"未找到相关结果",
"没有找到你需要的信息",
"无法确定该问题的答案",
"暂无相关信息",
"知识库中没有找到相关内容",
"这个问题知识库中没有可用的信息",
}
for _, answer := range noAnswerPhrases {
turn := core.Turn{
ID: "test-turn",
Status: "succeeded",
Answer: answer,
References: []core.Reference{
{ID: "r1", Score: 0.9},
},
}
d, err := e.Evaluate(turn)
if err != nil {
t.Fatalf("unexpected error for answer %q: %v", answer, err)
}
if d.Verdict != VerdictUncovered {
t.Errorf("answer %q: verdict = %s, want uncovered", answer, d.Verdict)
}
if d.ReasonCode != ReasonUncoveredExplicitNoAnswer {
t.Errorf("answer %q: reason_code = %s, want %s", answer, d.ReasonCode, ReasonUncoveredExplicitNoAnswer)
}
}
}
func TestEvaluator_ProductionMissingProductPhraseTrumpsReferences(t *testing.T) {
func TestEvaluator_UncoveredCanonicalNoAnswerWithoutReferences(t *testing.T) {
e := NewEvaluator()
turn := core.Turn{
ID: "missing-product-turn",
Status: "succeeded",
Answer: "根据提供的检索信息,无法提供对应手册。原因:产品不存在于资料中;检索资料未提及该产品。建议联系技术支持部门。",
References: []core.Reference{
{ID: "unrelated-1", Score: 0.395940},
{ID: "unrelated-2", Score: 0.311595},
},
ID: "test-turn",
Status: "succeeded",
Answer: canonicalNoAnswer,
KnowledgeSearched: true,
}
d, err := e.Evaluate(turn)
if err != nil {
t.Fatal(err)
}
if d.Verdict != VerdictUncovered || d.ReasonCode != ReasonUncoveredExplicitNoAnswer {
t.Fatalf("verdict=%s reason=%s", d.Verdict, d.ReasonCode)
t.Fatalf("verdict/reason = %s/%s, want %s/%s", d.Verdict, d.ReasonCode, VerdictUncovered, ReasonUncoveredExplicitNoAnswer)
}
}
func TestEvaluator_NovaPartialNotFoundWithSevenReferencesIsCovered(t *testing.T) {
e := NewEvaluator()
turn := core.Turn{
ID: "nova-partial-turn",
Status: "succeeded",
Answer: "已找到Nova系列的相关资料;未找到VX500 2.5D视觉一体方案的明确说明。",
References: []core.Reference{
{ID: "r1"}, {ID: "r2"}, {ID: "r3"}, {ID: "r4"},
{ID: "r5"}, {ID: "r6"}, {ID: "r7"},
},
}
d, err := e.Evaluate(turn)
if err != nil {
t.Fatal(err)
}
if d.Verdict != VerdictCovered || d.ReasonCode != ReasonCoveredHasReferences || d.ReferenceCount != 7 {
t.Fatalf("verdict/reason/references = %s/%s/%d, want %s/%s/7", d.Verdict, d.ReasonCode, d.ReferenceCount, VerdictCovered, ReasonCoveredHasReferences)
}
}
@@ -205,9 +190,9 @@ func TestEvaluator_TopScoreZeroWhenNoReferences(t *testing.T) {
}
}
func TestEvaluator_NoAnswerTrumpsReferences(t *testing.T) {
// When answer contains explicit "no answer" phrase, it should be uncovered
// even if references exist. No-answer detection takes priority.
func TestEvaluator_ReferencesTrumpNaturalLanguageNoAnswer(t *testing.T) {
// References are durable coverage evidence; a natural-language phrase must
// not override them.
e := NewEvaluator()
turn := core.Turn{
ID: "test-turn",
@@ -222,11 +207,11 @@ func TestEvaluator_NoAnswerTrumpsReferences(t *testing.T) {
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if d.Verdict != VerdictUncovered {
t.Errorf("no-answer with references: verdict = %s, want uncovered", d.Verdict)
if d.Verdict != VerdictCovered {
t.Errorf("answer with references: verdict = %s, want covered", d.Verdict)
}
if d.ReasonCode != ReasonUncoveredExplicitNoAnswer {
t.Errorf("no-answer with references: reason_code = %s, want %s", d.ReasonCode, ReasonUncoveredExplicitNoAnswer)
if d.ReasonCode != ReasonCoveredHasReferences {
t.Errorf("answer with references: reason_code = %s, want %s", d.ReasonCode, ReasonCoveredHasReferences)
}
}
@@ -245,12 +230,12 @@ func TestEvaluator_NoAnswerPhraseMatchesInsideAnswer(t *testing.T) {
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if d.ReasonCode != ReasonUncoveredExplicitNoAnswer {
t.Fatalf("reason_code = %s, want %s", d.ReasonCode, ReasonUncoveredExplicitNoAnswer)
if d.ReasonCode != ReasonCoveredHasReferences {
t.Fatalf("reason_code = %s, want %s", d.ReasonCode, ReasonCoveredHasReferences)
}
}
func TestEvaluator_ProductionUnavailableAuthorityPhraseTrumpsReferences(t *testing.T) {
func TestEvaluator_ProductionUnavailableAuthorityPhraseWithReferencesIsCovered(t *testing.T) {
e := NewEvaluator()
turn := core.Turn{
ID: "cra-price-turn",
@@ -265,11 +250,11 @@ func TestEvaluator_ProductionUnavailableAuthorityPhraseTrumpsReferences(t *testi
if err != nil {
t.Fatalf("Evaluate: %v", err)
}
if d.Verdict != VerdictUncovered {
t.Fatalf("verdict = %s, want %s", d.Verdict, VerdictUncovered)
if d.Verdict != VerdictCovered {
t.Fatalf("verdict = %s, want %s", d.Verdict, VerdictCovered)
}
if d.ReasonCode != ReasonUncoveredExplicitNoAnswer {
t.Fatalf("reason_code = %s, want %s", d.ReasonCode, ReasonUncoveredExplicitNoAnswer)
if d.ReasonCode != ReasonCoveredHasReferences {
t.Fatalf("reason_code = %s, want %s", d.ReasonCode, ReasonCoveredHasReferences)
}
}
@@ -282,10 +267,10 @@ func TestEvaluator_RetrievalGapRequiresHumanEscalation(t *testing.T) {
reason string
}{
{
name: "production NOVA price answer",
name: "retrieval wording with evidence remains covered",
answer: "当前暂未查询到Nova系列机器人的公开销售价格信息。\n\n如需获取报价,建议您联系越疆科技官方销售渠道或对应区域的销售工程师咨询具体价格方案。",
verdict: VerdictUncovered,
reason: ReasonUncoveredExplicitNoAnswer,
verdict: VerdictCovered,
reason: ReasonCoveredHasReferences,
},
{
name: "retrieval gap without escalation is a valid negative result",
+6 -8
View File
@@ -104,14 +104,12 @@ func TestService_EvaluateAndPersist_UncoveredExplicitNoAnswer(t *testing.T) {
s, _ := NewService(repo)
turn := core.Turn{
ID: "turn-3",
TenantKey: "t1",
Channel: "dingtalk",
Status: "succeeded",
Answer: "未找到相关结果",
References: []core.Reference{
{ID: "r1", Score: 0.5},
},
ID: "turn-3",
TenantKey: "t1",
Channel: "dingtalk",
Status: "succeeded",
Answer: canonicalNoAnswer,
KnowledgeSearched: true,
}
d, err := s.EvaluateAndPersist(context.Background(), turn)
+1 -1
View File
@@ -33,7 +33,7 @@ const (
)
// EvaluatorVersion is the current version of the evaluator.
const EvaluatorVersion = "v4"
const EvaluatorVersion = "v5"
var (
ErrTurnNotSucceeded = errors.New("coverage evaluator: turn status must be succeeded")
+13
View File
@@ -49,6 +49,19 @@ func (s *AliyunOSSStore) Put(ctx context.Context, key string, data []byte, conte
return nil
}
// Exists reports whether an object is already stored under key, so identical
// content is never uploaded twice.
func (s *AliyunOSSStore) Exists(ctx context.Context, key string) (bool, error) {
if s == nil || s.client == nil || strings.TrimSpace(key) == "" {
return false, errors.New("outbound image: invalid OSS exists request")
}
exists, err := s.client.IsObjectExist(ctx, s.bucket, key)
if err != nil {
return false, errors.New("outbound image: OSS exists check failed")
}
return exists, nil
}
func (s *AliyunOSSStore) Get(ctx context.Context, key string, maxBytes int64) ([]byte, string, error) {
if s == nil || s.client == nil || strings.TrimSpace(key) == "" || maxBytes <= 0 {
return nil, "", errors.New("outbound image: invalid OSS get request")
+137
View File
@@ -0,0 +1,137 @@
package outboundimage
import (
"bytes"
"image"
"image/color"
"image/draw"
"image/jpeg"
_ "image/png"
"strings"
)
const (
// compressMaxEdge bounds the longest side of a published screenshot. Manual
// screenshots are captured at 2x-3x, so capping the longest side keeps text
// readable while cutting the upload size by an order of magnitude.
compressMaxEdge = 1600
// compressJPEGQuality is the re-encode quality for compressed screenshots.
compressJPEGQuality = 82
// compressMinBytes skips images that are already small: re-encoding them
// saves little upload time and could only lose detail.
compressMinBytes = 128 << 10
// compressMaxPixels refuses to decode huge images so one oversized
// screenshot cannot exhaust the container's memory.
compressMaxPixels = 24 << 20
)
// CompressImage re-encodes an outbound screenshot as a smaller JPEG. It returns
// the payload to publish together with its content type, and reports whether
// compression applied. Images that are undecodable, already small, animated, or
// not smaller after re-encoding are returned untouched, so publishing always
// has a usable payload.
func CompressImage(data []byte, contentType string) ([]byte, string, bool) {
if !compressibleImageType(contentType) || len(data) < compressMinBytes {
return data, contentType, false
}
decoded, _, err := image.Decode(bytes.NewReader(data))
if err != nil {
return data, contentType, false
}
bounds := decoded.Bounds()
if bounds.Dx() <= 0 || bounds.Dy() <= 0 || bounds.Dx()*bounds.Dy() > compressMaxPixels {
return data, contentType, false
}
encoded := encodeCompressedImage(downscaleRGBA(flattenToRGBA(decoded), compressMaxEdge))
if encoded == nil || len(encoded) >= len(data) {
return data, contentType, false
}
return encoded, "image/jpeg", true
}
// compressibleImageType reports whether the content type is a still raster
// format this package can decode and re-encode. GIF and WebP are left alone to
// preserve animation and format fidelity.
func compressibleImageType(contentType string) bool {
switch strings.ToLower(strings.TrimSpace(strings.Split(contentType, ";")[0])) {
case "image/png", "image/jpeg", "image/jpg":
return true
default:
return false
}
}
func encodeCompressedImage(source image.Image) []byte {
var encoded bytes.Buffer
if err := jpeg.Encode(&encoded, source, &jpeg.Options{Quality: compressJPEGQuality}); err != nil {
return nil
}
return encoded.Bytes()
}
// flattenToRGBA copies an image onto an opaque white canvas. JPEG has no alpha
// channel, so transparent regions must be composited before encoding or they
// would be rendered black.
func flattenToRGBA(source image.Image) *image.RGBA {
bounds := source.Bounds()
target := image.NewRGBA(image.Rect(0, 0, bounds.Dx(), bounds.Dy()))
draw.Draw(target, target.Bounds(), image.NewUniform(color.White), image.Point{}, draw.Src)
draw.Draw(target, target.Bounds(), source, bounds.Min, draw.Over)
return target
}
// downscaleRGBA shrinks an image so neither side exceeds maxEdge, averaging the
// source pixels that fall into each destination pixel (box filter) so
// screenshot text stays legible.
func downscaleRGBA(source *image.RGBA, maxEdge int) *image.RGBA {
width, height := source.Rect.Dx(), source.Rect.Dy()
if width <= maxEdge && height <= maxEdge {
return source
}
targetWidth, targetHeight := width, height
if width >= height {
targetWidth = maxEdge
targetHeight = height * maxEdge / width
} else {
targetHeight = maxEdge
targetWidth = width * maxEdge / height
}
if targetWidth < 1 {
targetWidth = 1
}
if targetHeight < 1 {
targetHeight = 1
}
target := image.NewRGBA(image.Rect(0, 0, targetWidth, targetHeight))
for y := 0; y < targetHeight; y++ {
startY := y * height / targetHeight
endY := (y + 1) * height / targetHeight
if endY <= startY {
endY = startY + 1
}
for x := 0; x < targetWidth; x++ {
startX := x * width / targetWidth
endX := (x + 1) * width / targetWidth
if endX <= startX {
endX = startX + 1
}
var red, green, blue, count uint32
for row := startY; row < endY; row++ {
line := source.Pix[row*source.Stride:]
for column := startX; column < endX; column++ {
offset := column * 4
red += uint32(line[offset])
green += uint32(line[offset+1])
blue += uint32(line[offset+2])
count++
}
}
offset := y*target.Stride + x*4
target.Pix[offset] = uint8(red / count)
target.Pix[offset+1] = uint8(green / count)
target.Pix[offset+2] = uint8(blue / count)
target.Pix[offset+3] = 0xff
}
}
return target
}
+188
View File
@@ -0,0 +1,188 @@
package outboundimage
import (
"bytes"
"context"
"image"
"image/color"
"image/jpeg"
"image/png"
"strings"
"testing"
)
func pngFixture(t *testing.T, width, height int, alpha bool) []byte {
t.Helper()
source := image.NewRGBA(image.Rect(0, 0, width, height))
seed := uint32(12345)
for y := 0; y < height; y++ {
for x := 0; x < width; x++ {
// A pseudo-random pattern keeps the PNG as large as a real manual
// screenshot, so the size thresholds are exercised for real.
seed = seed*1664525 + 1013904223
alphaValue := uint8(0xff)
if alpha && x < width/2 {
alphaValue = 0x00
}
source.SetRGBA(x, y, color.RGBA{
R: uint8(seed >> 24),
G: uint8(seed >> 16),
B: uint8(seed >> 8),
A: alphaValue,
})
}
}
var encoded bytes.Buffer
if err := png.Encode(&encoded, source); err != nil {
t.Fatal(err)
}
return encoded.Bytes()
}
func TestCompressImageDownscalesLargeScreenshotToSmallerJPEG(t *testing.T) {
original := pngFixture(t, 3200, 1800, false)
data, contentType, applied := CompressImage(original, "image/png")
if !applied {
t.Fatal("expected a large screenshot to be compressed")
}
if contentType != "image/jpeg" {
t.Fatalf("contentType=%q", contentType)
}
if len(data) >= len(original) {
t.Fatalf("compressed size=%d original=%d", len(data), len(original))
}
decoded, err := jpeg.Decode(bytes.NewReader(data))
if err != nil {
t.Fatalf("compressed payload is not a JPEG: %v", err)
}
if decoded.Bounds().Dx() != compressMaxEdge || decoded.Bounds().Dy() != compressMaxEdge*1800/3200 {
t.Fatalf("bounds=%v", decoded.Bounds())
}
}
func TestCompressImageFlattensTransparencyInsteadOfBlack(t *testing.T) {
original := pngFixture(t, 2000, 2000, true)
data, _, applied := CompressImage(original, "image/png")
if !applied {
t.Fatal("expected a transparent screenshot to be compressed")
}
decoded, err := jpeg.Decode(bytes.NewReader(data))
if err != nil {
t.Fatal(err)
}
red, green, blue, _ := decoded.At(10, 10).RGBA()
if red < 0xf000 || green < 0xf000 || blue < 0xf000 {
t.Fatalf("transparent area rendered as %v instead of white", []uint32{red, green, blue})
}
}
func TestCompressImageLeavesSmallOrUndecodablePayloadsUntouched(t *testing.T) {
small := pngFixture(t, 64, 64, false)
if len(small) >= compressMinBytes {
t.Fatalf("fixture is not small: %d", len(small))
}
for _, test := range []struct {
name string
data []byte
contentType string
}{
{name: "small", data: small, contentType: "image/png"},
{name: "not an image", data: bytes.Repeat([]byte("x"), compressMinBytes+1), contentType: "image/png"},
{name: "gif stays animated", data: bytes.Repeat([]byte("x"), compressMinBytes+1), contentType: "image/gif"},
{name: "text", data: bytes.Repeat([]byte("x"), compressMinBytes+1), contentType: "text/plain"},
} {
t.Run(test.name, func(t *testing.T) {
data, contentType, applied := CompressImage(test.data, test.contentType)
if applied || contentType != test.contentType || !bytes.Equal(data, test.data) {
t.Fatalf("payload changed: applied=%v contentType=%q bytes=%d", applied, contentType, len(data))
}
})
}
}
type capturePublisher struct {
data []byte
contentType string
calls int
}
func (p *capturePublisher) Publish(_ context.Context, data []byte, contentType string) (string, error) {
p.calls++
p.data = append([]byte(nil), data...)
p.contentType = contentType
return "https://img.example/1.png", nil
}
type pngFetcher struct {
data []byte
}
func (f *pngFetcher) Fetch(_ context.Context, _ string) ([]byte, string, error) {
return f.data, "image/png", nil
}
func TestRewritePublishesCompressedScreenshots(t *testing.T) {
original := pngFixture(t, 3200, 1800, false)
reference := "resource://AbCdEfGhIjKlMnOpQrStUv"
input := "步骤一\n![图1](" + reference + ")\n"
publisher := &capturePublisher{}
NewRewriter(&pngFetcher{data: original}, publisher, 5).Rewrite(context.Background(), input)
if publisher.calls != 1 || publisher.contentType != "image/jpeg" {
t.Fatalf("calls=%d contentType=%q", publisher.calls, publisher.contentType)
}
if len(publisher.data) >= len(original) {
t.Fatalf("published bytes=%d original=%d", len(publisher.data), len(original))
}
disabled := &capturePublisher{}
NewRewriter(&pngFetcher{data: original}, disabled, 5, WithImageCompression(false)).Rewrite(context.Background(), input)
if disabled.contentType != "image/png" || !bytes.Equal(disabled.data, original) {
t.Fatalf("compression opt-out changed the payload: contentType=%q bytes=%d", disabled.contentType, len(disabled.data))
}
}
func TestStableHashPublisherReusesExistingObject(t *testing.T) {
store := &existingObjectStore{}
publisher, err := NewStableHashPublisher(store, "dobotai/im-images", "https://img.example/i")
if err != nil {
t.Fatal(err)
}
first, err := publisher.Publish(context.Background(), []byte("same-image"), "image/jpeg")
if err != nil {
t.Fatal(err)
}
if store.puts != 1 {
t.Fatalf("puts=%d", store.puts)
}
store.exists = true
second, err := publisher.Publish(context.Background(), []byte("same-image"), "image/jpeg")
if err != nil {
t.Fatal(err)
}
if first != second {
t.Fatalf("url1=%q url2=%q", first, second)
}
if store.puts != 1 {
t.Fatalf("existing object was uploaded again: puts=%d", store.puts)
}
if !strings.HasPrefix(second, "https://img.example/i/") {
t.Fatalf("url=%q", second)
}
}
type existingObjectStore struct {
exists bool
puts int
}
func (s *existingObjectStore) Put(_ context.Context, _ string, _ []byte, _ string) error {
s.puts++
return nil
}
func (s *existingObjectStore) Exists(_ context.Context, _ string) (bool, error) {
return s.exists, nil
}
+14
View File
@@ -6,6 +6,7 @@ import (
"encoding/hex"
"errors"
"fmt"
"log"
"net/url"
"path"
"regexp"
@@ -18,6 +19,13 @@ type ObjectWriter interface {
Put(ctx context.Context, key string, data []byte, contentType string) error
}
// ObjectStater reports whether an object already exists at key. Stores that
// implement it let the publisher skip re-uploading identical content, which is
// what makes a repeated question instant: object keys are content hashes.
type ObjectStater interface {
Exists(ctx context.Context, key string) (bool, error)
}
type StableHashPublisher struct {
store ObjectWriter
prefix string
@@ -52,6 +60,12 @@ func (p *StableHashPublisher) Publish(ctx context.Context, data []byte, contentT
idHex := hex.EncodeToString(idBytes)
id := fmt.Sprintf("%s-%s-%s-%s-%s", idHex[:8], idHex[8:12], idHex[12:16], idHex[16:20], idHex[20:])
key := path.Join(p.prefix, id)
if stater, ok := p.store.(ObjectStater); ok {
if exists, err := stater.Exists(ctx, key); err == nil && exists {
log.Printf("outbound image: existing object reused key=%s bytes=%d contentType=%s", key, len(data), contentType)
return p.publicBaseURL + "/" + id, nil
}
}
if err := p.store.Put(ctx, key, data, contentType); err != nil {
return "", errors.New("outbound image: publish object failed")
}
+140 -23
View File
@@ -2,11 +2,15 @@ package outboundimage
import (
"context"
"log"
"net"
"net/url"
"path"
"regexp"
"sort"
"strings"
"sync"
"time"
)
var (
@@ -38,41 +42,56 @@ type Rewriter struct {
fetcher Fetcher
publisher Publisher
maxImages int
compress bool
}
func NewRewriter(fetcher Fetcher, publisher Publisher, maxImages int) *Rewriter {
return &Rewriter{fetcher: fetcher, publisher: publisher, maxImages: maxImages}
// RewriterOption customises how one answer's images are prepared for delivery.
type RewriterOption func(*Rewriter)
// WithImageCompression switches the pre-publish JPEG re-encode. It is enabled by
// default because multi-megabyte screenshots dominate the delivery budget.
func WithImageCompression(enabled bool) RewriterOption {
return func(r *Rewriter) { r.compress = enabled }
}
func NewRewriter(fetcher Fetcher, publisher Publisher, maxImages int, options ...RewriterOption) *Rewriter {
rewriter := &Rewriter{fetcher: fetcher, publisher: publisher, maxImages: maxImages, compress: true}
for _, option := range options {
if option != nil {
option(rewriter)
}
}
return rewriter
}
// rewriteConcurrency bounds how many images are fetched and published in
// parallel. Manual screenshots are multi-megabyte objects whose upload time
// dominates the delivery budget, so a serial pass cannot publish a long answer
// before the delivery deadline and silently loses the remaining images.
const rewriteConcurrency = 4
func (r *Rewriter) Rewrite(ctx context.Context, markdown string) string {
if r == nil || r.fetcher == nil || r.publisher == nil || r.maxImages <= 0 {
markdown = normalizePublicHTMLImages(stripResourceImages(markdown))
markdown = resourceReferenceRE.ReplaceAllString(resourceHTMLUnsafeImageRE.ReplaceAllString(markdown, ""), "")
return sanitizePublicMarkdownImages(markdown)
}
urls := map[string]string{}
attempted := 0
started := time.Now()
urls := r.resolveReferences(ctx, markdown)
resolve := func(reference string) string {
if publishedURL, ok := urls[reference]; ok {
return publishedURL
return urls[reference]
}
if len(urls) > 0 {
published, dropped := 0, 0
for _, publishedURL := range urls {
if publishedURL == "" {
dropped++
continue
}
published++
}
if attempted >= r.maxImages {
urls[reference] = ""
return ""
}
attempted++
data, contentType, err := r.fetcher.Fetch(ctx, reference)
if err != nil {
urls[reference] = ""
return ""
}
publishedURL, err := r.publisher.Publish(ctx, data, contentType)
if err != nil || !strings.HasPrefix(publishedURL, "https://") {
urls[reference] = ""
return ""
}
urls[reference] = publishedURL
return publishedURL
log.Printf("outbound image: rewrite finished references=%d published=%d dropped=%d elapsed=%s",
len(urls), published, dropped, time.Since(started).Round(time.Millisecond))
}
markdown = resourceMarkdownImageRE.ReplaceAllStringFunc(markdown, func(image string) string {
@@ -94,6 +113,104 @@ func (r *Rewriter) Rewrite(ctx context.Context, markdown string) string {
return sanitizePublicMarkdownImages(markdown)
}
// resolveReferences returns reference -> public URL for every image reference
// in content, in document order and deduplicated, capped at maxImages. Each
// image is fetched and published with at most rewriteConcurrency workers; a
// reference that cannot be published maps to "".
func (r *Rewriter) resolveReferences(ctx context.Context, content string) map[string]string {
references := r.collectReferences(content)
if len(references) == 0 {
return map[string]string{}
}
results := make(map[string]string, len(references))
var mutex sync.Mutex
var budgetOnce sync.Once
slots := make(chan struct{}, rewriteConcurrency)
var wait sync.WaitGroup
for _, reference := range references {
wait.Add(1)
go func(reference string) {
defer wait.Done()
slots <- struct{}{}
defer func() { <-slots }()
publishedURL := ""
if err := ctx.Err(); err != nil {
budgetOnce.Do(func() {
log.Printf("outbound image: delivery budget exhausted, dropping remaining images: %v", err)
})
} else {
publishedURL = r.publishReference(ctx, reference)
}
mutex.Lock()
results[reference] = publishedURL
mutex.Unlock()
}(reference)
}
wait.Wait()
return results
}
// collectReferences gathers the resource references that carry a real image,
// in document order, dropping duplicates and anything beyond maxImages. Bare or
// unquoted references are never fetched.
func (r *Rewriter) collectReferences(content string) []string {
type hit struct {
at int
reference string
}
hits := make([]hit, 0, 16)
for _, pattern := range []*regexp.Regexp{resourceMarkdownImageRE, resourceHTMLDoubleImageRE, resourceHTMLSingleImageRE} {
for _, match := range pattern.FindAllStringSubmatchIndex(content, -1) {
if len(match) < 4 || match[2] < 0 || match[3] > len(content) {
continue
}
hits = append(hits, hit{at: match[0], reference: content[match[2]:match[3]]})
}
}
sort.SliceStable(hits, func(i, j int) bool { return hits[i].at < hits[j].at })
references := make([]string, 0, len(hits))
seen := make(map[string]bool, len(hits))
for _, candidate := range hits {
if seen[candidate.reference] {
continue
}
seen[candidate.reference] = true
if len(references) >= r.maxImages {
break
}
references = append(references, candidate.reference)
}
return references
}
// publishReference fetches one image from WeKnora and publishes it to our own
// object storage. Failures are logged with their reason instead of being
// silently dropped from the answer.
func (r *Rewriter) publishReference(ctx context.Context, reference string) string {
started := time.Now()
data, contentType, err := r.fetcher.Fetch(ctx, reference)
if err != nil {
log.Printf("outbound image: fetch failed, image dropped reference=%s error=%v", reference, err)
return ""
}
if r.compress {
if compressed, compressedType, applied := CompressImage(data, contentType); applied {
log.Printf("outbound image: compressed reference=%s bytes=%d contentType=%s compressed_bytes=%d compressed_content_type=%s",
reference, len(data), contentType, len(compressed), compressedType)
data, contentType = compressed, compressedType
}
}
publishedURL, err := r.publisher.Publish(ctx, data, contentType)
if err != nil || !strings.HasPrefix(publishedURL, "https://") {
log.Printf("outbound image: publish failed, image dropped reference=%s bytes=%d contentType=%s error=%v",
reference, len(data), contentType, err)
return ""
}
log.Printf("outbound image: published reference=%s bytes=%d contentType=%s elapsed=%s",
reference, len(data), contentType, time.Since(started).Round(time.Millisecond))
return publishedURL
}
func sanitizePublicMarkdownImages(markdown string) string {
markdown = markdownImageRE.ReplaceAllStringFunc(markdown, func(image string) string {
match := markdownImageRE.FindStringSubmatch(image)
+47
View File
@@ -4,15 +4,19 @@ import (
"context"
"errors"
"strings"
"sync"
"testing"
)
type fakeFetcher struct {
mutex sync.Mutex
calls map[string]int
fail map[string]bool
}
func (f *fakeFetcher) Fetch(_ context.Context, ref string) ([]byte, string, error) {
f.mutex.Lock()
defer f.mutex.Unlock()
if f.calls == nil {
f.calls = map[string]int{}
}
@@ -24,10 +28,13 @@ func (f *fakeFetcher) Fetch(_ context.Context, ref string) ([]byte, string, erro
}
type fakePublisher struct {
mutex sync.Mutex
calls int
}
func (p *fakePublisher) Publish(_ context.Context, data []byte, contentType string) (string, error) {
p.mutex.Lock()
defer p.mutex.Unlock()
p.calls++
if contentType != "image/png" || len(data) == 0 {
return "", errors.New("bad image")
@@ -375,3 +382,43 @@ func TestDocumentImageDowngradePreservesImageAndLinkContracts(t *testing.T) {
t.Fatalf("unsafe document image became a link: %s", got)
}
}
func TestRewriteDropsImagesWhenDeliveryBudgetIsAlreadyExhausted(t *testing.T) {
fetcher := &fakeFetcher{}
publisher := &fakePublisher{}
r := NewRewriter(fetcher, publisher, 10)
ctx, cancel := context.WithCancel(context.Background())
cancel()
input := "正文\n![一](resource://FirstRefAbCdEfGhIjKlMnO)\n![二](resource://SecondRefAbCdEfGhIjKlMn)"
got := r.Rewrite(ctx, input)
if strings.Contains(got, "resource://") {
t.Fatalf("exhausted budget leaked a resource reference: %s", got)
}
if !strings.Contains(got, "正文") {
t.Fatalf("text was lost: %s", got)
}
if len(fetcher.calls) != 0 || publisher.calls != 0 {
t.Fatalf("no image may be fetched or published without budget: fetch=%v publish=%d", fetcher.calls, publisher.calls)
}
}
func TestRewritePublishesEveryImageWithinTheLimit(t *testing.T) {
fetcher := &fakeFetcher{}
publisher := &fakePublisher{}
r := NewRewriter(fetcher, publisher, 12)
input := "正文"
for i := 0; i < 12; i++ {
input += "\n![图](resource://Ref" + string(rune('A'+i)) + "bCdEfGhIjKlMnO)"
}
got := r.Rewrite(context.Background(), input)
if strings.Contains(got, "resource://") {
t.Fatalf("resource reference remained: %s", got)
}
if publisher.calls != 12 || len(fetcher.calls) != 12 {
t.Fatalf("calls fetch=%d publish=%d, want 12", len(fetcher.calls), publisher.calls)
}
}
+55
View File
@@ -0,0 +1,55 @@
package outboundtext
import (
"regexp"
)
// DingTalk cards render a small Markdown subset and no math typesetting, so a
// LaTeX formula reaches the user as raw source ($r = \sqrt{X^2 + Y^2}$).
// NormalizeMath rewrites such formulas into plain Unicode text that the card
// can show. It only touches content that actually carries math delimiters or
// well-known math commands, so ordinary prose is left byte-identical.
var (
mathBlockDelimRE = regexp.MustCompile(`\$\$([\s\S]+?)\$\$`)
mathSpanRE = regexp.MustCompile(`(?s)\$\$.+?\$\$|\\\[.+?\\\]|\\\(.+?\\\)|\$[^$\n]{1,400}\$`)
parenDelimRE = regexp.MustCompile(`\\\(([\s\S]+?)\\\)`)
bracketDelimRE = regexp.MustCompile(`\\\[([\s\S]+?)\\\]`)
sqrtRE = regexp.MustCompile(`\\sqrt\s*(?:\[([^\]]{1,20})\])?\s*\{([^{}]{1,200})\}`)
fracRE = regexp.MustCompile(`\\frac\s*\{([^{}]{1,200})\}\s*\{([^{}]{1,200})\}`)
supBracedRE = regexp.MustCompile(`\^\s*\{([^{}]{1,40})\}`)
subBracedRE = regexp.MustCompile(`_\s*\{([^{}]{1,40})\}`)
supCharRE = regexp.MustCompile(`\^([0-9A-Za-z+\-=()])`)
subCharRE = regexp.MustCompile(`_([0-9A-Za-z+\-=()])`)
commandRE = regexp.MustCompile(`\\([A-Za-z]+)`)
braceDelimRE = regexp.MustCompile(`\{([^{}]{0,200})\}`)
blankRunRE = regexp.MustCompile(`[ \t]{2,}`)
)
var mathCommands = map[string]string{
"times": "×", "cdot": "·", "div": "÷", "pm": "±", "mp": "∓",
"leq": "≤", "le": "≤", "geq": "≥", "ge": "≥", "neq": "≠", "ne": "≠",
"approx": "≈", "sim": "∼", "simeq": "≃", "equiv": "≡", "propto": "∝",
"infty": "∞", "partial": "∂", "nabla": "∇",
"alpha": "α", "beta": "β", "gamma": "γ", "delta": "δ", "Delta": "Δ",
"epsilon": "ε", "theta": "θ", "lambda": "λ", "mu": "μ", "rho": "ρ",
"sigma": "σ", "tau": "τ", "phi": "φ", "omega": "ω", "Omega": "Ω",
"eta": "η", "kappa": "κ", "nu": "ν", "xi": "ξ", "zeta": "ζ", "psi": "ψ",
"sum": "Σ", "prod": "Π", "int": "∫", "iint": "∬", "oint": "∮",
"in": "∈", "notin": "∉", "subset": "⊂", "subseteq": "⊆", "cup": "∪", "cap": "∩",
"rightarrow": "→", "to": "→", "longrightarrow": "→", "leftarrow": "←",
"Rightarrow": "⇒", "leftrightarrow": "↔", "mapsto": "↦",
"forall": "∀", "exists": "∃", "angle": "∠", "degree": "°", "circ": "°",
"perp": "⊥", "parallel": "∥", "therefore": "∴", "because": "∵",
"ldots": "…", "cdots": "…", "dots": "…", "prime": "′",
"quad": " ", "qquad": " ", "limits": "", "displaystyle": "",
"left": "", "right": "", "bmod": "mod ", "pmod": "mod ",
"text": "", "mathrm": "", "mathit": "", "operatorname": "",
"overline": "", "underline": "",
}
var escapedPunctuation = map[string]string{
`\\`: "\n", `\%`: "%", `\&`: "&", `\#`: "#", `\_`: "_", `\$`: "$",
`\{`: "{", `\}`: "}", `\,`: " ", `\;`: " ", `\!`: "",
`\(`: "", `\)`: "", `\[`: "", `\]`: "",
}
+128
View File
@@ -0,0 +1,128 @@
package outboundtext
import "strings"
func NormalizeMath(markdown string) string {
if !carriesMath(markdown) {
return markdown
}
content := mathSpanRE.ReplaceAllStringFunc(markdown, convertMathSpan)
// Outside math spans only LaTeX commands are converted. Identifiers and URLs
// (resource://…_pltQ) must never be rewritten: a mangled reference can no
// longer be published, and the outbound sanitizer then drops the whole image.
content = commandRE.ReplaceAllStringFunc(content, replaceCommand)
content = blankRunRE.ReplaceAllString(content, " ")
return content
}
func convertMathSpan(span string) string {
inner := span
switch {
case strings.HasPrefix(inner, "$$"):
inner = strings.TrimSuffix(strings.TrimPrefix(inner, "$$"), "$$")
return "\n" + convertMath(inner) + "\n"
case strings.HasPrefix(inner, `\[`):
inner = strings.TrimSuffix(strings.TrimPrefix(inner, `\[`), `\]`)
case strings.HasPrefix(inner, `\(`):
inner = strings.TrimSuffix(strings.TrimPrefix(inner, `\(`), `\)`)
default:
inner = strings.TrimSuffix(strings.TrimPrefix(inner, "$"), "$")
}
return convertMath(inner)
}
func convertMath(inner string) string {
content := inner
for from, to := range escapedPunctuation {
content = strings.ReplaceAll(content, from, to)
}
for i := 0; i < 3; i++ {
next := sqrtRE.ReplaceAllStringFunc(content, expandRoot)
next = fracRE.ReplaceAllString(next, "($1)/($2)")
next = supBracedRE.ReplaceAllStringFunc(next, convertSuperscript)
next = subBracedRE.ReplaceAllStringFunc(next, convertSubscript)
next = supCharRE.ReplaceAllStringFunc(next, convertSuperscript)
next = subCharRE.ReplaceAllStringFunc(next, convertSubscript)
if next == content {
break
}
content = next
}
content = commandRE.ReplaceAllStringFunc(content, replaceCommand)
content = braceDelimRE.ReplaceAllString(content, "$1")
return content
}
func carriesMath(markdown string) bool {
if !strings.Contains(markdown, "$") && !strings.Contains(markdown, `\(`) &&
!strings.Contains(markdown, `\[`) && !strings.Contains(markdown, `\sqrt`) &&
!strings.Contains(markdown, `\frac`) && !strings.Contains(markdown, `\times`) &&
!strings.Contains(markdown, `\times`) {
return false
}
return true
}
func expandRoot(match string) string {
parts := sqrtRE.FindStringSubmatch(match)
degree, radicand := strings.TrimSpace(parts[1]), strings.TrimSpace(parts[2])
if degree != "" {
return "√[" + degree + "](" + radicand + ")"
}
return "√(" + radicand + ")"
}
func convertSuperscript(match string) string {
body := strings.TrimPrefix(match, "^")
body = strings.TrimSuffix(strings.TrimPrefix(body, "{"), "}")
if mapped, ok := mapRun(body, superscripts); ok {
return mapped
}
return "^(" + body + ")"
}
func convertSubscript(match string) string {
body := strings.TrimPrefix(match, "_")
body = strings.TrimSuffix(strings.TrimPrefix(body, "{"), "}")
if mapped, ok := mapRun(body, subscripts); ok {
return mapped
}
return "_(" + body + ")"
}
func mapRun(body string, table map[rune]string) (string, bool) {
if body == "" {
return "", false
}
var b strings.Builder
for _, r := range body {
mapped, ok := table[r]
if !ok {
return "", false
}
b.WriteString(mapped)
}
return b.String(), true
}
func replaceCommand(match string) string {
name := strings.TrimPrefix(match, `\`)
if mapped, ok := mathCommands[name]; ok {
return mapped
}
return name
}
var superscripts = map[rune]string{
'0': "⁰", '1': "¹", '2': "²", '3': "³", '4': "⁴", '5': "⁵", '6': "⁶",
'7': "⁷", '8': "⁸", '9': "⁹", '+': "⁺", '-': "⁻", '=': "⁼",
'(': "⁽", ')': "⁾", 'n': "ⁿ", 'i': "ⁱ", 'T': "ᵀ",
}
var subscripts = map[rune]string{
'0': "₀", '1': "₁", '2': "₂", '3': "₃", '4': "₄", '5': "₅", '6': "₆",
'7': "₇", '8': "₈", '9': "₉", '+': "₊", '-': "₋", '=': "₌",
'(': "₍", ')': "₎", 'a': "ₐ", 'e': "ₑ", 'i': "ᵢ", 'j': "ⱼ", 'k': "ₖ",
'n': "ₙ", 'o': "ₒ", 'p': "ₚ", 'r': "ᵣ", 's': "ₛ", 't': "ₜ", 'u': "ᵤ",
'v': "ᵥ", 'x': "ₓ",
}
+36
View File
@@ -0,0 +1,36 @@
package outboundtext
import "testing"
func TestNormalizeMath(t *testing.T) {
cases := []struct{ name, in, want string }{
{"inline sqrt", `$r = \sqrt{X^2 + Y^2}$`, `r = √(X² + Y²)`},
{"inline inside prose", `计算公式为 $r = \sqrt{X^2 + Y^2}$(单位:cm)`, `计算公式为 r = √(X² + Y²)(单位:cm)`},
{"block frac times", `$$\frac{a}{b} \times c$$`, "\n(a)/(b) × c\n"},
{"paren delimiters", `\(x \leq y\)`, `x ≤ y`},
{"bracket delimiters", `\[ \sum_{i=1}^{n} x_i \]`, ` Σᵢ₌₁ⁿ xᵢ `},
{"nth root", `$\sqrt[3]{V}$`, `√[3](V)`},
{"no math untouched", `C:\new\folder 与 100% 覆盖`, `C:\new\folder 与 100% 覆盖`},
{"dollar free prose", `价格是 5 元, 无公式`, `价格是 5 元, 无公式`},
{"resource url untouched", `![图](resource://efVTXx0gY9Lk2vuZM_pltQ) 公式 $r = \sqrt{X^2 + Y^2}$`, `![图](resource://efVTXx0gY9Lk2vuZM_pltQ) 公式 r = √(X² + Y²)`},
{"code span untouched", "`x_1` 与 $x_1$", "`x_1` 与 x₁"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got := NormalizeMath(tc.in)
if got != tc.want {
t.Fatalf("NormalizeMath(%q) = %q, want %q", tc.in, got, tc.want)
}
if again := NormalizeMath(got); again != got {
t.Fatalf("NormalizeMath is not idempotent: %q -> %q", got, again)
}
})
}
}
func TestNormalizeMathKeepsPlainMarkdown(t *testing.T) {
plain := "### 标题\n\n- 项目一\n- 项目二\n\n普通段落,没有公式。"
if got := NormalizeMath(plain); got != plain {
t.Fatalf("plain markdown changed: %q", got)
}
}
-20
View File
@@ -1,20 +0,0 @@
package qa
import "testing"
func TestLocalizedQueryPreservesOriginalWhenAuto(t *testing.T) {
if got := localizedQuery("原始问题", "auto"); got != "原始问题" {
t.Fatalf("got %q", got)
}
}
func TestLocalizedQueryAddsRequestedLanguageWithoutChangingInput(t *testing.T) {
original := "机械臂怎么复位?"
got := localizedQuery(original, "ja")
if got != "日本語で回答してください。\n\n"+original {
t.Fatalf("got %q", got)
}
if original != "机械臂怎么复位?" {
t.Fatal("original query changed")
}
}
+153
View File
@@ -0,0 +1,153 @@
package qa
import (
"context"
"testing"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/reasoningmode"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/session"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/weknora"
)
func reasoningAnswerFixture() (string, string, string) {
narration := "我需要先确认这台设备的型号和手册。"
body := "## CR20V 关节电机最大旋转扭矩\n\n| 关节 | 扭矩 |\n| --- | --- |\n| J1 | 98 N·m |"
return narration, body, narration + "\n\n" + body
}
func reasoningEventClient(answer string) *fakeEventStreamingWeKnora {
return &fakeEventStreamingWeKnora{
fakeWeKnora: &fakeWeKnora{},
events: []weknora.StreamEvent{
{ResponseType: "answer", FullAnswer: "我需要先确认这台设备的型号和手册。"},
{ResponseType: "tool_call", ToolName: "knowledge_search"},
{ResponseType: "tool_result", ToolName: "knowledge_search"},
{ResponseType: "answer", FullAnswer: answer, Done: true},
},
streamResult: weknora.ChatResult{RequestID: "request-reasoning", Answer: answer},
}
}
func TestAskEventsReasoningKeepsOnlyAnswerBody(t *testing.T) {
_, body, answer := reasoningAnswerFixture()
repo := newMemoryRepository()
client := reasoningEventClient(answer)
service, err := New(repo, fixedResolver{}, client, session.GroupShared)
if err != nil {
t.Fatal(err)
}
req := streamQARequest("message-reasoning")
req.ReasoningExecution = string(reasoningmode.ExecutionReasoning)
result, err := service.AskEvents(context.Background(), req, func(weknora.StreamEvent) {})
if err != nil {
t.Fatalf("AskEvents: %v", err)
}
if result.Answer != body {
t.Fatalf("delivered answer must keep the body only, got %q (streamCalls=%d)", result.Answer, client.streamCalls)
}
repo.mu.Lock()
stored := repo.turns["tenant:dingtalk:message-reasoning"]
repo.mu.Unlock()
if stored.Answer != body {
t.Fatalf("stored answer must keep the body only, got %q", stored.Answer)
}
}
func TestAskEventsQuickKeepsFullAnswer(t *testing.T) {
_, _, answer := reasoningAnswerFixture()
repo := newMemoryRepository()
service, err := New(repo, fixedResolver{}, reasoningEventClient(answer), session.GroupShared)
if err != nil {
t.Fatal(err)
}
result, err := service.AskEvents(context.Background(), streamQARequest("message-quick"), func(weknora.StreamEvent) {})
if err != nil {
t.Fatalf("AskEvents: %v", err)
}
if result.Answer != answer {
t.Fatalf("quick answer must be untouched, got %q", result.Answer)
}
}
// Production shape (turn efae317e, 给我CRA5的负载图): the agent narrates its
// steps, then publishes the answer — lead-in, image and all — as its own answer
// run. Cutting at the first structural line threw the lead-in and the image
// away; the run boundary keeps them.
func cra5Fixture() (string, string, string) {
narration := "用户想要 CRA5(CR5A)的末端负载曲线图。让我先在知识库中搜索相关的产品手册和 FAQ。"
body := "我已经找到了 CRA5(CR5A)的末端负载曲线图,并确认了负载曲线的读法说明。下面给出完整回答。\n\n" +
"CRA5(即 CR5A)的末端负载曲线如下:\n\n" +
"![CR5A 末端负载曲线](resource://FsVe56UYMz64sm8NsUZNrg)\n\n" +
"---\n\n" +
"## 一、负载曲线怎么读\n\n" +
"负载曲线反映的是:末端负载能力与负载质心偏心距离成负相关。"
return narration, body, narration + body
}
func TestAskEventsReasoningKeepsLeadInAndImageFromAnswerRun(t *testing.T) {
narration, body, answer := cra5Fixture()
// The heuristic on its own drops the lead-in and the image; the observation
// must win over it.
if heuristic := reasoningmode.TrimReasoningNarration(answer); heuristic == body {
t.Fatal("fixture no longer exposes the heuristic defect")
}
repo := newMemoryRepository()
client := &fakeEventStreamingWeKnora{
fakeWeKnora: &fakeWeKnora{},
events: []weknora.StreamEvent{
{ResponseType: "answer", FullAnswer: narration, AnswerRunID: "step-1-answer", AnswerRunStart: 0},
{ResponseType: "tool_call", ToolName: "knowledge_search"},
{ResponseType: "tool_result", ToolName: "knowledge_search"},
{ResponseType: "thinking"},
{ResponseType: "answer", FullAnswer: answer, Done: true, AnswerRunID: "final-answer", AnswerRunStart: len(narration)},
},
streamResult: weknora.ChatResult{RequestID: "request-cra5", Answer: answer},
}
service, err := New(repo, fixedResolver{}, client, session.GroupShared)
if err != nil {
t.Fatal(err)
}
req := streamQARequest("message-cra5")
req.ReasoningExecution = string(reasoningmode.ExecutionReasoning)
result, err := service.AskEvents(context.Background(), req, func(weknora.StreamEvent) {})
if err != nil {
t.Fatalf("AskEvents: %v", err)
}
if result.Answer != body {
t.Fatalf("delivered answer must keep the answer body (lead-in + image), got %q", result.Answer)
}
}
func TestAskEventsReasoningFallsBackWhenFinalRunInterrupted(t *testing.T) {
_, _, answer := reasoningAnswerFixture()
narration := "我需要先确认这台设备的型号和手册。"
repo := newMemoryRepository()
client := &fakeEventStreamingWeKnora{
fakeWeKnora: &fakeWeKnora{},
events: []weknora.StreamEvent{
{ResponseType: "answer", FullAnswer: narration, AnswerRunID: "step-1-answer", AnswerRunStart: 0},
{ResponseType: "tool_call", ToolName: "knowledge_search"},
{ResponseType: "tool_result", ToolName: "knowledge_search"},
// The last answer run keeps running across another tool call, so the
// stream cannot prove where the body starts: fall back to the heuristic.
{ResponseType: "answer", FullAnswer: answer, AnswerRunID: "step-2-answer", AnswerRunStart: len(narration)},
{ResponseType: "tool_call", ToolName: "knowledge_search"},
{ResponseType: "tool_result", ToolName: "knowledge_search"},
{ResponseType: "answer", FullAnswer: answer, Done: true, AnswerRunID: "step-2-answer", AnswerRunStart: len(narration)},
},
streamResult: weknora.ChatResult{RequestID: "request-interrupted", Answer: answer},
}
service, err := New(repo, fixedResolver{}, client, session.GroupShared)
if err != nil {
t.Fatal(err)
}
req := streamQARequest("message-interrupted")
req.ReasoningExecution = string(reasoningmode.ExecutionReasoning)
result, err := service.AskEvents(context.Background(), req, func(weknora.StreamEvent) {})
if err != nil {
t.Fatalf("AskEvents: %v", err)
}
if want := reasoningmode.TrimReasoningNarration(answer); result.Answer != want {
t.Fatalf("interrupted run must fall back to the heuristic: want %q, got %q", want, result.Answer)
}
}
+102
View File
@@ -0,0 +1,102 @@
package qa
import (
"context"
"fmt"
"net/http"
"net/http/httptest"
"os"
"strings"
"testing"
"time"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/reasoningmode"
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/weknora"
)
// TestReplayRecordedStreamDiagnostics replays a recorded WeKnora SSE stream
// (path in WK_RAW_STREAM) through the real client and prints what the narration
// observation sees. Temporary diagnostic, removed after the investigation.
func TestReplayRecordedStreamDiagnostics(t *testing.T) {
path := os.Getenv("WK_RAW_STREAM")
if path == "" {
t.Skip("WK_RAW_STREAM not set")
}
raw, err := os.ReadFile(path)
if err != nil {
t.Fatalf("read stream: %v", err)
}
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/event-stream")
_, _ = w.Write(raw)
}))
defer server.Close()
client, err := weknora.NewClient(server.URL, "test-key", 30*time.Second)
if err != nil {
t.Fatalf("client: %v", err)
}
var observation reasoningmode.NarrationObservation
runStarts := []string{}
eventTypes := map[string]int{}
observer := func(e weknora.StreamEvent) {
eventTypes[e.ResponseType]++
switch e.ResponseType {
case "answer":
if e.AnswerRunID != "" {
observation.Observed = true
}
if e.AnswerRunID != "" && e.AnswerRunID != lastRunID(runStarts) {
runStarts = append(runStarts, fmt.Sprintf("%s@%d", e.AnswerRunID, e.AnswerRunStart))
observation.AnswerRuns++
observation.LastRunStart = e.AnswerRunStart
observation.LastRunClean = true
}
case "thinking", "tool_call", "tool_result":
observation.ToolEvents++
observation.LastRunClean = false
}
}
result, err := client.AgentChatEvents(context.Background(), "session", weknora.AgentChatRequest{AgentID: "agent", Query: "q"}, observer)
if err != nil {
t.Fatalf("agent chat events: %v", err)
}
t.Logf("eventTypes=%v", eventTypes)
t.Logf("runs=%v", runStarts)
t.Logf("observation: observed=%v runs=%d tools=%d lastRunStart=%d lastRunClean=%v",
observation.Observed, observation.AnswerRuns, observation.ToolEvents, observation.LastRunStart, observation.LastRunClean)
content := result.Answer
observed := reasoningmode.TrimReasoningNarrationObserved(content, observation)
heuristic := reasoningmode.TrimReasoningNarration(content)
t.Logf("contentLen=%d observedKept=%d heuristicKept=%d", len(content), len(observed), len(heuristic))
if n := len(content) - len(observed); n > 0 {
t.Logf("observedDropPrefix=%q", firstRunes(content[:n], 160))
}
if n := len(content) - len(heuristic); n > 0 {
t.Logf("heuristicDropPrefix=%q", firstRunes(content[:n], 160))
}
t.Logf("observedKeptHead=%q", firstRunes(observed, 160))
}
func lastRunID(runs []string) string {
if len(runs) == 0 {
return ""
}
id := runs[len(runs)-1]
if i := strings.LastIndex(id, "@"); i >= 0 {
return id[:i]
}
return id
}
func firstRunes(s string, n int) string {
r := []rune(s)
if len(r) <= n {
return string(r)
}
return string(r[:n]) + "…"
}
+39 -19
View File
@@ -227,7 +227,7 @@ func (s *Service) ask(ctx context.Context, request core.QARequest, answerObserve
return s.weknora.UpdateSessionTitle(updateCtx, existingSessionID, title, "Managed by DobotAI; business state remains outside WeKnora")
})
query := localizedQuery(request.Query, request.ResponseLocale)
query := request.Query
if request.ConversationType == core.ConversationGroup && strings.TrimSpace(request.SenderName) != "" {
query = request.SenderName + ": " + query
}
@@ -238,15 +238,43 @@ func (s *Service) ask(ctx context.Context, request core.QARequest, answerObserve
chatRequest := weknora.AgentChatRequest{
Query: query, AgentID: executionAgentID, AgentEnabled: true, Channel: "im", EnableMemory: true,
Images: images,
// The language preference travels as an Accept-Language request
// header (see weknora.resolveLocale), never as prompt text: WeKnora's
// system prompt enforces "ALWAYS respond in {{language}}" which is
// fed by that header, so a prompt-prefix instruction loses to it.
Locale: strings.TrimSpace(request.ResponseLocale),
}
// Track whether this turn actually triggered a knowledge-base search by
// observing knowledge_search tool-call events. The flag feeds coverage
// classification (image turns) and is persisted with the turn.
var knowledgeSearched bool
// The stream itself separates the agent's step narration from the final
// answer body: every step publishes its answer chunks under its own answer
// run id (data.event_id), and the last run that no tool/thinking event
// interrupts is the final body. Track that structure so the delivered
// answer keeps the body's lead-in — and any image reference inside it —
// instead of guessing the boundary from the text.
var narrationObservation reasoningmode.NarrationObservation
lastAnswerRunID := ""
effectiveEventObserver := weknora.StreamObserver(func(e weknora.StreamEvent) {
if isKnowledgeTool(e.ToolName) {
knowledgeSearched = true
}
switch e.ResponseType {
case "answer":
if e.AnswerRunID != "" {
narrationObservation.Observed = true
if e.AnswerRunID != lastAnswerRunID {
lastAnswerRunID = e.AnswerRunID
narrationObservation.AnswerRuns++
narrationObservation.LastRunStart = e.AnswerRunStart
narrationObservation.LastRunClean = true
}
}
case "thinking", "tool_call", "tool_result":
narrationObservation.ToolEvents++
narrationObservation.LastRunClean = false
}
if eventObserver != nil {
eventObserver(e)
}
@@ -323,6 +351,16 @@ func (s *Service) ask(ctx context.Context, request core.QARequest, answerObserve
turn.WeKnoraRequestID = chat.RequestID
turn.KnowledgeSearched = knowledgeSearched
turn.AnswerMessageID = newID()
// A reasoning run reports every agent step's narrative through WeKnora's
// answer channel and appends the final answer body to that same
// accumulation. Keep the body only in the stored/delivered answer: the
// narration stays visible while the card streams (line by line) and is
// dropped once the final body starts.
if reasoningmode.IsReasoningExecution(request.ReasoningExecution) {
if trimmed := reasoningmode.TrimReasoningNarrationObserved(chat.Answer, narrationObservation); trimmed != chat.Answer {
turn.Answer, chat.Answer = trimmed, trimmed
}
}
if err := s.repository.FinishTurn(ctx, turn); err != nil {
return core.QAResult{}, fmt.Errorf("finish turn: %w", err)
}
@@ -373,24 +411,6 @@ func isKnowledgeTool(toolName string) bool {
}
}
func localizedQuery(query, locale string) string {
query = strings.TrimSpace(query)
switch strings.TrimSpace(locale) {
case "zh-CN":
return "请使用简体中文回答。\n\n" + query
case "en":
return "Answer in English.\n\n" + query
case "ja":
return "日本語で回答してください。\n\n" + query
case "ko":
return "한국어로 답변해 주세요.\n\n" + query
case "es":
return "Responde en español.\n\n" + query
default:
return query
}
}
func normalizeRequestIdentity(request core.QARequest) core.QARequest {
request.TenantKey = strings.TrimSpace(request.TenantKey)
request.Channel = strings.TrimSpace(request.Channel)
@@ -0,0 +1,90 @@
package reasoningmode
import "strings"
// NarrationObservation carries what the WeKnora event stream knows about the
// boundary between an agent step's narration and the final answer body.
//
// The text-only heuristic (BodyStartOffset) cannot tell a step narrative from
// an answer lead-in such as "我已经找到了 CRA5 的末端负载曲线图…下面给出完整回答。":
// both are plain prose, so the heuristic starts the body at the first
// structural line and silently drops everything before it — including answer
// prose, image references and table rows that sit in that lead-in.
//
// The stream knows the boundary: every agent step writes its narrative to the
// answer channel as its own answer run (WeKnora's data.event_id), and the
// final answer body is the last run that no tool/thinking event interrupts.
type NarrationObservation struct {
// Observed reports that the stream carried answer run information at all
// (WeKnora tags answer chunks with data.event_id). Without it the text
// heuristic stays in charge, so older deployments keep their behaviour.
Observed bool
// AnswerRuns counts distinct answer runs (consecutive chunks sharing one
// event id count as one run).
AnswerRuns int
// ToolEvents counts thinking / tool_call / tool_result events.
ToolEvents int
// LastRunStart is the byte offset in the accumulated answer where the
// most recent answer run started.
LastRunStart int
// LastRunClean reports that no tool/thinking event arrived after that run
// started, which makes the run the final answer body rather than another
// narrative step.
LastRunClean bool
}
// BodyStartOffset returns the observed body start and whether the observation
// is usable. An answer with no tool event carries no narration at all, so the
// whole answer is the body.
func (o NarrationObservation) BodyStartOffset(contentLength int) (int, bool) {
if !o.Observed {
return 0, false
}
// A single answer run carries the whole answer: the agent published no
// separate step narration, so there is nothing that can be cut safely.
if o.AnswerRuns <= 1 {
return 0, true
}
if o.ToolEvents == 0 {
return 0, true
}
if o.LastRunClean && o.LastRunStart > 0 && o.LastRunStart < contentLength {
return o.LastRunStart, true
}
return 0, false
}
// SplitReasoningAnswerObserved returns the final answer body using the stream
// observation when it is usable and the text heuristic otherwise. ok is false
// when the content carries no body boundary (nothing must be hidden).
func SplitReasoningAnswerObserved(content string, observation NarrationObservation) (string, bool) {
if offset, ok := observation.BodyStartOffset(len(content)); ok {
if offset <= 0 {
// The stream showed no narration at all: the whole answer is body.
return content, true
}
// An image reference in front of the boundary is answer content (the
// agent published a picture in its lead-in), never step narration, so
// the safe answer is to keep everything.
if strings.Contains(content[:offset], "![") {
return content, true
}
body := strings.TrimLeft(content[offset:], "\r\n")
if strings.TrimSpace(body) != "" {
return body, true
}
}
return SplitReasoningAnswer(content)
}
// TrimReasoningNarrationObserved drops the step narration using the stream
// observation, falling back to the text heuristic when the stream carried no
// usable boundary. Unlike the heuristic it never drops the answer lead-in or
// an image reference that belongs to the body.
func TrimReasoningNarrationObserved(content string, observation NarrationObservation) string {
body, ok := SplitReasoningAnswerObserved(content, observation)
if !ok {
return content
}
return body
}
@@ -0,0 +1,127 @@
package reasoningmode
import (
"strings"
"testing"
)
// The production case this observation fixes: the agent narrates its steps,
// then the final answer opens with a lead-in ("我已经找到了…下面给出完整回答。"),
// carries the requested load-curve image, and only then reaches its first
// heading. The text heuristic started the body at the heading and dropped the
// lead-in plus the image reference with it.
const craLoadCurveAnswer = "我需要先确认 CRA5 与 CR5A 是同一型号,然后检索负载曲线。\n" +
"知识库检索命中 3 篇文档,其中用户手册含末端负载曲线图。\n" +
"我已经找到了 CRA5(CR5A)的末端负载曲线图,并确认了负载曲线的读法说明。下面给出完整回答。\n\n" +
"CRA5(即 CR5A)的末端负载曲线如下:\n\n" +
"![CR5A 末端负载曲线](resource://FsVe56UYMz64sm8NsUZNrg)\n\n" +
"---\n\n" +
"## 一、负载曲线怎么读\n\n" +
"负载曲线反映的是:末端负载能力与负载质心偏心距离成负相关。\n"
func TestNarrationObservationKeepsAnswerLeadInAndImage(t *testing.T) {
narration := strings.Index(craLoadCurveAnswer, "我已经找到了")
observation := NarrationObservation{
Observed: true, AnswerRuns: 3, ToolEvents: 5,
LastRunStart: narration, LastRunClean: true,
}
body := TrimReasoningNarrationObserved(craLoadCurveAnswer, observation)
if !strings.HasPrefix(body, "我已经找到了") {
t.Fatalf("body must start at the answer lead-in, got %q", firstLine(body))
}
if !strings.Contains(body, "![CR5A 末端负载曲线](resource://FsVe56UYMz64sm8NsUZNrg)") {
t.Fatalf("body must keep the image reference, got %q", firstLine(body))
}
if strings.Contains(body, "我需要先确认") {
t.Fatalf("body must not contain the step narration, got %q", firstLine(body))
}
// The text-only heuristic is what dropped the lead-in and the image.
heuristic := TrimReasoningNarration(craLoadCurveAnswer)
if !strings.HasPrefix(heuristic, "## 一、负载曲线怎么读") {
t.Fatalf("heuristic is expected to start at the heading, got %q", firstLine(heuristic))
}
if strings.Contains(heuristic, "resource://") {
t.Fatalf("heuristic is expected to drop the image reference")
}
}
func TestNarrationObservationFallsBackWithoutUsableBoundary(t *testing.T) {
narration := strings.Index(craLoadCurveAnswer, "我已经找到了")
tests := []struct {
name string
observation NarrationObservation
}{
{"not observed", NarrationObservation{AnswerRuns: 3, ToolEvents: 5, LastRunStart: narration, LastRunClean: true}},
{"last run interrupted by a tool", NarrationObservation{Observed: true, AnswerRuns: 3, ToolEvents: 5, LastRunStart: narration}},
{"multi run starting at zero", NarrationObservation{Observed: true, AnswerRuns: 3, ToolEvents: 5, LastRunStart: 0, LastRunClean: true}},
{"run start beyond the answer", NarrationObservation{Observed: true, AnswerRuns: 3, ToolEvents: 5, LastRunStart: len(craLoadCurveAnswer) + 10, LastRunClean: true}},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
got := TrimReasoningNarrationObserved(craLoadCurveAnswer, tc.observation)
if want := TrimReasoningNarration(craLoadCurveAnswer); got != want {
t.Fatalf("expected heuristic fallback %q, got %q", firstLine(want), firstLine(got))
}
})
}
}
func TestNarrationObservationKeepsImageLeadIn(t *testing.T) {
// The reported production shape: the lead-in carries the answer prose plus
// the picture, and the boundary observed by the stream only lands after it.
content := "根据知识库中《末端负载曲线-CRA系列》文档,以下是 **CR5A 的末端负载曲线图**:\n\n" +
"![CR5A 末端负载曲线](resource://FsVe56UYMz64sm8NsUZNrg)\n\n" +
"### 如何查看该负载曲线\n负载能力与偏心距离成负相关。\n"
offset := strings.Index(content, "### 如何查看该负载曲线")
observation := NarrationObservation{Observed: true, AnswerRuns: 4, ToolEvents: 6, LastRunStart: offset, LastRunClean: true}
if got := TrimReasoningNarrationObserved(content, observation); got != content {
t.Fatalf("image lead-in must survive the trim, got %q", got)
}
}
func TestNarrationObservationKeepsSingleRunAnswer(t *testing.T) {
// The quick-answer agent publishes the whole answer as a single answer run
// (with tool_call/tool_result steps around it). Nothing in that stream is
// narration, so the lead-in and the image must survive.
observation := NarrationObservation{Observed: true, AnswerRuns: 1, ToolEvents: 4, LastRunStart: 0, LastRunClean: true}
got := TrimReasoningNarrationObserved(craLoadCurveAnswer, observation)
if got != craLoadCurveAnswer {
t.Fatalf("single answer run must keep the whole answer, got %q", firstLine(got))
}
if !strings.Contains(got, "resource://") {
t.Fatalf("the image reference must survive for a single answer run")
}
}
func TestNarrationObservationKeepsWholeAnswerWithoutToolSteps(t *testing.T) {
answer := "本机提供以下三种接线方式:\n\n| 方式 | 说明 |\n| --- | --- |\n| A | 直连 |\n"
observation := NarrationObservation{Observed: true, AnswerRuns: 1, ToolEvents: 0}
body, ok := SplitReasoningAnswerObserved(answer, observation)
if !ok {
t.Fatalf("a run without tool steps has no narration, so the whole answer is the body")
}
if body != answer {
t.Fatalf("expected the whole answer, got %q", body)
}
}
func TestNarrationObservationTrimsBlankLineAtRunStart(t *testing.T) {
content := "叙述步骤。\n\n\n## 标题\n\n正文。\n"
observation := NarrationObservation{Observed: true, AnswerRuns: 2, ToolEvents: 1,
LastRunStart: strings.Index(content, "\n\n\n##"), LastRunClean: true}
body := TrimReasoningNarrationObserved(content, observation)
if body != "## 标题\n\n正文。\n" {
t.Fatalf("expected the blank lines before the run content to be trimmed, got %q", body)
}
}
func firstLine(s string) string {
if i := strings.IndexByte(s, '\n'); i >= 0 {
return s[:i]
}
return s
}
+116
View File
@@ -0,0 +1,116 @@
package reasoningmode
import (
"regexp"
"strings"
)
// Reasoning agent runs report every intermediate step through WeKnora's
// response_type=answer channel: the step narratives are concatenated without
// any separator and the final answer body is appended to that same
// accumulation. A card update therefore shows one run-on "narration + body"
// blob. These helpers locate the boundary so callers can hide the narration
// and keep the body only.
//
// Calibration against 118 production answers of the built-in smart-reasoning
// agent: 106 had a structural first body line (97 heading, 4 bold-only line,
// 3 table row, 2 blockquote) and every one of them was directly preceded by a
// blank line; the remaining 12 answers were greetings without narration. The
// hidden narration was a short lead-in (median 111 chars, max 526 chars).
var (
answerHeadingLineRE = regexp.MustCompile(`^#{1,6}[\t ]+\S`)
answerBoldLineRE = regexp.MustCompile(`^\*\*[^*\n]+\*\*[\t ]*$`)
answerTableRowRE = regexp.MustCompile(`^\|.*\|[\t ]*$`)
answerQuoteLineRE = regexp.MustCompile(`^>[\t ]*\S`)
answerFenceLineRE = regexp.MustCompile("^(`{3,}|~{3,})")
)
// IsReasoningExecution reports whether execution runs the multi-step reasoning
// agent whose answer stream mixes step narration with the final answer body.
func IsReasoningExecution(execution string) bool {
switch ExecutionMode(strings.TrimSpace(execution)) {
case ExecutionReasoning, ExecutionReasoningSingle:
return true
}
return false
}
// SplitReasoningAnswer returns the final answer body. ok is false when the
// content carries no body boundary yet (nothing must be hidden in that case).
func SplitReasoningAnswer(content string) (string, bool) {
offset, ok := BodyStartOffset(content)
if !ok {
return content, false
}
return content[offset:], true
}
// TrimReasoningNarration drops the step narration that precedes the final
// answer body. Content without a body boundary is returned unchanged.
func TrimReasoningNarration(content string) string {
body, ok := SplitReasoningAnswer(content)
if !ok {
return content
}
return body
}
// BodyStartOffset returns the byte offset where the final answer body starts.
// A line qualifies as the body start when it is a structural Markdown line
// (heading, bold-only line, table row or blockquote) and is not inside a
// fenced code block, either directly following a blank line or being the very
// first non-blank line of the content.
//
// The first-line rule matters: 31 of 118 production reasoning answers open
// directly with their body ("# 标题" first). Without it the scan would skip
// that opening line and fire on the *next* structural line, so finalize
// deleted the answer's title plus its opening paragraph (up to 526 chars) while
// the streaming feed, whose display inserts blank lines between steps, kept
// them - mid-stream and finalize text disagreed.
func BodyStartOffset(content string) (int, bool) {
lines := strings.Split(content, "\n")
fence := ""
offset := 0
previousBlank := false
seenContent := false
for i, raw := range lines {
trimmed := strings.TrimSpace(raw)
if marker := answerFenceLineRE.FindString(trimmed); marker != "" &&
(fence == "" || strings.HasPrefix(trimmed, fence)) {
if fence == "" {
fence = marker
} else {
fence = ""
}
seenContent = true
previousBlank = false
offset += len(raw) + 1
continue
}
if fence == "" {
if !seenContent && isAnswerBodyLine(trimmed) {
return offset, true
}
if i > 0 && previousBlank && isAnswerBodyLine(trimmed) {
return offset, true
}
}
if trimmed != "" {
seenContent = true
}
previousBlank = trimmed == ""
offset += len(raw) + 1
}
return 0, false
}
func isAnswerBodyLine(trimmed string) bool {
if trimmed == "" {
return false
}
return answerHeadingLineRE.MatchString(trimmed) ||
answerBoldLineRE.MatchString(trimmed) ||
answerTableRowRE.MatchString(trimmed) ||
answerQuoteLineRE.MatchString(trimmed)
}
+116
View File
@@ -0,0 +1,116 @@
package reasoningmode
import (
"strings"
"testing"
)
func TestBodyStartOffsetRecognizesStructuralBodyLines(t *testing.T) {
leadIn := "根据检索到的资料,关于 CR20V 各关节电机最大旋转扭矩的说明如下:"
heading := "## CR20V 关节电机最大旋转扭矩"
tableBody := "**Magician 传感器套件(Magician Sensor Kit)重量如下:**\n\n| 项目 | 重量 |\n| --- | --- |\n| 整机 | 1.2 kg |"
rowBody := "| 关节 | 扭矩 |\n| --- | --- |"
quoteBody := "> 提示:以下参数来自产品手册。"
tests := []struct {
name string
answer string
want string
ok bool
}{
{
name: "narration then heading",
answer: "我已经收集了充分的信息。现在我可以明确回答用户的问题了。\n\n关键发现:\n1. 手册给出额定扭矩\n\n需要向用户说明\n\n" + heading + "\n\n各关节参数如下。",
want: heading + "\n\n各关节参数如下。",
ok: true,
},
{
name: "lead-in then bold body line",
answer: "我已经获取到完整且明确的信息,可以给出准确答案了。\n\n" + tableBody,
want: tableBody,
ok: true,
},
{
name: "short lead-in then table",
answer: leadIn + "\n\n" + rowBody,
want: rowBody,
ok: true,
},
{
name: "narration then blockquote",
answer: "我需要先确认使用场景。\n\n" + quoteBody,
want: quoteBody,
ok: true,
},
{
name: "greeting has no body boundary",
answer: "你好!我是越疆智能助手,可以帮你查询机械臂资料。",
want: "你好!我是越疆智能助手,可以帮你查询机械臂资料。",
ok: false,
},
{
name: "answer opens with its body: nothing is hidden",
answer: heading + "\n\n各关节参数如下。",
want: heading + "\n\n各关节参数如下。",
ok: true,
},
{
name: "title and opening paragraph survive finalize",
answer: "# 200万高职具身智能实训室搭配方案\n\n结合越疆现有的一体化实训平台,围绕具身智能教学场景,我们从教具、软件、课程三方面给出搭配建议。\n\n## 一、先明确三条搭配原则\n\n- 匹配课程大纲\n- 控制预算\n",
want: "# 200万高职具身智能实训室搭配方案\n\n结合越疆现有的一体化实训平台,围绕具身智能教学场景,我们从教具、软件、课程三方面给出搭配建议。\n\n## 一、先明确三条搭配原则\n\n- 匹配课程大纲\n- 控制预算\n",
ok: true,
},
{
name: "leading blank lines before the title",
answer: "\n\n# 标题\n\n正文。",
want: "# 标题\n\n正文。",
ok: true,
},
{
name: "structural line inside a fence is ignored",
answer: "我先确认一下配置项。\n\n```\n## 这是代码里的注释\n| a | b |\n```\n\n参考上面的写法。",
want: "我先确认一下配置项。\n\n```\n## 这是代码里的注释\n| a | b |\n```\n\n参考上面的写法。",
ok: false,
},
{
name: "structural line glued to narration is not a boundary",
answer: "需要向用户说明" + heading,
want: "需要向用户说明" + heading,
ok: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
body, ok := SplitReasoningAnswer(tt.answer)
if ok != tt.ok {
t.Fatalf("ok = %v, want %v", ok, tt.ok)
}
if body != tt.want {
t.Fatalf("body = %q, want %q", body, tt.want)
}
if !strings.HasSuffix(tt.answer, body) {
t.Fatalf("body %q is not a suffix of the answer: the rule ate content", body)
}
if trimmed := TrimReasoningNarration(tt.answer); trimmed != tt.want {
t.Fatalf("TrimReasoningNarration = %q, want %q", trimmed, tt.want)
}
})
}
}
func TestIsReasoningExecution(t *testing.T) {
tests := map[string]bool{
"reasoning": true,
" reasoning ": true,
"reasoning_single": true,
"quick": false,
"": false,
"something": false,
}
for execution, want := range tests {
if got := IsReasoningExecution(execution); got != want {
t.Fatalf("IsReasoningExecution(%q) = %v, want %v", execution, got, want)
}
}
}
@@ -0,0 +1,108 @@
package reasoningmode
import (
"bufio"
"encoding/json"
"os"
"testing"
)
// TestReplayProductionCorpusThroughFeed replays production reasoning answers
// through the streaming feed the card pump actually uses, and checks two
// properties of every frame the card receives:
//
// 1. a frame that already contains a structural body line must start at that
// body (no narration glued in front of it). A frame that shows narration
// above the body is exactly the "the last reasoning sentence stays on the
// card" defect.
// 2. once a frame is body-only, no later frame may go back to showing
// narration.
//
// Enable with REASONING_CORPUS=<path to reasoning_corpus.jsonl>.
func TestReplayProductionCorpusThroughFeed(t *testing.T) {
path := os.Getenv("REASONING_CORPUS")
if path == "" {
t.Skip("set REASONING_CORPUS to replay production answers")
}
file, err := os.Open(path)
if err != nil {
t.Fatalf("open corpus: %v", err)
}
defer file.Close()
chunkSizes := []int{11, 47, 199, 1001}
var rows, frames, bodyFrames int
type violation struct {
id string
size int
marks bool
prefix int
bytes int
}
var violations []violation
scanner := bufio.NewScanner(file)
scanner.Buffer(make([]byte, 0, 1<<20), 1<<22)
for scanner.Scan() {
var row struct {
ID string `json:"id"`
Content string `json:"content"`
}
if err := json.Unmarshal(scanner.Bytes(), &row); err != nil {
t.Fatalf("decode row: %v", err)
}
rows++
for _, size := range chunkSizes {
for _, marks := range []bool{false, true} {
feed := NewPartialFeed("reasoning")
bodySeen := false
for start := 0; start < len(row.Content); start += size {
end := start + size
if end > len(row.Content) {
end = len(row.Content)
}
frame := feed.OnAnswer(row.Content[:end])
frames++
if firstStructuralLine(frame) == "" {
// no body in this frame yet: narration only, expected.
continue
}
offset, ok := BodyStartOffset(frame)
prefix := -1
if ok {
prefix = len([]rune(frame[:offset]))
}
if ok && prefix == 0 {
bodySeen = true
bodyFrames++
continue
}
// The frame carries a structural line but does not start at
// it: narration is still visible above the body.
violations = append(violations, violation{
id: row.ID,
size: size,
marks: marks,
prefix: prefix,
bytes: len(frame),
})
_ = bodySeen
}
}
}
}
if err := scanner.Err(); err != nil {
t.Fatalf("scan corpus: %v", err)
}
t.Logf("rows=%d frames=%d body_only_frames=%d narration_above_body_frames=%d", rows, frames, bodyFrames, len(violations))
for i, v := range violations {
if i >= 8 {
t.Logf("... %d more", len(violations)-i)
break
}
t.Logf("violation id=%s chunk=%d step_boundaries=%t body_prefix_runes=%d bytes=%d", v.id, v.size, v.marks, v.prefix, v.bytes)
}
if len(violations) > 0 {
t.Errorf("%d frames showed narration above the answer body", len(violations))
}
}
@@ -0,0 +1,92 @@
package reasoningmode
import (
"bufio"
"encoding/json"
"os"
"strings"
"testing"
)
// TestReplayProductionCorpus replays real production answers of the built-in
// smart-reasoning agent through the body-boundary rule.
// Enable with REASONING_CORPUS=<path to reasoning_corpus.jsonl>.
func TestReplayProductionCorpus(t *testing.T) {
path := os.Getenv("REASONING_CORPUS")
if path == "" {
t.Skip("set REASONING_CORPUS to replay production answers")
}
file, err := os.Open(path)
if err != nil {
t.Fatalf("open corpus: %v", err)
}
defer file.Close()
var total, trimmed int
scanner := bufio.NewScanner(file)
scanner.Buffer(make([]byte, 0, 1<<20), 1<<22)
for scanner.Scan() {
var row struct {
ID string `json:"id"`
Content string `json:"content"`
}
if err := json.Unmarshal(scanner.Bytes(), &row); err != nil {
t.Fatalf("decode row: %v", err)
}
total++
body, ok := SplitReasoningAnswer(row.Content)
if !ok {
continue
}
trimmed++
hidden := strings.TrimSpace(strings.TrimSuffix(row.Content, body))
t.Logf("id=%s hidden=%d body=%d head=%q", row.ID[:8], len([]rune(hidden)), len([]rune(body)), shorten(hidden, 60))
// Guard: the hidden prefix must be narration only. A structural line in
// there means the rule fired past the body start and ate answer content
// (that is exactly how the opening title + paragraph of 31 answers were
// deleted before the first-line rule existed).
if line := firstStructuralLine(hidden); line != "" {
t.Errorf("id=%s hidden prefix carries a structural body line %q: the rule ate answer content", row.ID[:8], shorten(line, 40))
}
if runes := len([]rune(hidden)); runes > maxHiddenNarrationRunes {
t.Errorf("id=%s hidden prefix is %d runes, over the %d-rune narration bound", row.ID[:8], runes, maxHiddenNarrationRunes)
}
}
if err := scanner.Err(); err != nil {
t.Fatalf("scan corpus: %v", err)
}
t.Logf("total=%d narration_trimmed=%d unchanged=%d", total, trimmed, total-trimmed)
}
func shorten(value string, limit int) string {
runes := []rune(value)
if len(runes) <= limit {
return value
}
return string(runes[len(runes)-limit:])
}
// maxHiddenNarrationRunes bounds the narration the rule may hide. Calibration
// on 118 production answers: the hidden lead-in was at most 526 runes.
const maxHiddenNarrationRunes = 600
// firstStructuralLine returns the first line that qualifies as answer body
// (heading, bold-only line, table row or blockquote), skipping fenced code.
func firstStructuralLine(text string) string {
fence := ""
for _, raw := range strings.Split(text, "\n") {
trimmed := strings.TrimSpace(raw)
if marker := answerFenceLineRE.FindString(trimmed); marker != "" &&
(fence == "" || strings.HasPrefix(trimmed, fence)) {
if fence == "" {
fence = marker
} else {
fence = ""
}
continue
}
if fence == "" && isAnswerBodyLine(trimmed) {
return trimmed
}
}
return ""
}
+86
View File
@@ -0,0 +1,86 @@
package reasoningmode
import "strings"
// PartialFeed assembles the streaming card content of one turn.
//
// A reasoning run streams every agent step's narrative through the answer
// channel and appends the final answer body to the same accumulation, so the
// raw text reaches the card as one run-on paragraph. The feed
//
// - joins step segments with a blank line, so the card shows the reasoning
// line by line instead of glued sentences, and
// - switches to the final body only as soon as the body starts. The DingTalk
// card update replaces the card content, so that switch is what the user
// sees as the "refresh": the reasoning block disappears and the body is
// streamed on its own.
//
// Quick executions are a pass-through and keep the previous behaviour.
type PartialFeed struct {
reasoning bool
seen string
display string
stepPending bool
bodyOnly string
bodyStarted bool
started bool
}
// NewPartialFeed builds the feed for one turn's execution mode.
func NewPartialFeed(execution string) *PartialFeed {
return &PartialFeed{reasoning: IsReasoningExecution(execution)}
}
// Started reports whether any answer text has been streamed yet.
func (f *PartialFeed) Started() bool {
return f.started
}
// StepBoundary marks the end of an agent step (tool call or tool result) so
// the next narrative segment starts on a new line.
func (f *PartialFeed) StepBoundary() {
if !f.reasoning {
return
}
f.stepPending = true
}
// OnAnswer records the accumulated answer text reported by the stream and
// returns the content the card must show right now.
func (f *PartialFeed) OnAnswer(fullAnswer string) string {
if !f.reasoning {
if strings.TrimSpace(fullAnswer) != "" {
f.started = true
}
return fullAnswer
}
delta := fullAnswer
switch {
case strings.HasPrefix(fullAnswer, f.seen):
delta = fullAnswer[len(f.seen):]
case fullAnswer != f.seen:
// The stream restarted (new WeKnora session or a retry): treat the
// text as a fresh segment instead of gluing it to the previous one.
f.stepPending = true
}
f.seen = fullAnswer
if delta != "" {
f.started = true
if f.stepPending && strings.TrimSpace(f.display) != "" {
f.display += "\n\n"
}
f.stepPending = false
f.display += delta
}
if body, ok := SplitReasoningAnswer(f.display); ok {
f.bodyStarted = true
f.bodyOnly = body
}
if f.bodyStarted {
return f.bodyOnly
}
return f.display
}
@@ -0,0 +1,87 @@
package reasoningmode
import (
"strings"
"testing"
)
func TestPartialFeedJoinsReasoningSegmentsAndRefreshesToBody(t *testing.T) {
feed := NewPartialFeed(string(ExecutionReasoning))
// Step 1: first narrative segment streams in.
step1 := "我先查一下知识库。"
if got := feed.OnAnswer(step1); got != step1 {
t.Fatalf("first segment = %q, want %q", got, step1)
}
if !feed.Started() {
t.Fatal("Started must be true after the first answer text")
}
// Step 1 ends with a tool call, step 2 starts with its own narrative.
feed.StepBoundary()
step2 := "找到了相关文档,我确认一下参数。"
got := feed.OnAnswer(step1 + step2)
want := step1 + "\n\n" + step2
if got != want {
t.Fatalf("second segment = %q, want %q (segments must be line separated)", got, want)
}
// Step 2 ends, step 3 carries the final answer body.
feed.StepBoundary()
body := "## CR20V 关节电机最大旋转扭矩\n\n| 关节 | 扭矩 |"
got = feed.OnAnswer(step1 + step2 + body)
if got != body {
t.Fatalf("body-only content = %q, want %q (reasoning must be hidden)", got, body)
}
// The body keeps streaming; the visible content stays body-only.
grown := body + "\n| J1 | 100 N·m |"
if got := feed.OnAnswer(step1 + step2 + grown); got != grown {
t.Fatalf("grown body = %q, want %q", got, grown)
}
if strings.Contains(got, "我先查一下知识库") {
t.Fatalf("reasoning leaked into the refreshed card: %q", got)
}
}
func TestPartialFeedKeepsGluedNarrationReadable(t *testing.T) {
// WeKnora concatenates step narratives without any separator; the feed must
// not let the card show them as one run-on sentence.
feed := NewPartialFeed(string(ExecutionReasoning))
first := "我正在理解问题。"
feed.OnAnswer(first)
feed.StepBoundary()
second := "正在检索知识库。"
if got := feed.OnAnswer(first + second); got != first+"\n\n"+second {
t.Fatalf("glued narration = %q", got)
}
}
func TestPartialFeedStartsANewSegmentWhenTheStreamRestarts(t *testing.T) {
feed := NewPartialFeed(string(ExecutionReasoning))
feed.OnAnswer("第一段推理。")
got := feed.OnAnswer("重新开始的一段推理。")
if got != "第一段推理。\n\n重新开始的一段推理。" {
t.Fatalf("restart segment = %q", got)
}
}
func TestPartialFeedQuickExecutionIsPassThrough(t *testing.T) {
feed := NewPartialFeed(string(ExecutionQuick))
narrativeAndBody := "我已收集信息。\n\n## 标题\n\n正文"
if got := feed.OnAnswer(narrativeAndBody); got != narrativeAndBody {
t.Fatalf("quick execution = %q, want the raw stream unchanged", got)
}
feed.StepBoundary()
if got := feed.OnAnswer(narrativeAndBody); got != narrativeAndBody {
t.Fatalf("quick execution after boundary = %q, want the raw stream unchanged", got)
}
}
func TestPartialFeedWithoutBoundaryKeepsShowingReasoning(t *testing.T) {
feed := NewPartialFeed(string(ExecutionReasoning))
greeting := "你好!我是越疆智能助手。"
if got := feed.OnAnswer(greeting); got != greeting {
t.Fatalf("greeting = %q, want unchanged", got)
}
}
+57 -8
View File
@@ -49,11 +49,28 @@ type Service struct {
repo Repository
quickAgentID string
reasoningAgentID string
defaultMode Mode
now func() time.Time
idleTimeout time.Duration
}
func NewService(repo Repository, quickAgentID, reasoningAgentID string) (*Service, error) {
// Option customises a reasoning mode Service.
type Option func(*Service) error
// WithDefaultMode sets the mode applied to users who never explicitly chose
// one with /think. Passing ModeQuick (the default when unset) preserves the
// historical behaviour of answering with the quick agent.
func WithDefaultMode(mode Mode) Option {
return func(s *Service) error {
if mode != ModeQuick && mode != ModeReasoning {
return fmt.Errorf("invalid default reasoning mode %q", mode)
}
s.defaultMode = mode
return nil
}
}
func NewService(repo Repository, quickAgentID, reasoningAgentID string, opts ...Option) (*Service, error) {
quickAgentID = strings.TrimSpace(quickAgentID)
reasoningAgentID = strings.TrimSpace(reasoningAgentID)
if repo == nil {
@@ -65,13 +82,29 @@ func NewService(repo Repository, quickAgentID, reasoningAgentID string) (*Servic
if quickAgentID == reasoningAgentID {
return nil, errors.New("quick and reasoning agent IDs must differ")
}
return &Service{
service := &Service{
repo: repo,
quickAgentID: quickAgentID,
reasoningAgentID: reasoningAgentID,
defaultMode: ModeQuick,
now: time.Now,
idleTimeout: defaultIdleTimeout,
}, nil
}
for _, opt := range opts {
if err := opt(service); err != nil {
return nil, err
}
}
return service, nil
}
// effectivelyReasoning reports whether a turn should be served by the
// reasoning agent given the resolved state and the configured default mode.
func (s *Service) effectivelyReasoning(state State) bool {
if state.Mode == ModeReasoning {
return true
}
return !state.UserSelected && s.defaultMode == ModeReasoning
}
func (s *Service) Prepare(ctx context.Context, scope Scope, text string) (Decision, error) {
@@ -95,10 +128,14 @@ func (s *Service) Prepare(ctx context.Context, scope Scope, text string) (Decisi
state.MaxRounds,
)}, nil
case CommandDisable:
_, wasReasoning, err := s.repo.Disable(ctx, scope, s.now())
previous, _, err := s.repo.ResolveState(ctx, scope, s.now())
if err != nil {
return Decision{}, err
}
wasReasoning := s.effectivelyReasoning(previous)
if _, _, err = s.repo.Disable(ctx, scope, s.now()); err != nil {
return Decision{}, err
}
if !wasReasoning {
return Decision{Handled: true, Reply: "当前未开启推理模式,默认是快速问答模式。"}, nil
}
@@ -106,7 +143,7 @@ func (s *Service) Prepare(ctx context.Context, scope Scope, text string) (Decisi
case CommandSingle:
return Decision{Query: command.Query, AgentID: s.reasoningAgentID, Execution: ExecutionReasoningSingle}, nil
case CommandHelp:
return Decision{Handled: true, Reply: helpReply()}, nil
return Decision{Handled: true, Reply: helpReply(s.defaultMode)}, nil
default:
return Decision{Handled: true, Reply: "用法:/think <问题>、/think on、/think off。"}, nil
}
@@ -119,13 +156,25 @@ func (s *Service) Prepare(ctx context.Context, scope Scope, text string) (Decisi
if state.Mode == ModeReasoning {
return Decision{Query: query, AgentID: s.reasoningAgentID, Execution: ExecutionReasoning}, nil
}
// The user never chose a mode explicitly: follow the operator default.
// Default-mode reasoning runs as a single reasoning turn so it never
// enters the rounds/idle-expiry machinery; the user can pin quick mode
// any time with /think off.
if s.effectivelyReasoning(state) {
return Decision{Query: query, AgentID: s.reasoningAgentID, Execution: ExecutionReasoningSingle}, nil
}
return Decision{Query: query, AgentID: s.quickAgentID, Execution: ExecutionQuick, Expired: expired}, nil
}
func helpReply() string {
func helpReply(defaultMode Mode) string {
defaultLabel := "默认模式:快速问答模式,基于知识库快速准确作答,响应快、成本低,适合信息查询和事实确认。\n" +
"智能推理模式:支持多步思考与工具调用,计费标准更高,适合复杂分析和推导。\n"
if defaultMode == ModeReasoning {
defaultLabel = "默认模式:智能推理模式,支持多步思考与工具调用,效果更强、计费标准更高。\n" +
"快速问答模式:基于知识库快速准确作答,响应快、成本低,可用 /think off 切换并保持。\n"
}
return "一、模式说明\n" +
"默认模式:快速问答模式,基于知识库快速准确作答,响应快、成本低,适合信息查询和事实确认。\n" +
"智能推理模式:支持多步思考与工具调用,计费标准更高,适合复杂分析和推导。\n\n" +
defaultLabel + "\n" +
"二、模式切换指令\n" +
"/think + 你的问题:仅当前问题使用智能推理\n" +
"/think on:开启持续智能推理模式\n" +
+86
View File
@@ -23,6 +23,7 @@ func (f *fakeRepository) ResolveState(context.Context, Scope, time.Time) (State,
func (f *fakeRepository) Enable(context.Context, Scope, time.Time, time.Duration) (State, error) {
f.enableCalls++
f.state.Mode = ModeReasoning
f.state.UserSelected = true
f.state.RoundsUsed = 0
return f.state, nil
}
@@ -31,6 +32,7 @@ func (f *fakeRepository) Disable(context.Context, Scope, time.Time) (State, bool
f.disableCalls++
wasReasoning := f.state.Mode == ModeReasoning
f.state.Mode = ModeQuick
f.state.UserSelected = true
f.state.RoundsUsed = 0
return f.state, wasReasoning, nil
}
@@ -177,3 +179,87 @@ func TestPrepareReturnsLocalHelpWithoutStateMutation(t *testing.T) {
t.Fatalf("help mutated mode: %+v", repo)
}
}
func newTestServiceWithDefault(t *testing.T, repo Repository, mode Mode) *Service {
t.Helper()
service, err := NewService(repo, "builtin-quick-answer", "builtin-smart-reasoning", WithDefaultMode(mode))
if err != nil {
t.Fatal(err)
}
service.now = func() time.Time { return time.Date(2026, 9, 15, 12, 0, 0, 0, time.UTC) }
return service
}
func TestNewServiceRejectsInvalidDefaultMode(t *testing.T) {
if _, err := NewService(&fakeRepository{}, "q", "r", WithDefaultMode("turbo")); err == nil {
t.Fatal("expected invalid default mode to fail")
}
}
func TestPrepareDefaultReasoningRoutesUnselectedUserToReasoningSingle(t *testing.T) {
repo := &fakeRepository{state: DefaultState()}
decision, err := newTestServiceWithDefault(t, repo, ModeReasoning).Prepare(context.Background(), testScope(), "普通问题")
if err != nil {
t.Fatal(err)
}
if decision.AgentID != "builtin-smart-reasoning" || decision.Execution != ExecutionReasoningSingle {
t.Fatalf("decision=%+v", decision)
}
}
func TestPrepareDefaultReasoningHonoursUserQuickSelection(t *testing.T) {
state := DefaultState()
state.UserSelected = true
decision, err := newTestServiceWithDefault(t, &fakeRepository{state: state}, ModeReasoning).
Prepare(context.Background(), testScope(), "普通问题")
if err != nil {
t.Fatal(err)
}
if decision.AgentID != "builtin-quick-answer" || decision.Execution != ExecutionQuick {
t.Fatalf("decision=%+v", decision)
}
}
func TestPrepareDefaultQuickKeepsLegacyBehaviour(t *testing.T) {
decision, err := newTestServiceWithDefault(t, &fakeRepository{state: DefaultState()}, ModeQuick).
Prepare(context.Background(), testScope(), "普通问题")
if err != nil {
t.Fatal(err)
}
if decision.AgentID != "builtin-quick-answer" || decision.Execution != ExecutionQuick {
t.Fatalf("decision=%+v", decision)
}
}
func TestThinkOffPinsQuickModeUnderReasoningDefault(t *testing.T) {
repo := &fakeRepository{state: DefaultState()}
service := newTestServiceWithDefault(t, repo, ModeReasoning)
decision, err := service.Prepare(context.Background(), testScope(), "/think off")
if err != nil {
t.Fatal(err)
}
if !decision.Handled || !strings.Contains(decision.Reply, "切回") {
t.Fatalf("decision=%+v", decision)
}
if !repo.state.UserSelected || repo.state.Mode != ModeQuick {
t.Fatalf("state=%+v", repo.state)
}
followUp, err := service.Prepare(context.Background(), testScope(), "后续问题")
if err != nil {
t.Fatal(err)
}
if followUp.AgentID != "builtin-quick-answer" {
t.Fatalf("follow-up escaped user quick selection: %+v", followUp)
}
}
func TestHelpReflectsReasoningDefaultMode(t *testing.T) {
decision, err := newTestServiceWithDefault(t, &fakeRepository{state: DefaultState()}, ModeReasoning).
Prepare(context.Background(), testScope(), "/help")
if err != nil {
t.Fatal(err)
}
if !strings.Contains(decision.Reply, "默认模式:智能推理模式") {
t.Fatalf("help=%q", decision.Reply)
}
}
+1
View File
@@ -22,6 +22,7 @@ var ErrInvalidState = errors.New("invalid reasoning mode state")
type State struct {
Mode Mode
UserSelected bool
MaxRounds int
RoundsUsed int
HintEnabled bool
+8 -6
View File
@@ -61,6 +61,7 @@ func (s *Store) Enable(ctx context.Context, scope reasoningmode.Scope, now time.
return reasoningmode.State{}, err
}
state.Mode = reasoningmode.ModeReasoning
state.UserSelected = true
state.RoundsUsed = 0
state.LastActivityAt = now
state.ExpiresAt = now.Add(idle)
@@ -92,6 +93,7 @@ func (s *Store) Disable(ctx context.Context, scope reasoningmode.Scope, now time
state, _ = reasoningmode.EffectiveState(state, now)
wasReasoning := state.Mode == reasoningmode.ModeReasoning
state.Mode = reasoningmode.ModeQuick
state.UserSelected = true
state.RoundsUsed = 0
state.ExpiresAt = time.Time{}
if err = updateReasoningState(ctx, tx, scope, state, now); err != nil {
@@ -256,14 +258,14 @@ func ensureReasoningState(ctx context.Context, tx pgx.Tx, scope reasoningmode.Sc
}
func selectReasoningStateForUpdate(ctx context.Context, tx pgx.Tx, scope reasoningmode.Scope) (reasoningmode.State, error) {
return scanReasoningState(tx.QueryRow(ctx, `SELECT mode,max_rounds,rounds_used,hint_enabled,last_activity_at,expires_at,onboarding_shown_at
return scanReasoningState(tx.QueryRow(ctx, `SELECT mode,user_selected,max_rounds,rounds_used,hint_enabled,last_activity_at,expires_at,onboarding_shown_at
FROM reasoning_mode_states
WHERE tenant_key=$1 AND channel=$2 AND external_actor_id=$3
FOR UPDATE`, scope.TenantKey, scope.Channel, scope.ExternalActorID))
}
func selectReasoningState(ctx context.Context, tx pgx.Tx, scope reasoningmode.Scope) (reasoningmode.State, error) {
return scanReasoningState(tx.QueryRow(ctx, `SELECT mode,max_rounds,rounds_used,hint_enabled,last_activity_at,expires_at,onboarding_shown_at
return scanReasoningState(tx.QueryRow(ctx, `SELECT mode,user_selected,max_rounds,rounds_used,hint_enabled,last_activity_at,expires_at,onboarding_shown_at
FROM reasoning_mode_states
WHERE tenant_key=$1 AND channel=$2 AND external_actor_id=$3`, scope.TenantKey, scope.Channel, scope.ExternalActorID))
}
@@ -276,7 +278,7 @@ func scanReasoningState(row pgx.Row) (reasoningmode.State, error) {
expires pgtype.Timestamptz
onboarding pgtype.Timestamptz
)
if err := row.Scan(&mode, &state.MaxRounds, &state.RoundsUsed, &state.HintEnabled, &last, &expires, &onboarding); err != nil {
if err := row.Scan(&mode, &state.UserSelected, &state.MaxRounds, &state.RoundsUsed, &state.HintEnabled, &last, &expires, &onboarding); err != nil {
return reasoningmode.State{}, fmt.Errorf("scan reasoning state: %w", err)
}
state.Mode = reasoningmode.Mode(mode)
@@ -295,11 +297,11 @@ func scanReasoningState(row pgx.Row) (reasoningmode.State, error) {
func updateReasoningState(ctx context.Context, tx pgx.Tx, scope reasoningmode.Scope, state reasoningmode.State, now time.Time) error {
_, err := tx.Exec(ctx, `UPDATE reasoning_mode_states SET
mode=$4,max_rounds=$5,rounds_used=$6,hint_enabled=$7,
last_activity_at=$8,expires_at=$9,onboarding_shown_at=$10,updated_at=$11
mode=$4,user_selected=$5,max_rounds=$6,rounds_used=$7,hint_enabled=$8,
last_activity_at=$9,expires_at=$10,onboarding_shown_at=$11,updated_at=$12
WHERE tenant_key=$1 AND channel=$2 AND external_actor_id=$3`,
scope.TenantKey, scope.Channel, scope.ExternalActorID,
string(state.Mode), state.MaxRounds, state.RoundsUsed, state.HintEnabled,
string(state.Mode), state.UserSelected, state.MaxRounds, state.RoundsUsed, state.HintEnabled,
nullableTime(state.LastActivityAt), nullableTime(state.ExpiresAt), nullableTime(state.OnboardingShownAt), now,
)
if err != nil {
+53 -3
View File
@@ -34,6 +34,12 @@ type StreamEvent struct {
FullAnswer string
ReferenceCount int
ToolName string
// AnswerRunID / AnswerRunStart describe the answer run the event belongs
// to. WeKnora tags every agent step's answer chunks with its own
// data.event_id, so a change of AnswerRunID marks the start of a new run
// and AnswerRunStart is the accumulated answer offset where it began.
AnswerRunID string
AnswerRunStart int
}
// StreamObserver is called with a safe StreamEvent for each SSE event.
@@ -54,6 +60,12 @@ type AgentChatRequest struct {
DisableTitle bool `json:"disable_title,omitempty"`
KnowledgeBaseIDs []string `json:"knowledge_base_ids,omitempty"`
Images []Image `json:"images,omitempty"`
// Locale is the response language preference ("zh-CN", "en", "ja", "ko",
// "es", or "auto"). It is transmitted as the Accept-Language request
// header — WeKnora's Language middleware turns it into the {{language}}
// prompt placeholder. "auto" and "" mean: detect from the query text.
// Not serialized in the JSON body.
Locale string `json:"-"`
}
type Image struct {
@@ -148,6 +160,9 @@ func (c *Client) agentChat(ctx context.Context, sessionID string, request AgentC
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "text/event-stream")
req.Header.Set("X-API-Key", c.apiKey)
if locale := resolveLocale(request.Locale, request.Query); locale != "" {
req.Header.Set("Accept-Language", locale)
}
resp, err := c.http.Do(req)
if err != nil {
return ChatResult{}, fmt.Errorf("call weknora agent chat: %w", err)
@@ -218,6 +233,12 @@ func parseSSEWithEventObserver(reader io.Reader, answerObserver AnswerObserver,
var result ChatResult
var dataLines []string
terminal := false
// A reasoning agent publishes every step narrative and the final answer
// body through the same answer channel, but each step is its own answer
// run (data.event_id). A change of run id marks a new run start, which is
// what lets callers separate narration from the final body instead of
// guessing from the text.
answerRunID, answerRunStart := "", 0
notify := func(e StreamEvent) {
if eventObserver != nil {
@@ -253,15 +274,20 @@ func parseSSEWithEventObserver(reader io.Reader, answerObserver AnswerObserver,
ReferenceCount: len(event.KnowledgeReferences),
})
case "answer":
if runID := extractAnswerRunID(event.Data); runID != "" && runID != answerRunID {
answerRunID, answerRunStart = runID, len(result.Answer)
}
hasContent := strings.TrimSpace(event.Content) != ""
result.Answer += event.Content
if hasContent && answerObserver != nil {
answerObserver(result.Answer)
}
notify(StreamEvent{
ResponseType: "answer",
Done: event.Done,
FullAnswer: result.Answer,
ResponseType: "answer",
Done: event.Done,
FullAnswer: result.Answer,
AnswerRunID: answerRunID,
AnswerRunStart: answerRunStart,
})
if event.Done {
terminal = true
@@ -339,6 +365,30 @@ func parseSSEWithEventObserver(reader io.Reader, answerObserver AnswerObserver,
return result, nil
}
// extractAnswerRunID extracts data.event_id, the identifier of the answer run
// an answer chunk belongs to. A reasoning agent gives every step its own id,
// so a change of id marks the start of a new run and, once no tool event
// follows, of the final answer body. Older deployments that do not tag answer
// events yield an empty id, which callers treat as "no observation".
func extractAnswerRunID(data json.RawMessage) string {
if len(data) == 0 {
return ""
}
var m map[string]json.RawMessage
if err := json.Unmarshal(data, &m); err != nil {
return ""
}
v, ok := m["event_id"]
if !ok {
return ""
}
var s string
if err := json.Unmarshal(v, &s); err != nil {
return ""
}
return strings.TrimSpace(s)
}
// extractToolName extracts the tool_name string from a JSON Data field.
// The Data may be any JSON shape; we look for the "tool_name" key and return
// its string value. Returns empty string if not found or not a string.
+67
View File
@@ -0,0 +1,67 @@
package weknora
import (
"strings"
"unicode"
)
// fixedLocales are the response languages the platform understands end-to-end
// (settings command, preference storage, and WeKnora's LanguageLocaleName map).
var fixedLocales = map[string]bool{
"zh-CN": true, "en": true, "ja": true, "ko": true, "es": true,
}
// resolveLocale maps a user preference + query text to an Accept-Language
// value understood by WeKnora's Language middleware.
//
// - "" -> "" (send no header; WeKnora keeps its own fallback)
// - "auto" -> best-effort detection from the query script
// - a known fixed locale (e.g. "en", "ja") -> passed through verbatim
//
// WeKnora matches the first Accept-Language tag and maps it to a human
// language name for its prompts, so the returned tag must be one it knows
// ("zh-CN", "en", "ja", "ko", "es", ...). An "auto" query whose script
// cannot be classified returns "" so the server behaves exactly as before.
func resolveLocale(locale, query string) string {
locale = strings.TrimSpace(locale)
if fixedLocales[locale] {
return locale
}
if locale == "auto" {
return detectLocale(query)
}
return ""
}
// detectLocale classifies a query by the dominant non-ASCII script. Latin
// text maps to "en"; CJK ideographs to "zh-CN"; kana to "ja"; hangul to "ko";
// Spanish is not reliably script-detectable and is never auto-selected.
// Pure punctuation/digits/empty text return "" (no guess).
func detectLocale(query string) string {
var cjk, kana, hangul, latin int
for _, r := range query {
switch {
case unicode.Is(unicode.Hiragana, r) || unicode.Is(unicode.Katakana, r):
kana++
case unicode.Is(unicode.Han, r):
cjk++
case unicode.Is(unicode.Hangul, r):
hangul++
case r >= 'A' && r <= 'Z' || r >= 'a' && r <= 'z':
latin++
}
}
// Japanese mixes kanji in, so any kana presence wins over Han count.
switch {
case kana > 0:
return "ja"
case hangul > 0:
return "ko"
case cjk > 0:
return "zh-CN"
case latin > 0:
return "en"
default:
return ""
}
}
+96
View File
@@ -0,0 +1,96 @@
package weknora
import (
"context"
"net/http"
"net/http/httptest"
"testing"
)
func TestResolveLocaleFixedPrefersPreference(t *testing.T) {
// A fixed preference must win over the query's own script.
for _, locale := range []string{"zh-CN", "en", "ja", "ko", "es"} {
if got := resolveLocale(locale, "how many CR series do we have?"); got != locale {
t.Fatalf("resolveLocale(%q) = %q, want %q", locale, got, locale)
}
}
}
func TestResolveLocaleAutoDetectsFromQuery(t *testing.T) {
cases := []struct{ query, want string }{
{"how many serise do we have for CR?", "en"},
{"cr一共有几个系列", "zh-CN"},
{"ロボットのアームリセット方法", "ja"},
{"机器人复位怎么操作", "zh-CN"},
{"로봇 암 리셋 방법", "ko"},
{"5013", ""}, // no script evidence: no guess
{"!!! ??", ""}, // punctuation only: no guess
}
for _, tc := range cases {
if got := resolveLocale("auto", tc.query); got != tc.want {
t.Fatalf("resolveLocale(auto, %q) = %q, want %q", tc.query, got, tc.want)
}
}
}
func TestResolveLocaleUnknownPreferenceSendsNothing(t *testing.T) {
// Empty and unrecognized locales must not fabricate a header; WeKnora
// keeps its own fallback behaviour, exactly like before this feature.
if got := resolveLocale("", "hello"); got != "" {
t.Fatalf("empty locale = %q, want none", got)
}
if got := resolveLocale("fr-CA", "hello"); got != "" {
t.Fatalf("unknown locale = %q, want none", got)
}
}
func TestAgentChatSendsAcceptLanguageHeader(t *testing.T) {
var gotHeader string
body := "data: {\"response_type\":\"answer\",\"content\":\"ok\"}\n\ndata: {\"response_type\":\"complete\",\"done\":true}\n\n"
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
gotHeader = r.Header.Get("Accept-Language")
w.Header().Set("Content-Type", "text/event-stream")
_, _ = w.Write([]byte(body))
}))
defer server.Close()
client, err := NewClient(server.URL, "test-key", 0)
if err != nil {
t.Fatal(err)
}
if _, err := client.AgentChat(context.Background(), "s1", AgentChatRequest{
Query: "how many serise do we have for CR?", AgentID: "a1", Locale: "en",
}); err != nil {
t.Fatal(err)
}
if gotHeader != "en" {
t.Fatalf("Accept-Language = %q, want %q", gotHeader, "en")
}
}
func TestAgentChatOmitsAcceptLanguageWhenNoPreference(t *testing.T) {
var sent bool
body := "data: {\"response_type\":\"answer\",\"content\":\"ok\"}\n\ndata: {\"response_type\":\"complete\",\"done\":true}\n\n"
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_, sent = r.Header["Accept-Language"]
w.Header().Set("Content-Type", "text/event-stream")
_, _ = w.Write([]byte(body))
}))
defer server.Close()
client, _ := NewClient(server.URL, "test-key", 0)
if _, err := client.AgentChat(context.Background(), "s1", AgentChatRequest{
Query: "cr有哪些系列", AgentID: "a1",
}); err != nil {
t.Fatal(err)
}
if sent {
t.Fatal("Accept-Language header must be absent without a locale")
}
}
func TestAgentChatRequestLocaleIsNotSerialized(t *testing.T) {
// Locale rides on the header only; the JSON body contract is unchanged.
// (Guards against a future refactor accidentally adding it to the payload.)
if got := resolveLocale("auto", "How to reset the robot arm?"); got != "en" {
t.Fatalf("auto detection broken: %q", got)
}
}
+66
View File
@@ -267,3 +267,69 @@ func TestAgentChatEventsErrorTurn(t *testing.T) {
t.Fatalf("expected error done event, got %+v", events[0])
}
}
// TestStreamEventAnswerRuns verifies answer chunks are attributed to the answer
// run that produced them: a reasoning agent publishes every step narrative and
// the final answer body through the same answer channel, each under its own
// data.event_id, so the run start marks where the final body begins.
func TestStreamEventAnswerRuns(t *testing.T) {
stream := "data: {\"response_type\":\"thinking\",\"content\":\"inner monologue\",\"done\":false}\n\n" +
"data: {\"response_type\":\"answer\",\"content\":\"我先检索知识库。\",\"data\":{\"event_id\":\"step-1-answer\"},\"done\":false}\n\n" +
"data: {\"response_type\":\"tool_call\",\"data\":{\"tool_name\":\"knowledge_search\",\"event_id\":\"step-1-tool\"},\"done\":false}\n\n" +
"data: {\"response_type\":\"tool_result\",\"data\":{\"tool_name\":\"knowledge_search\",\"event_id\":\"step-1-tool\"},\"done\":false}\n\n" +
"data: {\"response_type\":\"answer\",\"content\":\"我已经找到了负载曲线图:\",\"data\":{\"event_id\":\"final-answer\"},\"done\":false}\n\n" +
"data: {\"response_type\":\"answer\",\"content\":\"![图](resource://abc)\",\"data\":{\"event_id\":\"final-answer\"},\"done\":true}\n\n"
var events []StreamEvent
result, err := parseSSEWithEventObserver(strings.NewReader(stream), nil, func(e StreamEvent) {
events = append(events, e)
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
if got := len([]rune("我先检索知识库。")); got == 0 {
t.Fatal("unreachable")
}
narration := len("我先检索知识库。")
if result.Answer != "我先检索知识库。我已经找到了负载曲线图:![图](resource://abc)" {
t.Fatalf("unexpected answer: %q", result.Answer)
}
var answers []StreamEvent
for _, e := range events {
if e.ResponseType == "answer" {
answers = append(answers, e)
}
}
if len(answers) != 3 {
t.Fatalf("expected 3 answer events, got %d: %+v", len(answers), answers)
}
if answers[0].AnswerRunID != "step-1-answer" || answers[0].AnswerRunStart != 0 {
t.Fatalf("narration run: want id 'step-1-answer' start 0, got %+v", answers[0])
}
if answers[1].AnswerRunID != "final-answer" || answers[1].AnswerRunStart != narration {
t.Fatalf("final run must start at %d, got %+v", narration, answers[1])
}
if answers[2].AnswerRunID != "final-answer" || answers[2].AnswerRunStart != narration {
t.Fatalf("chunks of one run must share its start, got %+v", answers[2])
}
}
// TestStreamEventAnswerRunWithoutID verifies streams that carry no run id leave
// the run fields empty, so consumers fall back to the text heuristic.
func TestStreamEventAnswerRunWithoutID(t *testing.T) {
stream := "data: {\"response_type\":\"answer\",\"content\":\"hello \",\"done\":false}\n\n" +
"data: {\"response_type\":\"answer\",\"content\":\"world\",\"done\":true}\n\n"
var events []StreamEvent
if _, err := parseSSEWithEventObserver(strings.NewReader(stream), nil, func(e StreamEvent) {
events = append(events, e)
}); err != nil {
t.Fatalf("unexpected error: %v", err)
}
for i, e := range events {
if e.AnswerRunID != "" || e.AnswerRunStart != 0 {
t.Fatalf("event[%d] must carry no run info, got %+v", i, e)
}
}
}
+68 -84
View File
@@ -6,112 +6,96 @@ import (
"testing"
)
func TestDobotAIDingTalkContainerAssetsProvideSafeShortOperatorCommands(t *testing.T) {
func TestDockerDeploymentContainsOnlyDobotAIApplication(t *testing.T) {
compose, err := os.ReadFile("../deploy/dobotai-dingtalk/compose.yaml")
if err != nil {
t.Fatal(err)
}
for _, required := range []string{
"dobotai-dingtalk:",
"container_name: dobotai-dingtalk",
"network_mode: host",
"env_file:",
"read_only: true",
"/var/lib/dobotai/secrets",
} {
if !strings.Contains(string(compose), required) {
source := string(compose)
for _, required := range []string{"dobotai-dingtalk:", "container_name: dobotai-dingtalk", "network_mode: host", "env_file:", "read_only: true"} {
if !strings.Contains(source, required) {
t.Errorf("compose missing %q", required)
}
}
configure, err := os.ReadFile("../deploy/dobotai-dingtalk/dobotai-dingtalk-configure")
if err != nil {
t.Fatal(err)
}
for _, required := range []string{
"configure_env.py",
"/etc/dobotai-test/dobotai.env",
"dobotai-dingtalk-start",
} {
if !strings.Contains(string(configure), required) {
t.Errorf("configure command missing %q", required)
}
}
bind, err := os.ReadFile("../deploy/dobotai-dingtalk/dobotai-dingtalk-bind-admin")
if err != nil {
t.Fatal(err)
}
for _, required := range []string{
"docker exec -it dobotai-dingtalk",
"/usr/local/bin/dobotai bootstrap-admin",
} {
if !strings.Contains(string(bind), required) {
t.Errorf("bind command missing %q", required)
for _, forbidden := range []string{"dobotai-postgres", "postgres-data", "name: dobotai-network"} {
if strings.Contains(source, forbidden) {
t.Errorf("compose must not manage a local database resource %q", forbidden)
}
}
}
func TestDobotAIDingTalkCredentialConfigurationNeverPassesSecretsOnCommandLine(t *testing.T) {
func TestRootInstallerBuildsImageWithoutOperatorImageArgument(t *testing.T) {
data, err := os.ReadFile("../install-docker.sh")
if err != nil {
t.Fatal(err)
}
source := string(data)
for _, required := range []string{"docker build", "dobotai-dingtalk:local-", "install_docker_if_missing", "DOBOTAI_SOURCE_ROOT", "deploy/dobotai-dingtalk/install-docker.sh"} {
if !strings.Contains(source, required) {
t.Errorf("root installer missing %q", required)
}
}
if strings.Contains(source, "${DOBOTAI_IMAGE") {
t.Fatal("root installer must not require DOBOTAI_IMAGE")
}
}
func TestDockerfilePropagatesConfiguredGoProxyToModuleDownload(t *testing.T) {
data, err := os.ReadFile("../Dockerfile")
if err != nil {
t.Fatal(err)
}
source := string(data)
for _, required := range []string{"ARG GOPROXY=", "ENV GOPROXY=${GOPROXY}", "RUN go mod download"} {
if !strings.Contains(source, required) {
t.Errorf("Dockerfile missing %q", required)
}
}
if !(strings.Index(source, "ENV GOPROXY=${GOPROXY}") < strings.Index(source, "RUN go mod download")) {
t.Fatal("Dockerfile must set GOPROXY before downloading modules")
}
}
func TestRDSInitializerOnlyInitializesEmptySchemaAndRejectsInconsistentSchema(t *testing.T) {
data, err := os.ReadFile("../deploy/dobotai-dingtalk/rds-schema-init.sh")
if err != nil {
t.Fatal(err)
}
source := string(data)
for _, required := range []string{"pg_isready", "to_regclass('public.turns')", "empty", "normal", "inconsistent", "数据库不存在", "*.up.sql", "DATABASE_URL", "postgres:16-alpine"} {
if !strings.Contains(source, required) {
t.Errorf("RDS initializer missing %q", required)
}
}
for _, forbidden := range []string{"docker compose", "dobotai-postgres", "pg_dump -Fc", "DROP DATABASE", "DROP SCHEMA"} {
if strings.Contains(source, forbidden) {
t.Errorf("RDS initializer must not manage or destroy databases: %q", forbidden)
}
}
}
func TestRDSCredentialConfiguratorPromptsForExternalDatabase(t *testing.T) {
data, err := os.ReadFile("../deploy/dobotai-dingtalk/configure_env.py")
if err != nil {
t.Fatal(err)
}
source := string(data)
for _, required := range []string{"getpass.getpass", "os.replace", "0o600", "DINGTALK_CLIENT_SECRET"} {
for _, required := range []string{"DATABASE_URL", "阿里云RDS", "RDS主机地址", "RDS端口", "数据库名", "数据库账号", "数据库密码", "快速问答Agent ID", "智能推理Agent ID", "getpass.getpass", "os.replace", "0o600", "OSS整组录入", "DOBOTAI_OUTBOUND_IMAGE_OSS_ACCESS_KEY_SECRET"} {
if !strings.Contains(source, required) {
t.Errorf("credential configurator missing %q", required)
t.Errorf("RDS configurator missing %q", required)
}
}
for _, forbidden := range []string{"subprocess.run([", "-v", "DINGTALK_CLIENT_SECRET="} {
if strings.Contains(source, forbidden) {
t.Errorf("credential configurator must not expose secret through process arguments: %q", forbidden)
}
if strings.Contains(source, "RDS PostgreSQL连接串") {
t.Fatal("operator must not be asked to type a whole DATABASE_URL string")
}
}
func TestDobotAIDingTalkCredentialConfigurationExplainsEveryInputWithRedactedExamples(t *testing.T) {
configurator, err := os.ReadFile("../deploy/dobotai-dingtalk/configure_env.py")
if strings.Contains(source, "DobotAI租户键") || strings.Contains(source, "Agent Bindings JSON") || strings.Contains(source, "DobotAI Ingress Token") {
t.Fatal("operator must only be prompted for real RDS, WeKnora and DingTalk configuration")
}
installer, err := os.ReadFile("../deploy/dobotai-dingtalk/install-docker.sh")
if err != nil {
t.Fatal(err)
}
documentation, err := os.ReadFile("../deploy/dobotai-dingtalk/README.md")
if err != nil {
t.Fatal(err)
}
if !strings.Contains(string(configurator), "格式示例:{example}") || !strings.Contains(string(documentation), "格式示例") {
t.Fatal("configuration guidance must label every example")
}
for _, example := range []string{
"postgres://USER:<数据库密码>@127.0.0.1:55432/dobotai_test?sslmode=disable",
"http://127.0.0.1:8080",
`{"dingtalk:*":"你的Agent ID"}`,
"DINGTALK_CLIENT_SECRET",
} {
if !strings.Contains(string(configurator), example) {
t.Errorf("configurator missing redacted example %q", example)
}
if !strings.Contains(string(documentation), example) {
t.Errorf("documentation missing redacted example %q", example)
}
}
}
func TestDobotAIDingTalkStartRollsBackToPreviousContainerImage(t *testing.T) {
data, err := os.ReadFile("../deploy/dobotai-dingtalk/dobotai-dingtalk-start")
if err != nil {
t.Fatal(err)
}
source := string(data)
for _, required := range []string{
"previous_image=",
"set_runtime_image",
"docker inspect --format",
"set_runtime_image \"$previous_image\"",
"docker compose --env-file \"$runtime_env\" -f \"$compose_file\" up -d --no-build --force-recreate",
} {
if !strings.Contains(source, required) {
t.Errorf("start script missing container upgrade rollback %q", required)
}
if !strings.Contains(string(installer), "secrets.token_hex") || !strings.Contains(string(installer), "secrets.token_urlsafe") || !strings.Contains(string(installer), "DOBOTAI_TENANT_KEY=dobotai-") {
t.Fatal("installer must generate and persist internal tenant and ingress tokens")
}
}