feat(server): 分块并发调用模型,配置可从仓库根目录的 .env 读取
- extract:分块并发调用(每个任务最多 4 路),结果按分块顺序合并 - config:在 server/ 下运行时也读取仓库根目录的 .env(和 Compose 的 env_file 一致) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -6,6 +6,7 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import asyncio
|
||||||
import re
|
import re
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
from typing import Any
|
from typing import Any
|
||||||
@@ -17,6 +18,7 @@ from bid_doctor.document.pipeline.text_layer import Sentence, norm
|
|||||||
|
|
||||||
PROMPT_ID, PROMPT_VERSION = "extract.substantive", 1
|
PROMPT_ID, PROMPT_VERSION = "extract.substantive", 1
|
||||||
CHUNK_CHARS = 6000
|
CHUNK_CHARS = 6000
|
||||||
|
CONCURRENCY = 4 # 每个解析任务同时调用模型的分块数;TODO(路由表):按「供应商 × 租户」统一限流
|
||||||
INVALID = re.compile(r"投标无效|响应无效|无效投标|作无效处理|按无效|否决其投标|否决投标|不予受理")
|
INVALID = re.compile(r"投标无效|响应无效|无效投标|作无效处理|按无效|否决其投标|否决投标|不予受理")
|
||||||
RULE_NOTE = re.compile(r"(★|*|\*|斜体|下划线).{0,30}(实质性|无效|否决)|(实质性).{0,20}(★|*|\*|斜体|下划线)")
|
RULE_NOTE = re.compile(r"(★|*|\*|斜体|下划线).{0,30}(实质性|无效|否决)|(实质性).{0,20}(★|*|\*|斜体|下划线)")
|
||||||
DECLARED = re.compile(r"★[^。;]{0,12}?共\s*(\d+)\s*[条项处]")
|
DECLARED = re.compile(r"★[^。;]{0,12}?共\s*(\d+)\s*[条项处]")
|
||||||
@@ -125,20 +127,28 @@ async def run(ctx: TenantContext, job_id: UUID, sentences: list[Sentence]) -> tu
|
|||||||
note_keys = {n.sentence.key for n in notes}
|
note_keys = {n.sentence.key for n in notes}
|
||||||
declared = {n.marker for n in notes}
|
declared = {n.marker for n in notes}
|
||||||
out: list[Extracted] = []
|
out: list[Extracted] = []
|
||||||
for chunk in _chunks(sentences):
|
chunks = _chunks(sentences)
|
||||||
by_id = {_sid(s): s for s in chunk}
|
gate = asyncio.Semaphore(CONCURRENCY)
|
||||||
|
|
||||||
|
async def call(chunk: list[Sentence]) -> dict[str, Any]:
|
||||||
user = (
|
user = (
|
||||||
"## 标注规则说明\n" + note_text + "\n\n## 待审分块\n" + "\n".join(f"[{_sid(s)}] {s.marked}" for s in chunk)
|
"## 标注规则说明\n" + note_text + "\n\n## 待审分块\n" + "\n".join(f"[{_sid(s)}] {s.marked}" for s in chunk)
|
||||||
)
|
)
|
||||||
data = await ai.run_json(
|
async with gate:
|
||||||
ctx,
|
return await ai.run_json(
|
||||||
job_type="parse_job",
|
ctx,
|
||||||
job_id=job_id,
|
job_type="parse_job",
|
||||||
task="extract.substantive",
|
job_id=job_id,
|
||||||
prompt_id=PROMPT_ID,
|
task="extract.substantive",
|
||||||
version=PROMPT_VERSION,
|
prompt_id=PROMPT_ID,
|
||||||
user=user,
|
version=PROMPT_VERSION,
|
||||||
)
|
user=user,
|
||||||
|
)
|
||||||
|
|
||||||
|
# 分块并发调用模型(架构 §3.4);结果按分块顺序处理,清单顺序和原文一致
|
||||||
|
results = await asyncio.gather(*(call(c) for c in chunks))
|
||||||
|
for chunk, data in zip(chunks, results, strict=True):
|
||||||
|
by_id = {_sid(s): s for s in chunk}
|
||||||
for it in data.get("items") or []:
|
for it in data.get("items") or []:
|
||||||
if not isinstance(it, dict):
|
if not isinstance(it, dict):
|
||||||
continue
|
continue
|
||||||
|
|||||||
@@ -13,7 +13,7 @@ DEV_SECRET = "dev-insecure-secret-change-me-0000000000000000" # noqa: S105 -
|
|||||||
|
|
||||||
|
|
||||||
class Settings(BaseSettings):
|
class Settings(BaseSettings):
|
||||||
model_config = SettingsConfigDict(env_prefix="BD_", env_file=".env", extra="ignore")
|
model_config = SettingsConfigDict(env_prefix="BD_", env_file=(".env", "../.env"), extra="ignore")
|
||||||
|
|
||||||
env: Literal["dev", "test", "prod"] = "dev"
|
env: Literal["dev", "test", "prod"] = "dev"
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user