1
This commit is contained in:
@@ -43,6 +43,7 @@
|
||||
- **新增并行点的固定三处**:`common/pool.go` 加池变量(grpool 封装)→ `biz/consts` 加默认大小 → `config.yml` 的 `pool` 段加 `key: 并发度`(缺失或非法时回退默认值)
|
||||
- 禁止直接用裸 `go` 启动并行工作负载,一律走 `common` 的池(池清单与默认值见 README 配置说明)
|
||||
- **防死锁**:等待链单向「主 → A池 → B池」,被等待池的任务内不得再等待任何池(会饿死 worker);池无 Wait 方法,等待用调用方 `sync.WaitGroup`,任务结果经 buffered channel 回主 goroutine
|
||||
- **共享状态安全(内存)**:池内任务并发执行,共享实例(service 单例、model 句柄等)只允许**只读**访问;可变字段必须在**提交池之前**由主 goroutine 一次性预置,任务内禁止写共享字段——Go map 并发写直接 `fatal error: concurrent map writes`,无锁、无降级、不可恢复,只能崩溃重启。需要可变共享状态时按优先级:① 无共享(任务内新建、buffered channel 传递结果)② 锁(`sync.Mutex`/`RWMutex`,锁内只做内存操作,LLM/DB 等 IO 放锁外)③ `sync/atomic`(仅限 int 类标量计数/标志,如 `atomic.AddInt64`,并发计数禁用普通 `++`;复合结构不要用 atomic,指针 CAS 属例外)
|
||||
- **裸 `go` 允许的例外**:`go func(){ wg.Wait(); close(ch) }()` 收尾惯用法、SSE 心跳、流式管道(Stream 读写)等长生命周期/非工作负载协程
|
||||
|
||||
## 文档职责（三文档体系）
|
||||
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
+4
-1
@@ -26,7 +26,10 @@ const (
|
||||
MaxReactRounds = 10 // ReAct 轮次上限
|
||||
ToolResultMaxChars = 1500 // search 工具返回的单条 chunk 截断字数（控制上下文体积）
|
||||
|
||||
ParsePollIntervalSeconds = 5 // 解析/标注任务轮询间隔
|
||||
ParsePollIntervalSeconds = 15 // 解析/标注任务轮询间隔
|
||||
|
||||
DefaultLlmHttpTimeout = 600 // LLM HTTP 客户端超时（秒）：本地模型生成慢，且多请求排队时响应可超分钟级
|
||||
MaxExtractTokens = 4096 // 知识抽取单次生成的 token 预算（防失控生成烧掉上下文）
|
||||
|
||||
EmbedBatchSize = 16 // 单次向量化请求的文本批量
|
||||
|
||||
|
||||
@@ -33,8 +33,6 @@ import (
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
)
|
||||
|
||||
var httpClient = &http.Client{Timeout: 2 * time.Minute}
|
||||
|
||||
// ---------- OpenAI 兼容 HTTP 组件 ----------
|
||||
|
||||
type openAIMessage struct {
|
||||
@@ -76,10 +74,17 @@ type openAIStreamChunk struct {
|
||||
type OpenAIChatModel struct {
|
||||
cfg *entity.ModelConfig
|
||||
tools []*schema.ToolInfo
|
||||
extra map[string]any // 附加请求字段（如 gemma-4 关闭思维链）
|
||||
}
|
||||
|
||||
func NewOpenAIChatModel(cfg *entity.ModelConfig) *OpenAIChatModel {
|
||||
return &OpenAIChatModel{cfg: cfg}
|
||||
return &OpenAIChatModel{cfg: cfg, extra: map[string]any{}}
|
||||
}
|
||||
|
||||
// DisableThinking 关闭模型思维链：本地 gemma-4-E4B 偶发长思考会烧光上下文，导致返回空内容
|
||||
func (m *OpenAIChatModel) DisableThinking() *OpenAIChatModel {
|
||||
m.extra["thinking"] = false
|
||||
return m
|
||||
}
|
||||
|
||||
func (m *OpenAIChatModel) Generate(ctx context.Context, input []*schema.Message, opts ...emodel.Option) (*schema.Message, error) {
|
||||
@@ -88,6 +93,12 @@ func (m *OpenAIChatModel) Generate(ctx context.Context, input []*schema.Message,
|
||||
"messages": buildOpenAIMessages(input),
|
||||
"stream": false,
|
||||
}
|
||||
for k, v := range m.extra {
|
||||
payload[k] = v
|
||||
}
|
||||
if commonOpts := emodel.GetCommonOptions(nil, opts...); commonOpts.MaxTokens != nil {
|
||||
payload["max_tokens"] = *commonOpts.MaxTokens
|
||||
}
|
||||
m.withTools(payload)
|
||||
body, err := postOpenAI(ctx, m.cfg, m.endpoint("/chat/completions"), payload)
|
||||
if err != nil {
|
||||
@@ -1174,7 +1185,9 @@ func doOpenAIRequest(ctx context.Context, cfg *entity.ModelConfig, url string, p
|
||||
if cfg.ApiKey != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+cfg.ApiKey)
|
||||
}
|
||||
resp, err := httpClient.Do(req)
|
||||
// 本地模型生成慢且多请求排队时响应可超分钟级，超时取 config.yml chat.timeout（秒）
|
||||
timeout := g.Cfg().MustGet(ctx, "chat.timeout", consts.DefaultLlmHttpTimeout).Int()
|
||||
resp, err := (&http.Client{Timeout: time.Duration(timeout) * time.Second}).Do(req)
|
||||
if err != nil {
|
||||
g.Log().Errorf(ctx, "model call failed: model=%s url=%s err=%v", cfg.ModelName, url, err)
|
||||
return nil, err
|
||||
|
||||
@@ -3,6 +3,7 @@ package service
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
@@ -11,9 +12,9 @@ import (
|
||||
"rag-local/kb/dao"
|
||||
"rag-local/kb/model/entity"
|
||||
|
||||
emodel "github.com/cloudwego/eino/components/model"
|
||||
"github.com/cloudwego/eino/schema"
|
||||
"github.com/gogf/gf/v2/errors/gerror"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
)
|
||||
|
||||
var KgEntityService = &kgEntityService{}
|
||||
@@ -48,6 +49,8 @@ func (s *kgEntityService) ExtractDocument(ctx context.Context, datasetId, docume
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
// 池内并发共享 model 实例，只读字段须在此（单 goroutine）预置好
|
||||
model.DisableThinking()
|
||||
|
||||
type extractOut struct {
|
||||
chunkId int64
|
||||
@@ -69,7 +72,7 @@ func (s *kgEntityService) ExtractDocument(ctx context.Context, datasetId, docume
|
||||
ch <- extractOut{chunkId: c.Id, err: e}
|
||||
}); err != nil {
|
||||
wg.Done()
|
||||
g.Log().Warningf(ctx, "kg extract chunk %d submit failed: %v", c.Id, err)
|
||||
fmt.Printf("kg extract chunk %d submit failed: %v\n", c.Id, err)
|
||||
}
|
||||
}
|
||||
go func() { wg.Wait(); close(ch) }()
|
||||
@@ -78,12 +81,12 @@ func (s *kgEntityService) ExtractDocument(ctx context.Context, datasetId, docume
|
||||
for out := range ch {
|
||||
if out.err != nil {
|
||||
failed++
|
||||
g.Log().Warningf(ctx, "kg extract chunk %d failed: %v", out.chunkId, out.err)
|
||||
fmt.Printf("kg extract chunk %d failed: %v\n", out.chunkId, out.err)
|
||||
continue
|
||||
}
|
||||
if err := s.saveExtract(ctx, datasetId, out.chunkId, out.entities, out.relations); err != nil {
|
||||
failed++
|
||||
g.Log().Warningf(ctx, "kg extract chunk %d failed: %v", out.chunkId, err)
|
||||
fmt.Printf("kg extract chunk %d failed: %v\n", out.chunkId, err)
|
||||
}
|
||||
}
|
||||
return failed, nil
|
||||
@@ -106,13 +109,22 @@ func (s *kgEntityService) callExtract(ctx context.Context, model *OpenAIChatMode
|
||||
{Role: schema.System, Content: "你是知识抽取助手。从文档片段中抽取实体（人名、组织、地名、产品等专有名词）及实体间的关系（动词或介词短语）。只输出 JSON，不要 markdown 代码块或任何解释，格式：{\"entities\":[{\"name\":\"实体名\",\"type\":\"类型\"}],\"relations\":[{\"head\":\"主体\",\"relation\":\"关系\",\"tail\":\"客体\"}]}"},
|
||||
{Role: schema.User, Content: "文档片段：\n" + chunk.Content},
|
||||
}
|
||||
resp, err := model.Generate(ctx, msgs)
|
||||
resp, err := model.Generate(ctx, msgs, emodel.WithMaxTokens(consts.MaxExtractTokens))
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
data, err := parseKgJSON(resp.Content)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
// 模型偶发返回空内容/截断 JSON，重试一次通常正常
|
||||
fmt.Printf("kg extract chunk %d parse failed (%v), retrying\n", chunk.Id, err)
|
||||
resp, err = model.Generate(ctx, msgs, emodel.WithMaxTokens(consts.MaxExtractTokens))
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
data, err = parseKgJSON(resp.Content)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
}
|
||||
entities := make([]kgEntityItem, 0, len(data.Entities))
|
||||
for _, e := range data.Entities {
|
||||
@@ -154,7 +166,7 @@ func (s *kgEntityService) saveExtract(ctx context.Context, datasetId, chunkId in
|
||||
return nil
|
||||
}
|
||||
|
||||
// parseKgJSON 解析模型输出的 JSON，容忍 ```json 代码块包裹
|
||||
// parseKgJSON 解析模型输出的 JSON，容忍 ```json 代码块包裹及首个 JSON 对象后的多余内容（模型偶尔续写 ",{...}"）
|
||||
func parseKgJSON(content string) (*kgExtractResult, error) {
|
||||
content = strings.TrimSpace(content)
|
||||
content = strings.TrimPrefix(content, "```json")
|
||||
@@ -163,11 +175,52 @@ func parseKgJSON(content string) (*kgExtractResult, error) {
|
||||
content = strings.TrimSpace(content)
|
||||
var out kgExtractResult
|
||||
if err := json.Unmarshal([]byte(content), &out); err != nil {
|
||||
return nil, gerror.Wrap(err, "解析抽取 JSON 失败")
|
||||
if fixed, ok := extractFirstJSON(content); ok {
|
||||
if err2 := json.Unmarshal([]byte(fixed), &out); err2 == nil {
|
||||
return &out, nil
|
||||
}
|
||||
}
|
||||
return nil, gerror.Wrapf(err, "解析抽取 JSON 失败，模型输出：%q", truncateRunes(content, 200))
|
||||
}
|
||||
return &out, nil
|
||||
}
|
||||
|
||||
// extractFirstJSON 提取字符串中第一个完整 JSON 对象（平衡花括号，跳过字符串内的花括号）
|
||||
func extractFirstJSON(s string) (string, bool) {
|
||||
start := strings.IndexByte(s, '{')
|
||||
if start < 0 {
|
||||
return "", false
|
||||
}
|
||||
depth := 0
|
||||
inStr := false
|
||||
escaped := false
|
||||
for i := start; i < len(s); i++ {
|
||||
ch := s[i]
|
||||
if inStr {
|
||||
if escaped {
|
||||
escaped = false
|
||||
} else if ch == '\\' {
|
||||
escaped = true
|
||||
} else if ch == '"' {
|
||||
inStr = false
|
||||
}
|
||||
continue
|
||||
}
|
||||
switch ch {
|
||||
case '"':
|
||||
inStr = true
|
||||
case '{':
|
||||
depth++
|
||||
case '}':
|
||||
depth--
|
||||
if depth == 0 {
|
||||
return s[start : i+1], true
|
||||
}
|
||||
}
|
||||
}
|
||||
return "", false
|
||||
}
|
||||
|
||||
func (s *kgEntityService) List(ctx context.Context, datasetId int64, page, pageSize int) ([]*entity.KgEntity, int, error) {
|
||||
return dao.KgEntity.List(ctx, datasetId, page, pageSize)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user