feat: inject run_id as cache_salt for per-epoch resampling (algo #10)
This commit is contained in:
@@ -388,6 +388,7 @@ async def _run_single_question(
|
||||
log: HarnessLog,
|
||||
max_steps: int,
|
||||
plugins: list[object],
|
||||
run_id: str,
|
||||
) -> dict[str, Any]:
|
||||
"""执行单道题目的 Agent 推理。
|
||||
|
||||
@@ -402,6 +403,8 @@ async def _run_single_question(
|
||||
log: HarnessLog 实例(线程安全)。
|
||||
max_steps: AgentLoop 最大步数。
|
||||
plugins: pluggy 插件列表。
|
||||
run_id: 运行标识,用作 cache_salt——run_id 含 _e{epoch} 天然跨 epoch 重采样、
|
||||
同 epoch 续跑命中缓存(算法 #10 透传)。
|
||||
|
||||
返回:
|
||||
预测结果字典(含 video_id, question_id, prediction, answer 等)。
|
||||
@@ -431,6 +434,7 @@ async def _run_single_question(
|
||||
dispatcher,
|
||||
plugins=plugins,
|
||||
session_id=qa.question_id,
|
||||
cache_salt=run_id,
|
||||
)
|
||||
|
||||
result_dict = loop_result.result if isinstance(loop_result.result, dict) else {}
|
||||
@@ -558,6 +562,7 @@ async def run_inference(
|
||||
log=log,
|
||||
max_steps=max_steps,
|
||||
plugins=plugins,
|
||||
run_id=run_id,
|
||||
)
|
||||
logger.info(
|
||||
"[{}/{}] {} QA {} 完成 (stop={})",
|
||||
|
||||
+10
-3
@@ -108,6 +108,7 @@ class AgentLoop:
|
||||
plugins: list[object] | None = None,
|
||||
*,
|
||||
session_id: str | None = None,
|
||||
cache_salt: str | None = None,
|
||||
) -> LoopResult:
|
||||
"""执行 Thinking+JSON 推理循环。
|
||||
|
||||
@@ -117,6 +118,7 @@ class AgentLoop:
|
||||
tool_dispatcher: 工具调度器,ToolDispatcher Protocol 实例。
|
||||
plugins: pluggy 插件列表。
|
||||
session_id: 会话 ID,透传给 LLMProvider。
|
||||
cache_salt: 缓存盐,透传给 LLMProvider(如训练用 run_id 跨 epoch 重采样)。
|
||||
|
||||
返回:
|
||||
LoopResult 实例,包含推理步骤、token 用量、终止原因。
|
||||
@@ -138,7 +140,7 @@ class AgentLoop:
|
||||
# Phase 1: LLM 调用(步级重试:防穿透 GovernedLLMClient 的瞬时异常)
|
||||
try:
|
||||
response = await self._call_llm_with_step_retry(
|
||||
messages, token_usage, session_id=session_id
|
||||
messages, token_usage, session_id=session_id, cache_salt=cache_salt
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error("LLM API 调用失败({}): {}", type(e).__name__, e)
|
||||
@@ -269,6 +271,7 @@ class AgentLoop:
|
||||
token_usage: dict[str, int],
|
||||
*,
|
||||
session_id: str | None = None,
|
||||
cache_salt: str | None = None,
|
||||
) -> LLMResponse:
|
||||
"""带步级重试的 LLM 调用,兜底穿透治理层重试栈的瞬时异常。
|
||||
|
||||
@@ -292,7 +295,9 @@ class AgentLoop:
|
||||
step_attempt = 0
|
||||
while True:
|
||||
try:
|
||||
return await self._call_llm(messages, token_usage, session_id=session_id)
|
||||
return await self._call_llm(
|
||||
messages, token_usage, session_id=session_id, cache_salt=cache_salt
|
||||
)
|
||||
except self._retryable_exceptions as e:
|
||||
step_attempt += 1
|
||||
if step_attempt > self._step_retries:
|
||||
@@ -315,6 +320,7 @@ class AgentLoop:
|
||||
token_usage: dict[str, int],
|
||||
*,
|
||||
session_id: str | None = None,
|
||||
cache_salt: str | None = None,
|
||||
) -> LLMResponse:
|
||||
"""调用 LLMProvider 并累加 token 使用量。
|
||||
|
||||
@@ -322,11 +328,12 @@ class AgentLoop:
|
||||
messages: 消息历史。
|
||||
token_usage: 可变字典,就地累加。
|
||||
session_id: 会话 ID,透传给 LLMProvider。
|
||||
cache_salt: 缓存盐,透传给 LLMProvider(跨 epoch 重采样)。
|
||||
|
||||
返回:
|
||||
LLMResponse 实例。
|
||||
"""
|
||||
response = await self._llm.chat(messages, session_id=session_id)
|
||||
response = await self._llm.chat(messages, session_id=session_id, cache_salt=cache_salt)
|
||||
token_usage["prompt_tokens"] += response.prompt_tokens
|
||||
token_usage["completion_tokens"] += response.completion_tokens
|
||||
return response
|
||||
|
||||
@@ -94,6 +94,17 @@ def _invalid_tool_json() -> str:
|
||||
class TestAgentLoop:
|
||||
"""AgentLoop 推理循环引擎测试。"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_forwards_cache_salt(self) -> None:
|
||||
"""AgentLoop.run(cache_salt=...) 透传到 llm.chat(算法 #10 跨 epoch 重采样)。"""
|
||||
llm = AsyncMock()
|
||||
llm.chat.return_value = _make_response(_submit_json())
|
||||
|
||||
loop = AgentLoop(llm=llm, max_steps=10)
|
||||
await loop.run("system", "user", _StubDispatcher(), cache_salt="run:e2")
|
||||
|
||||
assert llm.chat.call_args.kwargs["cache_salt"] == "run:e2"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_submit_answer_terminates_loop(self) -> None:
|
||||
"""submit_answer 终止循环 → finished, result=args, steps_used=1。"""
|
||||
|
||||
@@ -698,6 +698,7 @@ class TestConcurrencyControl:
|
||||
*,
|
||||
session_id: str | None = None,
|
||||
parent_call_id: str | None = None,
|
||||
cache_salt: str | None = None,
|
||||
) -> LLMResponse:
|
||||
nonlocal current_concurrent, max_concurrent
|
||||
current_concurrent += 1
|
||||
|
||||
Reference in New Issue
Block a user