alvinfeiandClaude Opus 5.5
76098cf060
Reconcile (add / add_extracted): when the reconcile model call fails, returns an empty 200 reply or an unparseable reply, or reconcile raises before anything is stored, the write no longer falls back to adding every extracted memory as-is with success=true. The ADD-all fallback and its degraded flags are removed. The write returns success=false, agent_error_code=RECONCILE_FAILED and the cause in reconcile_error_code (LLM_UNAVAILABLE / EMPTY_RESPONSE / JSON_PARSE_FAILED / RECONCILE_EXCEPTION). Nothing new is stored, the episode counter does not advance, and add rolls back its L1 raw (retry_safe=true), so a re-run with a new request_id is clean. Only a literal [] still means "no changes". RECONCILE_ENABLED=false keeps writing directly. Working / coding path: an extraction model failure (LLM_ERROR, EMPTY_RESPONSE, JSON_PARSE_FAILED) and an experience or fact embedding failure (EMBEDDING_FAILED) now return success=false instead of success with 0 ops or a node without a vector. Embeddings are computed before the SQLite row is written and before an older fact is superseded. A failed write does not append its trace or record chain rounds. A model that answers with nothing to remember is still a success. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
719 lines
31 KiB
Python
719 lines
31 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
Coding Memory - Extractor
|
||
|
||
一次 LLM 调用，对同一段轨迹产出双层草稿：
|
||
|
||
- 经验层 experiences（有价值才抽）：task / search_keys / solution /
|
||
boundary_envs / confidence。价值门槛严格，rename a file 这种琐碎操作不该成
|
||
memory；只抽取未来可能复发的有价值问题。
|
||
- 事实层 fact（模型可返回 0 或 1 条）：statement / files。statement 是这次会话的
|
||
小结——用户要做什么 / 大致怎么做的 / 结果如何。模型未产出时是否存原文 fallback
|
||
由 `MEMORY_CODING_VERBATIM_FALLBACK_ENABLED` 控制，默认关闭。
|
||
|
||
复用 LLMConfig.model（与 chat extractor 共用同一 LLM 配置）。
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
import time
|
||
from typing import List, Optional, Dict, Any, Tuple
|
||
|
||
from ..agent.llm_provider import LLMProvider
|
||
from ..pipelines.base import ChatMessage
|
||
from ..prompts import (
|
||
get_prompt,
|
||
normalize_language,
|
||
resolve_language,
|
||
with_output_language,
|
||
)
|
||
from .preproc import (
|
||
extract_files,
|
||
keep_last_k_user_turns,
|
||
prepare_extractor_messages,
|
||
)
|
||
from .judge import _store_coding_log
|
||
from .types import CodingMemoryDraft, CodingWriteError, WorkingFactDraft
|
||
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _resolve_configured_temperature(llm_provider) -> Optional[float]:
|
||
"""从 MemoryConfig.llm 解析温度；None = 请求里不带该参数。"""
|
||
config = getattr(llm_provider, "config", None)
|
||
llm = getattr(config, "llm", None) if config is not None else None
|
||
if llm is not None:
|
||
if getattr(llm, "omit_temperature", False):
|
||
return None
|
||
temp = getattr(llm, "temperature", None)
|
||
if temp is not None:
|
||
return float(temp)
|
||
inner = getattr(llm_provider, "_llm_config", None)
|
||
temp = getattr(inner, "temperature", None) if inner is not None else None
|
||
return float(temp) if temp is not None else None
|
||
|
||
|
||
# subject 是等值比较的硬键，一句话说明抽取跑偏了，按"没抽到主体"处理。
|
||
_MAX_SUBJECT_CHARS = 80
|
||
# details 存原文，要装得下一份完整计划表 / 配置块，但不能让单条事实吃掉整个提示词。
|
||
_MAX_DETAILS_CHARS = 4000
|
||
# 兜底 statement 是检索向量，太长会被无关内容稀释；一两段用户原话足够。
|
||
# 同一个理由也约束正常路径的长度：提示词把 statement 的软上限定在 1200 字符左右，
|
||
# 靠"给定量写成数据而非对话原句"来达到，不靠砍覆盖 —— details 不进检索向量，
|
||
# statement 是检索命中后最主要的可读字段。
|
||
_VERBATIM_STATEMENT_CHARS = 600
|
||
|
||
|
||
# ================================================================
|
||
# Prompt
|
||
# ================================================================
|
||
|
||
from ..prompts.en import CODING_EXTRACT_PROMPT as EXTRACT_PROMPT # noqa: E402,F401 (re-export)
|
||
|
||
|
||
# ================================================================
|
||
# Prompt —— scene="coding"（软件工程场景）
|
||
# ================================================================
|
||
# 与上面 EXTRACT_PROMPT(scene="working") 的关系: 同一个抽取器、同一套占位符
|
||
# ({workspace_present}/{branch_present}/{conversation_block})、同一个输出 schema,
|
||
# 只换规范。两份并存而不是改一份, 是因为两个场景对 statement 的要求不同,
|
||
# 共用一份会让其中一个场景的规范被另一个稀释。
|
||
#
|
||
# 这份独有的三块(working 侧没有):
|
||
# 1. HARD CONSTRAINT 头部 —— 前置覆盖后文任何"想要叙事"的措辞: 语法主语必须是
|
||
# CODE 或 DECISION 而非 "the user"; 不许以 "the session was completed" 收尾;
|
||
# 必须保留具体症状(报错串/失败命令/错误输出), 那是未来检索的命中面。
|
||
# 2. CODE WORK 整节 —— 四件未来 agent 无法从代码反推的东西: 分支名(逐字)、
|
||
# 哪一版赢了(settled on A; interim B dropped)、精确坐标(仓相对路径+符号,
|
||
# 不是贴代码)、verified 还是仅 written。它明确不要求更长, 要求同样预算花在
|
||
# 更高价值的 token 上。
|
||
# 3. statement 的三段式 —— OPEN 症状(用户会怎么报) → THEN 坐标+决策 →
|
||
# CLOSE 证据状态。硬顶 ~400 latin 字符。
|
||
#
|
||
# 这份少掉的(working 侧有、coding 侧刻意不带):
|
||
# · "问用户"排除段 —— 卡在向用户要参数时强制 experiences: []。coding 场景的
|
||
# 轨迹是 agent 自主执行, 不存在这种停顿。
|
||
# · statement PART 2 的 TASK-SCOPED GIVENS 逐行记法 —— coding 侧的给定量是
|
||
# 代码坐标, 已由 CODE WORK 节覆盖。
|
||
# · details 的产物复制规范 —— coding 侧 details 不进检索向量也不注入。
|
||
#
|
||
# 修改 prompt 时应使用代表性 coding traces 做回归，确认坐标、决策与验证状态仍保留。
|
||
|
||
from ..prompts.en import CODING_EXTRACT_PROMPT_CODING as EXTRACT_PROMPT_CODING # noqa: E402,F401 (re-export)
|
||
|
||
|
||
# ================================================================
|
||
# CodingMemoryExtractor
|
||
# ================================================================
|
||
|
||
class CodingMemoryExtractor:
|
||
"""
|
||
LLM 驱动的双层抽取器。
|
||
|
||
输入：本次完整 New Messages + 历史最近 K 个 user-turn；两部分的 LLM 视图均限长
|
||
输出：(List[CodingMemoryDraft], List[WorkingFactDraft])
|
||
|
||
事实层一段轨迹只产一条，所以返回的 fact 列表长度恒为 0 或 1；用列表而不用
|
||
Optional 是为了让 writer / reconciler 的签名不必跟着改。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
llm_provider: LLMProvider,
|
||
max_tokens: int = 6000,
|
||
temperature: Optional[float] = None,
|
||
cache: Any = None,
|
||
scene: str = "working",
|
||
verbatim_fallback_enabled: bool = False,
|
||
):
|
||
# scene 决定用哪份抽取规范。直接构造 extractor 时默认 working；client 的
|
||
# scene=None 自动路由会先经三分类 judge，再构造对应 variant。
|
||
# "working" → EXTRACT_PROMPT （办公 / 通用生产力）
|
||
# "coding" → EXTRACT_PROMPT_CODING （软件工程）
|
||
# 非法值不抛错、退回 working：抽取是一次性的，为一个拼错的参数让整段
|
||
# 写入失败不值得。
|
||
self.scene = "coding" if str(scene or "").strip().lower() == "coding" else "working"
|
||
self._prompt_tmpl = (
|
||
EXTRACT_PROMPT_CODING if self.scene == "coding" else EXTRACT_PROMPT
|
||
)
|
||
self.llm_provider = llm_provider
|
||
self.max_tokens = max_tokens
|
||
# 默认关闭：原文兜底可能把 assistant/tool 轨迹直接写进 statement，随后被
|
||
# 搜索消费者当作记忆正文展示。需要旧行为时由 CodingConfig 显式开启。
|
||
self.verbatim_fallback_enabled = bool(verbatim_fallback_enabled)
|
||
# cache 非空时落 CODING_EXTRACT pipeline log（best-effort）
|
||
self.cache = cache
|
||
# None → 使用 LLMConfig.temperature（避免硬编码触发 kimi/deepseek 的温度硬约束）
|
||
if temperature is None:
|
||
# None 一路传到 provider = 不带该参数；此前兜 0.1 会绕过用户配置，
|
||
# 在只接受特定 temperature 的网关上（Opus 5 只收 1 或不传）直接 400
|
||
temperature = _resolve_configured_temperature(llm_provider)
|
||
self.temperature = temperature
|
||
|
||
@staticmethod
|
||
def _language_text(messages: List[ChatMessage]) -> str:
|
||
"""Use new user text for routing; tool output must not dominate it."""
|
||
user_parts = [
|
||
str(message.content or "").strip()
|
||
for message in messages
|
||
if message.role == "user" and str(message.content or "").strip()
|
||
]
|
||
if user_parts:
|
||
return "\n".join(user_parts)
|
||
|
||
non_tool_parts = [
|
||
str(message.content or "").strip()
|
||
for message in messages
|
||
if message.role != "tool" and str(message.content or "").strip()
|
||
]
|
||
if non_tool_parts:
|
||
return "\n".join(non_tool_parts)
|
||
return "\n".join(
|
||
str(message.content or "").strip()
|
||
for message in messages
|
||
if str(message.content or "").strip()
|
||
)
|
||
|
||
def _select_prompt_template(
|
||
self, messages: List[ChatMessage], language: Optional[str] = None,
|
||
) -> Tuple[str, str]:
|
||
"""Return (template, language code) for this write's new messages.
|
||
|
||
Explicit ``language`` wins; otherwise zh/en auto-detection over the new
|
||
user text (unchanged behaviour).
|
||
"""
|
||
code = resolve_language(language, self._language_text(messages))
|
||
if code == "en":
|
||
return self._prompt_tmpl, code
|
||
name = (
|
||
"CODING_EXTRACT_PROMPT_CODING" if self.scene == "coding"
|
||
else "CODING_EXTRACT_PROMPT"
|
||
)
|
||
return get_prompt(name, code), code
|
||
|
||
async def extract(
|
||
self,
|
||
messages: List[ChatMessage],
|
||
*,
|
||
history_messages: Optional[List[ChatMessage]] = None,
|
||
tool_result_max_bytes: int = 2048,
|
||
user_id: str,
|
||
agent_id: str = "default_agent",
|
||
workspace_id: Optional[str] = None,
|
||
branch: Optional[str] = None,
|
||
session_id: Optional[str] = None,
|
||
request_id: str = "",
|
||
language: Optional[str] = None,
|
||
) -> Tuple[List[CodingMemoryDraft], List[WorkingFactDraft]]:
|
||
"""
|
||
Args:
|
||
messages: 本次 write 的完整新消息；只做单条限长，不裁 turn
|
||
history_messages: 之前 write 的历史消息；仅取最近 5 个 user-turn 作消解上下文
|
||
user_id / agent_id / workspace_id / branch / session_id: SDK 注入
|
||
language: 显式 prompt 语言码；None 时按新 user 文本做 zh/en 自动检测
|
||
|
||
Returns:
|
||
(经验草稿, 事实草稿)。两层都可能为空。只有显式启用
|
||
verbatim_fallback_enabled 时，模型没产出 fact 才调用 _verbatim_fact。
|
||
|
||
Raises:
|
||
CodingWriteError: 未开启原文兜底时，LLM 调用失败（LLM_ERROR）、返回空
|
||
content（EMPTY_RESPONSE）或回复解析不出 JSON 对象（JSON_PARSE_FAILED）。
|
||
"""
|
||
if not messages:
|
||
return [], []
|
||
|
||
# New Messages 是本次 write 的抽取主体：无论传入 1 个还是 10 个 user-turn，
|
||
# 都必须完整进入 prompt；这里只限制每条 tool / assistant 的体积。
|
||
new_messages = prepare_extractor_messages(
|
||
messages, tool_max_bytes=tool_result_max_bytes,
|
||
)
|
||
|
||
# Last 5 User Turns 是之前 write 的历史，只用于指代消解和上下文。绝不能拿
|
||
# last-K 去裁本次输入，否则 write(10 turns) 会静默丢掉前 5 个 turn。
|
||
history_window = keep_last_k_user_turns(
|
||
prepare_extractor_messages(
|
||
history_messages or [], tool_max_bytes=tool_result_max_bytes,
|
||
),
|
||
k=5,
|
||
)
|
||
|
||
history_block = self._format_conversation(history_window).strip()
|
||
new_block = self._format_conversation(new_messages).strip()
|
||
explicit_language = normalize_language(language)
|
||
prompt_tmpl, prompt_language = self._select_prompt_template(
|
||
new_messages, explicit_language,
|
||
)
|
||
empty_block = get_prompt("CODING_EMPTY_BLOCK", prompt_language)
|
||
conversation_block = get_prompt(
|
||
"CODING_CONVERSATION_BLOCK", prompt_language,
|
||
).format(
|
||
history=history_block or empty_block,
|
||
new=new_block or empty_block,
|
||
)
|
||
|
||
prompt = prompt_tmpl.format(
|
||
workspace_present=str(bool(workspace_id)).lower(),
|
||
branch_present=str(bool(branch)).lower(),
|
||
conversation_block=conversation_block,
|
||
)
|
||
prompt = with_output_language(
|
||
prompt, prompt_language, explicit=explicit_language is not None,
|
||
)
|
||
|
||
_t0 = time.perf_counter()
|
||
try:
|
||
resp = await self.llm_provider.complete(
|
||
prompt=prompt,
|
||
max_tokens=self.max_tokens,
|
||
audit_stage="coding_extract",
|
||
)
|
||
except Exception as e:
|
||
logger.warning(f"[coding-extract] LLM call failed: {e}")
|
||
# 经验层不兜底。事实层原文兜底默认关闭，避免 assistant/tool 轨迹被
|
||
# 当作 statement 直接展示；关闭时 LLM 失败抛 CodingWriteError(LLM_ERROR)，
|
||
# 整次写入报失败，而不是「成功、0 条」。仅在配置显式开启时保留旧行为。
|
||
verbatim = (
|
||
self._verbatim_fact(messages)
|
||
if self.verbatim_fallback_enabled else None
|
||
)
|
||
fact_source = (
|
||
"verbatim_fallback" if verbatim
|
||
else "verbatim_fallback_disabled"
|
||
if not self.verbatim_fallback_enabled
|
||
else "none"
|
||
)
|
||
await _store_coding_log(
|
||
self.cache, request_id=request_id, user_id=user_id, agent_id=agent_id,
|
||
session_id=session_id or "", step="CODING_EXTRACT", prompt=prompt,
|
||
response=f"[LLM error] {type(e).__name__}: {str(e)[:300]}",
|
||
parsed=json.dumps(
|
||
{
|
||
"drafts": 0,
|
||
"llm_error": True,
|
||
"facts": [verbatim.to_dict()] if verbatim else [],
|
||
"fact_count": 1 if verbatim else 0,
|
||
"fact_source": fact_source,
|
||
},
|
||
ensure_ascii=False,
|
||
),
|
||
elapsed_ms=round((time.perf_counter() - _t0) * 1000, 2),
|
||
)
|
||
if not self.verbatim_fallback_enabled:
|
||
raise CodingWriteError(
|
||
"LLM_ERROR",
|
||
f"coding extract LLM call failed: {type(e).__name__}: {e}",
|
||
) from e
|
||
return [], ([verbatim] if verbatim else [])
|
||
_elapsed = round((time.perf_counter() - _t0) * 1000, 2)
|
||
|
||
# 模型答了且没有可记的内容（如 {"experiences": [], "fact": null}）是正常的空
|
||
# 抽取；200 但 content 为空、或回复解析不出 JSON 对象，是模型失败，不能当成
|
||
# 「成功、0 条」。原文兜底显式开启时沿用旧行为（交给下面的 verbatim 兜底）。
|
||
failure = self._output_failure(resp.content)
|
||
if failure is not None and not self.verbatim_fallback_enabled:
|
||
code, message = failure
|
||
await _store_coding_log(
|
||
self.cache, request_id=request_id, user_id=user_id, agent_id=agent_id,
|
||
session_id=session_id or "", step="CODING_EXTRACT", prompt=prompt,
|
||
response=resp.content or "",
|
||
parsed=json.dumps(
|
||
{"drafts": 0, "error": code, "message": message},
|
||
ensure_ascii=False,
|
||
),
|
||
elapsed_ms=_elapsed, prompt_tokens=resp.prompt_tokens,
|
||
completion_tokens=resp.completion_tokens, total_tokens=resp.tokens_used,
|
||
)
|
||
raise CodingWriteError(code, message)
|
||
|
||
raw_experiences, raw_facts = self._parse_output(resp.content)
|
||
|
||
# 经验层的 files 走规则抽取（整段轨迹碰过的文件）；事实层的 files 由 LLM
|
||
# 给，它挑的是"这次真正动过/查过的"，比规则抽取的全集更贴会话小结
|
||
files = extract_files(messages)
|
||
|
||
drafts: List[CodingMemoryDraft] = []
|
||
for raw in raw_experiences:
|
||
d = self._normalize_draft(
|
||
raw,
|
||
user_id=user_id,
|
||
agent_id=agent_id,
|
||
workspace_id=workspace_id,
|
||
branch=branch,
|
||
session_id=session_id,
|
||
files=files,
|
||
)
|
||
if d is not None:
|
||
drafts.append(d)
|
||
|
||
fact_drafts: List[WorkingFactDraft] = []
|
||
for raw in raw_facts:
|
||
f = self._normalize_fact_draft(raw)
|
||
if f is not None:
|
||
fact_drafts.append(f)
|
||
|
||
# details 不信 LLM 的抄写，改从未压缩的 messages 直接取。
|
||
#
|
||
# 抽取提示词里的对话是窗口化、截断过的（见 preproc），长 assistant 文本的
|
||
# 中段在建 prompt 时已经不在了。让模型"逐字抄写"一段它看不到的原文，只能
|
||
# 抄出带省略标记的头尾。
|
||
#
|
||
# details 的意义是留住被摘要丢掉的可寻址产物（计划表、配置块等）。它不进
|
||
# 检索向量（见 store._upsert_fact_vec），长度不影响召回，没有理由再走一遍
|
||
# 有损通道——直接取原文，只受 details 自己的上限约束。
|
||
#
|
||
# 注意：绕过一个有损环节不等于修好它。输入侧的截断阈值同样会影响 statement，
|
||
# 见 preproc.DEFAULT_ASSISTANT_HEAD_CHARS。
|
||
#
|
||
# LLM 给的 details 仍然留着当兜底：messages 为空这种情况下它是唯一来源。
|
||
verbatim_details = self._verbatim_details(messages)
|
||
if verbatim_details:
|
||
fact_drafts = [
|
||
WorkingFactDraft(
|
||
statement=f.statement,
|
||
files=f.files,
|
||
subject=f.subject,
|
||
details=verbatim_details,
|
||
)
|
||
for f in fact_drafts
|
||
]
|
||
|
||
fact_source = "llm"
|
||
if not fact_drafts:
|
||
if self.verbatim_fallback_enabled:
|
||
verbatim = self._verbatim_fact(messages)
|
||
if verbatim is not None:
|
||
fact_drafts.append(verbatim)
|
||
fact_source = "verbatim_fallback"
|
||
else:
|
||
fact_source = "none"
|
||
else:
|
||
fact_source = "verbatim_fallback_disabled"
|
||
|
||
logger.info(
|
||
f"[coding-extract] produced {len(drafts)} experience draft(s), "
|
||
f"{'1 session fact' if fact_drafts else 'no session fact'}"
|
||
f"{' (verbatim fallback: LLM returned none)' if fact_source == 'verbatim_fallback' else ''}"
|
||
f"{' (verbatim fallback disabled)' if fact_source == 'verbatim_fallback_disabled' else ''}"
|
||
f" (workspace={workspace_id!r}, branch={branch!r})"
|
||
)
|
||
# CODING_EXTRACT：无论产出几条 draft（含 0 条）都落一条，便于排查"没抽到"。
|
||
# 双层架构下 parsed 同时记录经验草稿与事实草稿，方便追查两层各自产出。
|
||
await _store_coding_log(
|
||
self.cache, request_id=request_id, user_id=user_id, agent_id=agent_id,
|
||
session_id=session_id or "", step="CODING_EXTRACT", prompt=prompt,
|
||
response=resp.content or "",
|
||
parsed=json.dumps(
|
||
{
|
||
"experiences": [d.to_dict() for d in drafts],
|
||
"experience_count": len(drafts),
|
||
"facts": [f.to_dict() for f in fact_drafts],
|
||
"fact_count": len(fact_drafts),
|
||
"fact_source": fact_source,
|
||
},
|
||
ensure_ascii=False, default=str,
|
||
),
|
||
elapsed_ms=_elapsed, prompt_tokens=resp.prompt_tokens,
|
||
completion_tokens=resp.completion_tokens, total_tokens=resp.tokens_used,
|
||
)
|
||
return drafts, fact_drafts
|
||
|
||
# ------------------------------------------------------------
|
||
# Helpers
|
||
# ------------------------------------------------------------
|
||
|
||
@classmethod
|
||
def _verbatim_details(cls, messages: List[ChatMessage]) -> str:
|
||
"""整段轨迹的原文，作为 details 的权威来源。
|
||
|
||
取的是未经 keep_last_k_user_turns 压缩的 messages：details 存在的理由就是
|
||
留住摘要会丢掉的可寻址产物（第 N 天第 M 餐这种格子），而压缩正是丢它的那一步。
|
||
它不进检索向量（store._upsert_fact_vec 只 embed subject + statement），所以
|
||
长度只影响存储，不影响召回质量，唯一的上限是 _MAX_DETAILS_CHARS。
|
||
|
||
从尾部截断：头部是结构化产物的开头（Day 1、配置首行），信息密度最高。
|
||
"""
|
||
if not messages:
|
||
return ""
|
||
body = cls._format_conversation(messages).strip()
|
||
if not body:
|
||
return ""
|
||
if len(body) <= _MAX_DETAILS_CHARS:
|
||
return body
|
||
return body[:_MAX_DETAILS_CHARS] + "\n…(truncated)"
|
||
|
||
@classmethod
|
||
def _verbatim_fact(
|
||
cls, messages: List[ChatMessage]
|
||
) -> Optional[WorkingFactDraft]:
|
||
"""LLM 没给 fact 时的兜底：把轨迹原文存成一条事实。
|
||
|
||
只在真的一个字都没有时返回 None（空消息、全空内容）。这是最后一道闸门，
|
||
判据必须是"有没有内容"，不能是"内容够不够好"——够不够好是上面那次 LLM
|
||
调用的判断，它已经判错过了，兜底再复读一遍同样的判断就没有意义。
|
||
|
||
statement 用于检索，所以取用户说的话（用户的话就是未来查询的措辞，
|
||
assistant 的复述和工具输出不是）；details 放整段原文，供命中后读取。
|
||
subject 留空——主体只能靠语义判断，规则抽不出来，空串会让 reconciler 退回
|
||
改动前的纯语义判覆盖，而不是拿一个猜错的锚点去卡。
|
||
"""
|
||
if not messages:
|
||
return None
|
||
|
||
user_texts: List[str] = []
|
||
for m in messages:
|
||
if m.role == "user" and not m.is_tool_message():
|
||
text = (m.content or "").strip()
|
||
if text:
|
||
user_texts.append(text)
|
||
|
||
details = cls._verbatim_details(messages)
|
||
if not details:
|
||
return None
|
||
|
||
# 用户原话优先，一句都没有（纯工具回放）就退回整段轨迹的开头
|
||
head = "\n".join(user_texts).strip() or details
|
||
statement = head[:_VERBATIM_STATEMENT_CHARS]
|
||
if len(head) > _VERBATIM_STATEMENT_CHARS:
|
||
statement += "…"
|
||
|
||
return WorkingFactDraft(statement=statement, files=[], subject="", details=details)
|
||
|
||
@staticmethod
|
||
def _format_conversation(messages: List[ChatMessage]) -> str:
|
||
"""渲染整段对话给 LLM 读。"""
|
||
rows: List[str] = []
|
||
for i, m in enumerate(messages):
|
||
if m.role == "user" and not m.is_tool_message():
|
||
rows.append(f"[{i}] user: {m.content}")
|
||
elif m.role == "assistant":
|
||
line = f"[{i}] assistant: {m.content}"
|
||
if m.tool_calls:
|
||
tc_summary = "; ".join(
|
||
f"{tc.name}({_compact_args(tc.arguments)})"
|
||
for tc in m.tool_calls
|
||
)
|
||
line += f" [tools: {tc_summary}]"
|
||
rows.append(line)
|
||
elif m.is_tool_message():
|
||
tn = m.tool_name or "?"
|
||
rows.append(f"[{i}] tool({tn}): {m.content}")
|
||
elif m.role == "system":
|
||
rows.append(f"[{i}] system: {m.content}")
|
||
else:
|
||
rows.append(f"[{i}] {m.role}: {m.content}")
|
||
return "\n".join(rows)
|
||
|
||
@classmethod
|
||
def _output_failure(cls, text: Optional[str]) -> Optional[Tuple[str, str]]:
|
||
"""回复本身是失败时返回 (error_code, message)；正常（含空抽取）返回 None。"""
|
||
if not (text or "").strip():
|
||
return "EMPTY_RESPONSE", "coding extract LLM returned empty response"
|
||
if cls._load_output_object(text) is None:
|
||
return (
|
||
"JSON_PARSE_FAILED",
|
||
f"coding extract output is not a JSON object ({len(text)} chars)",
|
||
)
|
||
return None
|
||
|
||
@staticmethod
|
||
def _load_output_object(text: Optional[str]) -> Optional[Dict[str, Any]]:
|
||
"""宽松解析 LLM 输出，返回顶层 JSON 对象；解析不出返回 None。"""
|
||
if not text:
|
||
return None
|
||
s = text.strip()
|
||
# 去 markdown fence
|
||
if s.startswith("```"):
|
||
lines = s.split("\n")
|
||
body, in_fence = [], False
|
||
for line in lines:
|
||
if line.startswith("```"):
|
||
in_fence = not in_fence
|
||
continue
|
||
body.append(line)
|
||
s = "\n".join(body).strip() or s
|
||
|
||
obj = None
|
||
# 尝试纯 JSON
|
||
try:
|
||
obj = json.loads(s)
|
||
except Exception:
|
||
# 找第一个 {...} 块
|
||
start = s.find("{")
|
||
end = s.rfind("}")
|
||
if start != -1 and end > start:
|
||
try:
|
||
obj = json.loads(s[start: end + 1])
|
||
except Exception:
|
||
obj = None
|
||
return obj if isinstance(obj, dict) else None
|
||
|
||
@classmethod
|
||
def _parse_output(
|
||
cls, text: str
|
||
) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
|
||
"""宽松解析 LLM 输出 JSON，返回 (experiences, facts) 两个原始列表。"""
|
||
if not text:
|
||
return [], []
|
||
obj = cls._load_output_object(text)
|
||
if obj is None:
|
||
logger.warning(
|
||
f"[coding-extract] failed to parse output ({len(text or '')} chars)"
|
||
)
|
||
logger.debug(f"[coding-extract] unparsable output: {text!r}")
|
||
return [], []
|
||
|
||
# "memories" 是双层化之前的 key，老 prompt 缓存 / 旧模型可能还这么回
|
||
experiences = cls._as_dict_list(obj.get("experiences"))
|
||
if not experiences:
|
||
experiences = cls._as_dict_list(obj.get("memories"))
|
||
return experiences, cls._as_fact_list(obj)
|
||
|
||
@classmethod
|
||
def _as_fact_list(cls, obj: Dict[str, Any]) -> List[Dict[str, Any]]:
|
||
"""取出这一段的唯一事实，返回长度 0 或 1 的列表。
|
||
|
||
现契约是单个 "fact" 对象。"facts" 数组是一段轨迹多条事实那版的形状，老
|
||
模型和缓存住的老 prompt 还会这么回，所以继续认，但只留第一条——多出来的
|
||
本来就不该存在，丢掉比让它们各自成为一条会话记录更接近本意。
|
||
"""
|
||
fact = obj.get("fact")
|
||
if isinstance(fact, dict):
|
||
return [fact]
|
||
|
||
legacy = cls._as_dict_list(obj.get("facts"))
|
||
if len(legacy) > 1:
|
||
logger.warning(
|
||
f"[coding-extract] legacy 'facts' array with {len(legacy)} entries; "
|
||
f"keeping the first, dropping {len(legacy) - 1}"
|
||
)
|
||
return legacy[:1]
|
||
|
||
@staticmethod
|
||
def _as_dict_list(value: Any) -> List[Dict[str, Any]]:
|
||
if not isinstance(value, list):
|
||
return []
|
||
return [v for v in value if isinstance(v, dict)]
|
||
|
||
@staticmethod
|
||
def _normalize_draft(
|
||
raw: Dict[str, Any],
|
||
*,
|
||
user_id: str,
|
||
agent_id: str,
|
||
workspace_id: Optional[str],
|
||
branch: Optional[str],
|
||
session_id: Optional[str],
|
||
files: List[str],
|
||
) -> Optional[CodingMemoryDraft]:
|
||
"""
|
||
把 LLM 输出的单条 memory 字典规范化为 CodingMemoryDraft。
|
||
|
||
返回 None 表示该条无效（缺 task 或 solution）。
|
||
"""
|
||
task = (raw.get("task") or "").strip()
|
||
solution = (raw.get("solution") or "").strip()
|
||
if not task or not solution:
|
||
logger.debug(f"[coding-extract] drop draft missing task/solution: {raw}")
|
||
return None
|
||
|
||
# search_keys 规范化
|
||
sk_raw = raw.get("search_keys") or []
|
||
if not isinstance(sk_raw, list):
|
||
sk_raw = []
|
||
search_keys = [str(s).strip() for s in sk_raw if str(s).strip()]
|
||
# 去重 + 与 task 同名的 key 去掉
|
||
seen = set()
|
||
deduped = []
|
||
for k in search_keys:
|
||
if k == task:
|
||
continue
|
||
if k in seen:
|
||
continue
|
||
seen.add(k)
|
||
deduped.append(k)
|
||
search_keys = deduped[:5] # 软上限 5
|
||
|
||
# confidence 规范化
|
||
try:
|
||
confidence = float(raw.get("confidence", 0.7))
|
||
except Exception:
|
||
confidence = 0.7
|
||
confidence = max(0.0, min(1.0, confidence))
|
||
|
||
boundary_envs = (raw.get("boundary_envs") or "").strip()
|
||
|
||
return CodingMemoryDraft(
|
||
task=task,
|
||
search_keys=search_keys,
|
||
solution=solution,
|
||
boundary_envs=boundary_envs,
|
||
confidence=confidence,
|
||
user_id=user_id,
|
||
agent_id=agent_id,
|
||
workspace_id=workspace_id,
|
||
branch=branch,
|
||
session_id=session_id,
|
||
files=list(files),
|
||
extraction_source="auto_extract",
|
||
)
|
||
|
||
@staticmethod
|
||
def _normalize_fact_draft(raw: Dict[str, Any]) -> Optional[WorkingFactDraft]:
|
||
"""把 LLM 输出的单条 fact 规范化。返回 None 表示 statement 为空。"""
|
||
statement = (raw.get("statement") or "").strip()
|
||
if not statement:
|
||
logger.debug(f"[coding-extract] drop fact missing statement: {raw}")
|
||
return None
|
||
|
||
files_raw = raw.get("files") or []
|
||
if not isinstance(files_raw, list):
|
||
files_raw = []
|
||
seen, files = set(), []
|
||
for f in files_raw:
|
||
f = str(f).strip()
|
||
if f and f not in seen:
|
||
seen.add(f)
|
||
files.append(f)
|
||
|
||
subject = " ".join(str(raw.get("subject") or "").split())
|
||
# subject 是硬键而不是散文，一句话说明整段跑偏了，宁可当"没定出主体"处理
|
||
# （退回纯语义判覆盖）也不要拿一句话去比对相等。
|
||
if len(subject) > _MAX_SUBJECT_CHARS:
|
||
logger.debug(
|
||
f"[coding-extract] subject too long, treated as absent: {subject!r}"
|
||
)
|
||
subject = ""
|
||
|
||
details = str(raw.get("details") or "").strip()
|
||
if len(details) > _MAX_DETAILS_CHARS:
|
||
# 从尾部截，头部是结构化产物的开头（Day 1、配置首行），信息密度最高
|
||
details = details[:_MAX_DETAILS_CHARS] + "\n…(truncated)"
|
||
|
||
return WorkingFactDraft(
|
||
statement=statement, files=files, subject=subject, details=details
|
||
)
|
||
|
||
|
||
# ================================================================
|
||
# Helpers
|
||
# ================================================================
|
||
|
||
def _compact_args(args: Dict[str, Any], max_chars: int = 600) -> str:
|
||
"""单行展示 tool args，并为关键命令参数保留足够证据窗口。
|
||
|
||
仍设上限，避免整段脚本/大 payload 撑爆 prompt；但 80 字符会在常见 PowerShell
|
||
命令的 ``-Depth`` / ``-ErrorAction`` 等参数前截断，与 fact 的证据忠实度契约冲突。
|
||
"""
|
||
if not args:
|
||
return ""
|
||
try:
|
||
s = json.dumps(args, ensure_ascii=False, separators=(",", ":"))
|
||
except Exception:
|
||
s = str(args)
|
||
if len(s) > max_chars:
|
||
s = s[: max_chars - 3] + "..."
|
||
return s
|