Repository navigation
feat: 支持codeact 特性 - #341
Conversation
raychen911
commented
Sep 22, 2026
- CodeAct Tools: 支持工具返回的大对象可以保留在本地,节省token
- Generation Method: 可生成用户定义的函数
1f8ffb3 to
81ab701
Compare
AI Code Review审查结论不通过 审查范围:5fac61f..1f8ffb3(46 个文件,+6064/-27),计划 "feat: 支持codeact 特性"。计划符合性:CodeAct 包(config/runtime/processor/objects/工具/存储)、LlmAgent 集成、文档与测试均大体交付,功能主路径按计划实现。主要风险:① FilterRunner 初始化顺序回归—— 发现的问题严重
问题: 触发条件: 任何以字符串形式传入过滤器配置的构造路径: 实际影响: 使用字符串过滤器名的工具/模型在构造阶段即抛未捕获的 修正方向: 将 严重
问题: 触发条件: 配置 实际影响: 修正方向: 在 中等
问题: 触发条件: 同一 agent 的两个及以上 invocation 并发执行含耗时 cell(如 实际影响: 并发请求的 CodeAct 执行被串行化,吞吐下降且延迟被最慢 cell 放大;单个挂起 cell 会阻塞所有其他 invocation 的 cell 执行,形成头阻塞。 修正方向: 将锁粒度改为按 invocation 隔离(如 中等
问题: 触发条件: 任何以 实际影响: 修正方向: 移除 较低
问题: 本次新增的 CodeAct 测试( 触发条件: 后续修改 实际影响: 限制/取消传播缺陷(见上文 SEVERE)与过滤器初始化回归均无回归保护,问题会随版本发布静默遗留。 修正方向: 新增针对 |
| try: | ||
| async with self._execution_lock: | ||
| with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): | ||
| for cell in cells: | ||
| await self._execute_cell(cell, state.globals) | ||
| except CodeActReturnSignal: | ||
| pass | ||
| except Exception: # noqa: BLE001 # pylint: disable=broad-except | ||
| stderr.write(traceback.format_exc()) | ||
|
|
||
| return create_codeact_execution_result(stdout=stdout.getvalue(), stderr=stderr.getvalue()) |
There was a problem hiding this comment.
问题: InProcessCodeActRuntime.execute 用 except Exception 兜底(:99),而 RunLimitException/RunCancelledException 均继承 TrpcAgentException(Exception),会被一并捕获并仅写入 stderr。CodeAct 模式下工具调用经 CodeActAgentProxy.invoke 每次执行 self._context.raise_if_limit(RunLimitType.MAX_TOOL_CALLS)(_objects.py:231),该异常在 cell 内被吞掉后转成 code_execution_result observation 反馈给模型,与 runner 中 except RunLimitException: raise 的传播路径(runners.py:630-639,含 cleanup_incomplete_function_calls 与 trace/raise)完全断开。docs/mkdocs/en/codeact.md:269 声称 "Per-invocation tool call limits" 被保留,实际并未生效。
触发条件: 配置 RunConfig(max_tool_calls=N)(或 agent 级限制)后运行 CodeAct agent,模型在 cell 内持续调用 await self.<tool>(...) 超过 N 次;或 cell 执行期间调用方触发 cancel_run_async/raise_if_cancelled。
实际影响: max_tool_calls 限制被静默绕过——工具调用成本无上限累加、运行无法按预期终止,模型反而把限制异常文本当作执行错误继续重试;取消信号被吞掉后 run 无法在检查点间及时停止。
修正方向: 在 except Exception 之前增加 except (RunLimitException, RunCancelledException): raise 原样传播(与 _llm_agent.py 中 except RunLimitException: raise 的既有约定一致),并补充对 CodeActAgentProxy 超限与取消路径的测试。
| _states: dict[str, _ExecutionState] = PrivateAttr(default_factory=dict) | ||
| _execution_lock: asyncio.Lock = PrivateAttr(default_factory=asyncio.Lock) | ||
|
|
||
| @override | ||
| async def execute( | ||
| self, | ||
| invocation_context: InvocationContext, | ||
| cells: list[str], | ||
| ) -> CodeActExecutionResult: | ||
| """Execute Python cells while retaining globals for the invocation.""" | ||
| execution_id = invocation_context.invocation_id | ||
| state = self._states.get(execution_id) | ||
| if state is None: | ||
| state = self._create_state(invocation_context) | ||
| self._states[execution_id] = state | ||
| await state.agent_proxy.refresh(invocation_context) | ||
|
|
||
| stdout = io.StringIO() | ||
| stderr = io.StringIO() | ||
| try: | ||
| async with self._execution_lock: | ||
| with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): | ||
| for cell in cells: | ||
| await self._execute_cell(cell, state.globals) |
There was a problem hiding this comment.
问题: _execution_lock 是实例级单一 asyncio.Lock(:74),execute 中所有 invocation 的 cell 执行共享同一把锁(:93)。CodeActConfig.runtime 默认是 Field(default_factory=InProcessCodeActRuntime),同一 agent 的多个并发 Runner.run_async 调用(如 server 并发请求)复用同一个 runtime 实例,全部串行执行。
触发条件: 同一 agent 的两个及以上 invocation 并发执行含耗时 cell(如 await asyncio.sleep(...)、阻塞 IO 或长循环)时,后续 invocation 排队等待。
实际影响: 并发请求的 CodeAct 执行被串行化,吞吐下降且延迟被最慢 cell 放大;单个挂起 cell 会阻塞所有其他 invocation 的 cell 执行,形成头阻塞。
修正方向: 将锁粒度改为按 invocation 隔离(如 self._locks: dict[str, asyncio.Lock] 随 _states 创建/释放),仅在需要保证同一 invocation 内 state 一致性时持锁,跨 invocation 并行执行。
| def _init_filters(self): | ||
| """Initialize filters.""" | ||
| for filter_name in self._filters_name: | ||
| filter_instance = get_filter(self._type, filter_name) | ||
| if not filter_instance: | ||
| raise ValueError(f"Filter {filter_name} not found, type: {self._type}") | ||
| self._filters.append(filter_instance) | ||
| self._filters = self._convert_filters() | ||
| self._filters_name = [filter.name for filter in self._filters] |
There was a problem hiding this comment.
问题: _init_filters 现在将 self._filters_name 覆写为解析后过滤器的 filter.name(:73),即注册类名/显示名,替换掉调用方传入的原始注册名列表。base commit 中该属性仅解析字符串→追加解析字符串→ 追加实例化的 self._filters,名称config 行为不 名字串原字符串期间保持传入值不变;本次行为变化独立于 KeyError 崩溃存在——即使修复了 UNSUPPORTED 解析顺序,此覆写仍会发生。
触发条件: 任何以 filters_name=["custom_registered_name"] 传入并成功解析(在首个缺陷修复后)的工具/模型构造;随后任何读取 runner.filters_name 的调用方(如日志、序列化、动态增删过滤器的场景)取到的是解析后的类名而非用户配置。
实际影响: filters_name 的配置契约被破坏——用户提供的名字序列无法原样取回,依赖原始名称做去重、比对或回填配置的逻辑得到错误数据,且与 add_filters 追加后的一致性预期不符。
修正方向: 移除 self._filters_name = [filter.name for filter in self._filters] 覆写,保持 _filters_name 为用户传入的原始列表,解析结果仅存入 _filters。
| try: | ||
| async with self._execution_lock: | ||
| with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): | ||
| for cell in cells: | ||
| await self._execute_cell(cell, state.globals) | ||
| except CodeActReturnSignal: | ||
| pass | ||
| except Exception: # noqa: BLE001 # pylint: disable=broad-except | ||
| stderr.write(traceback.format_exc()) | ||
|
|
||
| return create_codeact_execution_result(stdout=stdout.getvalue(), stderr=stderr.getvalue()) |
There was a problem hiding this comment.
问题: 本次新增的 CodeAct 测试(tests/agents/test_codeact.py、tests/tools/test_code_act_tool.py、tests/tools/test_code_act_implementation.py、tests/tests/tests/tests/tests/tests/tests/tests/tests/test_tools/test_code_act_implementation.py/test 的 tests 中 tests/tests代码)覆盖了 happy path(processor 与首 cell、首 path——processor 首 cell 测试(processor 结果、cell 解析结果截断 validation 校验结果 validation 截断 validation 结果解析 feedback validation 验证 cell 截断验证从 校验、截断 validation 截断 validation 结果 截断首验证、错误 校验 validation 校验/错误错误 校验 拒绝 feedback 拒绝结果 序列化 拒绝 错误 截断错误 sequence 处理 结果错误 拒绝 拒绝;但没有任何用例覆盖 run-limit 传播(max_tool_calls 超额、max_iterations 超额)与 cancellation(RunCancelledException/cancel_run_async)在 cell 执行期间的行为;tests/filter/test_filter_runner.py 未同步更新,其中的 test_with_filters_name(未 mock registry)在字符串解析提前到 __init__ 后会直接失败,测试套件无法捕获上述两项回归。
触发条件: 后续修改 execute 的异常处理或 __init__ 顺序时,无测试可及时发现;CI 运行 tests/filter/test_filter_runner.py 时当前即报错。
实际影响: 限制/取消传播缺陷(见上文 SEVERE)与过滤器初始化回归均无回归保护,问题会随版本发布静默遗留。
修正方向: 新增针对 InProcessCodeActRuntime.execute 传入 RunLimitException/RunCancelledException 场景的传播断言测试(及 CodeActAgentProxy.invoke 超限场景),并更新 tests/filter/test_filter_runner.py 使字符串解析路径在正确类型下被覆盖。
AI Code Review审查结论不通过 审查了 base_commit=5fac61f…head_commit=81ab701(feat: 支持codeact 特性,46 文件 +6066/−27):新增 CodeAct 协议(InProcessCodeActRuntime/CodeActResponseProcessor/CodeActObjectStore/CodeActTool/三种实现存储后端)整体设计完整、测试覆盖较好(协议、运行时持久化、ObjectRef、工具拒绝反馈、存储生命周期均有用例)。但发现 4 个由本变更引入的高置信问题:(1) SEVERE —— FilterRunner 重构把 发现的问题严重
问题: 触发条件: 任何以非空 实际影响: 所有使用命名 Filter 的工具/模型构造抛 修正方向: 不要在 严重
问题: 触发条件: 用户设置 实际影响: Runner 事件流为空:调用方(如 修正方向: 当 中等
问题: 触发条件: 生成/批准的 Python cell 中任何代码向 stderr 写出内容——最典型的是 实际影响: 交互式 CodeAct 模式下模型会看到 "Code execution error" 而反复重试直到耗尽 修正方向: 只在真正未检测到 中等
问题: 触发条件: 使用 实际影响: "版本化持久化实现"在真实 Redis 后端上变成 24 小时缓存:已批准的生产实现被静默丢弃, 修正方向: 对 CodeAct 实现存储使用持久化过期策略:在 |
| self._type = FilterType.UNSUPPORTED | ||
| self._init_filters() |
There was a problem hiding this comment.
问题: FilterRunner.__init__ 在 self._type = FilterType.UNSUPPORTED(第 39 行)之后无条件调用 self._init_filters()(第 40 行),而新提取的 _convert_filters() 对 filters_name 中的字符串调用 get_filter(self._type, name) 即 get_filter(UNSUPPORTED, ...)。FilterRegistry.__init__(_registry.py:110)只注册了 TOOL/MODEL/AGENT 三种类型、从不注册 UNSUPPORTED,因此 FilterRegistry.get(_registry.py:135 的 self._filter_registry[filter_type])直接抛 KeyError。同时本提交从 BaseTool.__init__(_base_tool.py:94)和 LLMModel.__init__(_llm_model.py:54)中删除了原有"先设置 _type 再调用 _init_filters()"的调用,使初始化顺序回归无法被规避。
触发条件: 任何以非空 filters_name 构造的 FunctionTool/BaseTool 子类(如 tests/tools/test_function_tool.py:96、examples/filter_with_tool/agent/agent.py:33、examples/filter_with_model/agent/agent.py:27 的 OpenAIModel(filters_name=[...]))或带 filters_name 的 OpenAIModel/AnthropicModel 模型,在构造时即触发。
实际影响: 所有使用命名 Filter 的工具/模型构造抛 KeyError(<FilterType.UNSUPPORTED: -1>) 而非本意的 ValueError("Filter not found"),任何依赖 filters_name 过滤能力的功能全部失效;test_init_with_filters_name_invalid_raises 断言的是 ValueError,此回归会使其崩溃(异常类型不符),属于框架级兼容性破坏。
修正方向: 不要在 FilterRunner.__init__ 内解析命名过滤器:恢复 BaseTool/LLMModel 在设置 _type 后调用 _init_filters() 的流程,或将 _convert_filters() 的解析延迟到 _type 确认之后的首次使用;若需保留构造期校验,应在 FilterRunner.__init__ 只做记录、由子类在设置 _type 后再解析。
| if event.is_final_response(): | ||
| if self.code_act and self.code_act.require_result: | ||
| codeact_text_only = True | ||
| else: | ||
| self._save_output_to_state(ctx, event) | ||
|
|
||
| # Process code execution responses if code executor is configured. | ||
| # Route model-generated cells to exactly one execution | ||
| # path. CodeAct and user/Skill code executors are separate. | ||
| # We collect code execution events first (this mutates event.content in place, | ||
| # stripping executable_code parts but keeping text/function_call), then yield | ||
| # the main event BEFORE the code execution events so the causal order in | ||
| # session is preserved: assistant declaration → code execution → result. | ||
| pending_code_events: list[Event] = [] | ||
| if self.code_executor and event.content: | ||
| async for code_event in CodeExecutionResponseProcessor.run_async(ctx, event): | ||
| codeact_final_event: Optional[Event] = None | ||
| code_processor = None | ||
| if self.code_act: | ||
| code_processor = CodeActResponseProcessor | ||
| elif self.code_executor: | ||
| code_processor = CodeExecutionResponseProcessor | ||
| if code_processor and event.content: | ||
| async for code_event in code_processor.run_async(ctx, event): | ||
| if code_event.object == "codeact.result": | ||
| codeact_final_event = code_event | ||
| if code_event.content and code_event.content.parts: | ||
| for part in code_event.content.parts: | ||
| if part.code_execution_result or part.executable_code: | ||
| code_was_executed = True | ||
| break | ||
| pending_code_events.append(code_event) | ||
|
|
||
| # Yield the main LLM response event first (now stripped of executable_code | ||
| # but still carrying text and function_call parts). | ||
| # Skip empty events (content became None after all parts were consumed). | ||
| if event.content is not None: | ||
| # CodeAct model output is control code, not a user-facing | ||
| # answer. Its executable-code and observation events are | ||
| # emitted below; only the validated codeact.result is final. | ||
| if event.content is not None and not self.code_act: |
There was a problem hiding this comment.
问题: CodeActConfig(require_result=False) 时,CodeAct 代理对纯文本最终回答不产生任何输出事件。_llm_agent.py:581 的 if event.content is not None and not self.code_act: 使 CodeAct 模式下模型正文事件永远不会被 yield/accumulate(codeact.result/代码事件之外的文本全部丢弃);_llm_agent.py:752 的 text-only 反馈分支要求 require_result 为 True;_llm_agent.py:774 在无代码执行时直接 running = ...False 退出循环。
触发条件: 用户设置 CodeActConfig(require_result=False) 且模型在 max_iterations 内给出纯文本最终答案(未调用 return_result)——例如未遵循 system instruction 的模型或使用不支持 CodeAct 指令的流式调用。
实际影响: Runner 事件流为空:调用方(如 _run_nested_agent、CodeActTool 的 _run_nested_agent/_llm_agent 使用者)收不到任何最终内容,CodeActTool 会抛 "produced no CodeAct result",即使模型已给出答案;output_key 也无法保存输出(_save_output_to_state 分支同样被跳过)。
修正方向: 当 require_result=False 且事件为纯文本最终响应时,应将模型事件(或其文本)以 codeact.result/普通文本事件 yield 出去并 accumulate_content(如非 CodeAct 模式那样),或至少在退出前把该事件标记为最终结果;同时为 require_result=False 的文本-only 场景补一个回归测试(现有 tests/agents/test_codeact.py 无任何 require_result=False 用例)。
| def create_codeact_execution_result( | ||
| *, | ||
| stdout: str = "", | ||
| stderr: str = "", | ||
| ) -> CodeActExecutionResult: | ||
| """Build a model-visible CodeAct execution result.""" | ||
| output = "" | ||
| outcome = Outcome.OUTCOME_OK | ||
| if stderr: | ||
| output = f"Code execution error:\n{stderr}\n" | ||
| outcome = Outcome.OUTCOME_FAILED | ||
| if stdout: | ||
| output += f"Code execution result:\n{stdout}\n" |
There was a problem hiding this comment.
问题: create_codeact_execution_result(_runtime.py:56-58)只要 stderr 非空就把整个结果标记为 outcome = Outcome.OUTCOME_FAILED,即使代码实际成功并打印了 return_result marker。_processor.py:90 只在 OUTCOME_OK 时解析 return_result,导致带非空 stderr 的成功 cell 永远无法产出 codeact.result;_code_act_tool.py:303 的 approved 路径在 outcome != OUTCOME_OK 时直接 raise ValueError("Approved CodeAct implementation ... failed")。
触发条件: 生成/批准的 Python cell 中任何代码向 stderr 写出内容——最典型的是 warnings.warn(...)、第三方库的编译警告/DeprecationWarning、未被捕获的 print(..., file=sys.stderr)——同时 cell 成功执行并调用 return_result 提交了合法结果。
实际影响: 交互式 CodeAct 模式下模型会看到 "Code execution error" 而反复重试直到耗尽 max_iterations 抛 RunLimitException;更严重的是 implementation_policy=prefer_approved/approved_only 下,一旦某次生成的代码包含写 stderr 的语句并被批准,该 approved implementation 将永久失败(每次复用都抛 ValueError),唯一修复是人工回滚/删库。
修正方向: 只在真正未检测到 return_result marker 或 stdout 中出现明显异常时标记 FAILED:例如改为"stderr 非空且无 __TRPC_AGENT_CODEACT_RESULT__ marker 才判 FAILED",或把 stderr 内容当作可展示的 warning 正文追加进 output 而保留 OUTCOME_OK,同时补充一个"成功 return_result + 非空 stderr"的单元测试。
| async def _command(self, method: str, *args: Any, **kwargs: Any) -> Any: | ||
| async with self._storage.create_db_session() as connection: | ||
| return await self._storage.execute_command( | ||
| connection, | ||
| RedisCommand( | ||
| method=method, | ||
| args=args, | ||
| kwargs=kwargs, | ||
| expire=RedisExpire(), | ||
| ), | ||
| ) |
There was a problem hiding this comment.
问题: RedisCodeActImplementationStore._command(code_storage/_redis.py:170-178)对每个写命令硬编码 expire=RedisExpire(),其默认 Ttl(enable=True, ttl_seconds=86400, cleanup_interval_seconds=3600)(types/_ttl.py:17-25)。RedisStorage.execute_command(storage/_redis.py:275-279)对 EXPIRE_METHOD(set/sadd)中的方法自动执行 expire,因此 save_candidate 的版本 key、sadd 的 versions 集合、approve 的 approved 指针全部在写入后 24 小时过期。
触发条件: 使用 RedisCodeActImplementationStore(文档推荐用于多 worker 共享持久化)时,距离上一次写入超过 24 小时后的任意调用:get_approved 返回 None(approved 指针或目标版本 key 已过期);approve 对过期版本抛 KeyError;list_implementations 静默返回空。现有测试 tests/tools/test_code_act_implementation.py:124-163 的 _FakeRedisStorage 完全忽略 expire 字段,因此 CI 无法发现。
实际影响: "版本化持久化实现"在真实 Redis 后端上变成 24 小时缓存:已批准的生产实现被静默丢弃,PREFER_APPROVED/APPROVED_ONLY 策略失效(退化为每次重新生成或直接报错),且无任何日志提示数据丢失;与文档"persist for production"的承诺直接冲突。
修正方向: 对 CodeAct 实现存储使用持久化过期策略:在 _command 中传入 RedisExpire(ttl=Ttl(enable=False, ...))(或显式 ttl_seconds=0),仅在确实需要 TTL 的 key 上启用;并增加用真实 RedisStorage/可识别 expire 的测试替身验证 24h 后数据仍在。
6cc0b25 to
6b10927
Compare
AI Code Review审查结论不通过 审查范围:commit 5fac61f..6b10927「feat: 支持codeact 特性」,共 44 个文件、5983 行新增。核心交付为 CodeAct 执行协议(CodeActConfig/CodeActResponseProcessor/InProcessCodeActRuntime/CodeActObjectStore)、CodeActTool 生成式函数工具及其实现存储(InMemory/File/Redis/SQL)、LlmAgent 集成、filter 注册表重构、SqlStorage.ensure_metadata 与 FileStorage 新增,以及 3 个测试文件与示例文档。计划符合性:CodeAct Tools(self 代理、持久变量、ObjectRef、return_result 协议)与 Generation Method(候选/审批/复用)均已实现且有测试覆盖,行为与文档基本一致。主要风险:(1) CodeAct 模式导致 LlmAgent 在获得最终结果后完全放弃 发现的问题严重
问题: 触发条件: 启用 实际影响: 用户配置的 修正方向: 在 CodeAct 分支中若 严重
问题: 触发条件: 用户启用 实际影响: 用户期望的 修正方向: 将 codeact 迭代纳入框架统一的 中等
问题: 触发条件: 模型生成的 cell 中包含 实际影响: agent 执行崩溃(未被捕获的 BaseException 从 修正方向: 在 中等
问题: 触发条件: 模型或多轮循环中大量生成大型字符串/对象(如每轮 cell 返回大 JSON 字符串、日志文本、或倾向把每个对象都 wrap),对象总量随迭代轮数线性增长,直到进程内存耗尽。 实际影响: ObjectStore 内存无界增长,长时间 CodeAct 任务可导致进程 OOM;同时用户自定义对象被 wrap 后跨代码单元引用, 修正方向: 为 中等
问题: 触发条件: 同一 实际影响: 工具调用可能路由到错误的 invocation agent、 修正方向: 将 中等
问题: 触发条件: 模型生成代码中 print 了包含 实际影响: 最终结果被攻击者/模型注入的伪造值替代, 修正方向: 解析时打印行与 中等
问题: 触发条件: 任何 agent/tool 构造时传入 实际影响: 修正方向: 保持 中等
问题: 触发条件: 实际影响: 异步模式下所有 SQL CodeAct 存储操作在初始化时崩溃,无法建表,功能不可用;同步分支虽然用 修正方向: 异步分支统一改为 中等
问题: 触发条件: 使用 实际影响: 多 Worker/多进程共享 FileStore 时候选保存失败(OSError 在 Windows 上为 修正方向: 对 较低
问题: 触发条件: 任意一次 实际影响: 约每 160 次生成候选中有 1 次 File/Redis store 的 修正方向: 将版本校验改为与生成端一致的契约(如 |
| if codeact_final_event is not None: | ||
| self._save_output_to_state(ctx, codeact_final_event) | ||
| return |
There was a problem hiding this comment.
问题: _llm_agent.py 在 CodeAct 模式下,当处理器产出 codeact.result 事件时,只调用 self._save_output_to_state(ctx, codeact_final_event) 就 return,但 codeact_final_event 的 object 字段已由 CodeActResponseProcessor 设置为 "codeact.result",而 _save_output_to_state 读取的是 event.content.parts 文本,可以保存;但真正的问题在于 CodeAct 模式下普通最终文本事件被完全跳过:if event.content is not None and not self.code_act: yield event —— 在 CodeAct 模式下不产出最终文本事件,且 require_result=True 时 self._save_output_to_state(ctx, event) 也被跳过,只有 require_result=False 的文本事件才保存。
触发条件: 启用 code_act(默认 require_result=True)且 agent 配置了 output_key(或使用 _save_output_to_state 的用户扩展)时运行任何 CodeAct 任务;模型输出被 CodeActResponseProcessor 截断为只含 codeact.result 事件。
实际影响: 用户配置的 output_key 始终不写入 session state;同时 CodeAct 模式下 require_result=True 时 session 中完全没有最终文本输出事件,外部消费者(session 历史、A2A 等)拿不到最终答案,只能看到 codeact.result 对象,破坏既有输出契约。
修正方向: 在 CodeAct 分支中若 codeact_final_event 存在,应在 return 前同时将 codeact.result 文本保存到 ctx.state[output_key](复用 _save_output_to_state 并对 codeact.result 对象事件做一次保存),或保留一个可见的最终文本 Event 再退出循环,保证 output_key 与最终输出事件的行为与普通模式一致。
| codeact_iterations += 1 | ||
| if codeact_iterations > self.code_act.max_iterations: | ||
| raise RunLimitException( | ||
| agent_name=self.name, | ||
| limit_type=RunLimitType.MAX_ITERATIONS, | ||
| configured_value=self.code_act.max_iterations, | ||
| observed_value=codeact_iterations, | ||
| ) | ||
|
|
There was a problem hiding this comment.
问题: _llm_agent.py 新增的 codeact_iterations 计数器独立于框架原有 ctx.raise_if_limit(RunLimitType.MAX_ITERATIONS),且它与 ctx.raise_if_limit 各自独立计数、不同步;同时 ctx.raise_if_limit 的 MAX_ITERATIONS 计数在 agents 通用路径中由 AgentRunLimits 维护,codeact_iterations 不经过它。
触发条件: 用户启用 code_act 并同时配置 RunConfig 中的 max_iterations(作用于框架通用迭代上限)以及 CodeActConfig.max_iterations(作用于 codeact 上限)时,两条路径同时生效;任何一个先到即中止,且 CodeActConfig.max_iterations 的抛出可以绕过框架预设的迭代数限制,导致实际执行轮数比预期少。
实际影响: 用户期望的 max_iterations 语义被双重叠加,CodeAct 任务可能在达到预期轮数(如 10)前被框架通用上限(如 3)提前杀死,或反过来;造成不可预期的行为差异,且 codeact_iterations 计数未计入 _observed_run_limits,导致 RunLimitException 报告的 observed_value 与真实总迭代不一致。
修正方向: 将 codeact 迭代纳入框架统一的 ctx._observed_run_limits 计数与 raise_if_limit(RunLimitType.MAX_ITERATIONS) 机制,或用 CodeActConfig.max_iterations 覆盖/映射到框架通用上限,避免两套独立计数。
| async with self._execution_lock: | ||
| with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): | ||
| for cell in cells: | ||
| await self._execute_cell(cell, state.globals) | ||
| except CodeActReturnSignal: | ||
| pass | ||
| except Exception: # noqa: BLE001 # pylint: disable=broad-except | ||
| stderr.write(traceback.format_exc()) |
There was a problem hiding this comment.
问题: InProcessCodeActRuntime.execute 中 _execute_cell 通过 eval(compile(..., flags=ast.PyCF_ALLOW_TOP_LEVEL_AWAIT), globals_) 执行模型生成的代码,若代码顶层为异步表达式(如 await self.tool() 直接在顶层而非赋值或表达式语句),eval 会返回一个协程对象,随后 if inspect.isawaitable(result): await result 在 async 函数内执行 await;但当 cell 是顶层 await 表达式语句且无赋值时,eval 会把 await 表达式提升为 coroutine 返回值并 await 它,这是安全的;问题在于 _execution_lock 是在 await state.agent_proxy.refresh(...) 之后加锁,而 refresh 解析 agent.tools(可能包含动态 ToolSet 的 get_tools 调用),未持有锁前并发两轮刷新可能竞争;另外当 cell 执行抛出 CodeActReturnSignal 之外的 BaseException(如 KeyboardInterrupt、SystemExit 或用户代码 raise 的 BaseException),不会被 except Exception 捕获,直接传播出 execute,导致 create_codeact_execution_result 永远不被调用、_llm_agent 的 async for code_event in code_processor.run_async(ctx, event) 抛出未处理异常而中断整个 agent。
触发条件: 模型生成的 cell 中包含 raise KeyboardInterrupt / SystemExit / 自定义 BaseException,或在 await 语句中抛出非 Exception 基类异常。
实际影响: agent 执行崩溃(未被捕获的 BaseException 从 CodeActResponseProcessor 传播到 LlmAgent 主循环的 async for,被最外层 except Exception 捕获并生成 agent_execution_failed 错误事件,但这是程序错误处理缺口;BaseException 的子类如 CancelledError 更是直接中断循环)。
修正方向: 在 try/except 中加入 except BaseException 分支(如捕获 CodeActReturnSignal 与 KeyboardInterrupt/SystemExit 的明确转换),并将 _execute_cell 的 await 结果包裹在 if inspect.iscoroutine(result): result = await result 中并在 execute 内捕获 CodeActReturnSignal 与通用 BaseException 转换成 CodeActExecutionResult。
| def wrap(self, value: Any) -> Any: | ||
| """Return small JSON values inline and retain larger values by reference.""" | ||
| if isinstance(value, CodeActObjectRef): | ||
| return value | ||
| if isinstance(value, _PRIMITIVE_TYPES) and (not isinstance(value, str) or len(value) <= self.inline_threshold): | ||
| return value | ||
| if self._can_inline(value): | ||
| return value | ||
|
|
||
| object_id = uuid.uuid4().hex | ||
| self._objects[object_id] = value | ||
| return CodeActObjectRef(self, object_id) |
There was a problem hiding this comment.
问题: CodeActObjectStore.wrap 对所有长度超过 inline_threshold(默认 256)的字符串直接生成 CodeActObjectRef 放入 _objects 字典,且 wrap 未对要保留的对象数量做任何上限控制;InProcessCodeActRuntime 的 _states 字典也只按 invocation_id 保存,不限制每个 state 的对象数量或总内存。
触发条件: 模型或多轮循环中大量生成大型字符串/对象(如每轮 cell 返回大 JSON 字符串、日志文本、或倾向把每个对象都 wrap),对象总量随迭代轮数线性增长,直到进程内存耗尽。
实际影响: ObjectStore 内存无界增长,长时间 CodeAct 任务可导致进程 OOM;同时用户自定义对象被 wrap 后跨代码单元引用,release 前无法回收,加剧泄漏。
修正方向: 为 CodeActObjectStore 增加对象数量/总字节上限(如 max_objects 配置),超限时以描述字符串占位代替引用,或在 release 时主动清理;同时文档说明该限制。
| async def invoke(*args: Any, **kwargs: Any) -> Any: | ||
| if args: | ||
| raise TypeError(f"self.{name} accepts keyword arguments only") | ||
| if tool.is_progress_streaming: | ||
| raise TypeError(f"self.{name} is progress-streaming and is not supported by CodeAct") | ||
| self._context.raise_if_limit(RunLimitType.MAX_TOOL_CALLS) | ||
| previous_call_id = self._context.function_call_id | ||
| self._context.function_call_id = f"codeact-{uuid.uuid4().hex}" | ||
| try: | ||
| result = await tool.run_async( | ||
| tool_context=self._context, | ||
| args={ | ||
| key: self._store.unwrap(value) | ||
| for key, value in kwargs.items() | ||
| }, | ||
| ) | ||
| finally: | ||
| self._context.function_call_id = previous_call_id |
There was a problem hiding this comment.
问题: CodeActAgentProxy.__getattr__ 中 async def invoke(...) 在 self._context.raise_if_limit(RunLimitType.MAX_TOOL_CALLS) 前先执行 tool.is_progress_streaming 检查,但这里没有检查 self._context 是否与当前 invocation 一致;CodeActAgentProxy.refresh 更新 self._context,但 _tools 映射与 self._context 可能来自不同 invocation(InProcessCodeActRuntime 的 refresh(invocation_context) 每轮都重新设置 context),_tools 是 refresh 前解析的,若两个 invocation 并发使用同一 runtime(_execution_lock 串行化但 refresh 在锁外),self.<tool> 可能调用到上一 invocation 的工具且使用错误的 context.function_call_id,进而污染工具调用计数与回调归属。
触发条件: 同一 InProcessCodeActRuntime 实例被多个并发 invocation 共享(文档说明只能在单 worker 使用,但并发场景仍可能从 runner 传入同一配置实例),两个 invocation 交替 execute,refresh 交替改写共享 proxy 的 _context 与 _tools。
实际影响: 工具调用可能路由到错误的 invocation agent、MAX_TOOL_CALLS 限制跨 invocation 串扰,且 function_call_id 互踩(一个 invocation 的工具调用 ID 被另一个覆盖),破坏审计与长作业关联。
修正方向: 将 _ExecutionState(含 proxy 与其 context)与 invocation 绑定(已按 invocation 存储 state),在 execute 加锁内完成 refresh,或在 invoke 内用当前 invocation 的 context 重建工具映射。
| def parse_codeact_result( | ||
| output: str, | ||
| output_schema: type[BaseModel] | None = None, | ||
| ) -> ParsedCodeActResult: | ||
| """Parse the last ``return_result`` payload from runtime output.""" | ||
| matches = re.findall(re.escape(CODEACT_RESULT_PREFIX) + r"([^\r\n]*)", output or "") | ||
| if not matches: | ||
| return ParsedCodeActResult() | ||
|
|
||
| payload = matches[-1] | ||
| try: | ||
| value = json.loads(payload) | ||
| except json.JSONDecodeError as exc: | ||
| return ParsedCodeActResult(validation_error=f"return_result emitted invalid JSON: {exc}") | ||
|
|
||
| if output_schema is not None: | ||
| try: | ||
| validated = output_schema.model_validate(value) | ||
| except ValidationError as exc: |
There was a problem hiding this comment.
问题: _config.py 的 parse_codeact_result 通过 re.findall(re.escape(CODEACT_RESULT_PREFIX) + r"([^\r\n]*)", output) 解析输出,并提取「最后一个」匹配;_CODEACT_PRELUDE 中 return_result 使用 print(prefix + json.dumps(value)) 单行打印。若模型/用户代码主动 print 出包含 prefix 的伪造行(如 print("__TRPC_AGENT_CODEACT_RESULT__:{'fake'}") 或代码本身的调试输出包含该前缀),且该伪造行排在真实 return_result 之后,parse_codeact_result 会取到伪造值;反过来排在之前则优先取真实值。系统协议并未禁止普通 print 输出该前缀。
触发条件: 模型生成代码中 print 了包含 CODEACT_RESULT_PREFIX 的行(例如调试时输出含该前缀的字符串),且它晚于真实 return_result 的 print。
实际影响: 最终结果被攻击者/模型注入的伪造值替代,output_schema 校验可能不通过导致循环重试,或通过 Pydantic 松散校验产生错误结果。
修正方向: 解析时打印行与 return_result 的调用记录绑定(如让 return_result 同时写入一个不在 stdout 的专用标记,或让处理器改为捕获 stdout 中第一个匹配并校验),并在指令中明确禁止输出该前缀。
| def _init_filters(self): | ||
| """Initialize filters.""" | ||
| for filter_name in self._filters_name: | ||
| filter_instance = get_filter(self._type, filter_name) | ||
| if not filter_instance: | ||
| raise ValueError(f"Filter {filter_name} not found, type: {self._type}") | ||
| self._filters.append(filter_instance) | ||
| self._filters = self._convert_filters() | ||
| self._filters_name = [filter.name for filter in self._filters] |
There was a problem hiding this comment.
问题: _filter_runner.py 将 _init_filters 改为调用 _convert_filters() 并把 self._filters_name 重写为 [filter.name for filter in self._filters],从而丢失用户的原始注册名语义:原来 filters_name 保留用户传入的名称字符串,现在等价于「已解析出的过滤器实例名」。BaseAgent.model_post_init 会为每个 agent 通过 self.filters.append(AgentCallbackFilter(...)) 追加回调过滤器,而 _convert_filters 只返回实例列表,filters_name 被改写后 AgentCallbackFilter 等无注册名的实例也会出现在 filters 列表末尾并且每次 _init_filters 重复追加同名实例(_convert_filters 无去重、model_post_init 每次追加);CodeActTool 构造时以 filters=[...] 传入多个过滤器实例,再次初始化时也会被 _init_filters 重新解析成重复实例。
触发条件: 任何 agent/tool 构造时传入 filters 实例(如 CodeActTool(filters=[...]) 或 agent 的 filters),或 BaseAgent.model_post_init 多次调用 _init_filters(pydantic 子类实例化、复用)。
实际影响: filters 列表中的过滤器实例重复或在传入实例与注册名之间不一致,导致 run_stream_filters/filter 链重复执行(如日志、权限检查多次);filters_name 返回的值不再等于用户配置的注册名,破坏依赖该属性的外部代码。
修正方向: 保持 filters_name 为用户原始值,仅用 _convert_filters 返回解析结果;_init_filters 中做去重(按名称或身份判断已存在实例),避免重复追加。
| async def ensure_metadata(self, metadata: MetaData) -> None: | ||
| """Create and forward-migrate another service's tables. | ||
|
|
||
| This allows multiple framework services to share one ``SqlStorage`` | ||
| engine and connection pool while retaining separate SQLAlchemy metadata. | ||
| """ | ||
| await self.create_sql_engine() | ||
| if self._db_engine is None: | ||
| raise RuntimeError("SQL storage engine is not initialized") | ||
| if isinstance(self._db_engine, AsyncEngine): | ||
| async with self._db_engine.begin() as connection: | ||
| await connection.run_sync(metadata.create_all) | ||
| await connection.run_sync(lambda sync_connection: self._migrate_missing_columns( | ||
| sync_connection, | ||
| metadata, |
There was a problem hiding this comment.
问题: SqlStorage.ensure_metadata 在 async 分支中执行 metadata.create_all(self._db_engine),其中 self._db_engine 是 AsyncEngine(is_async=True 时创建),而 MetaData.create_all(bind) 需要同步 Engine;SQLAlchemy 异步引擎(asyncpg / aiomysql / aiosqlite 等)直接对 AsyncEngine 调用 create_all 会引发 InvalidRequestError/不支持操作的异常(AsyncEngine 不是同步 Engine,create_all 会尝试 engine.connect() 拿到同步连接而失败)。
触发条件: SqlCodeActImplementationStore(is_async=True, db_url=...)(或通过 SqlStorage(is_async=True) 注入)首次使用时调用 ensure_metadata。
实际影响: 异步模式下所有 SQL CodeAct 存储操作在初始化时崩溃,无法建表,功能不可用;同步分支虽然用 with self._db_engine.begin() as connection(正确),但异步分支存在同样问题。
修正方向: 异步分支统一改为 await connection.run_sync(metadata.create_all)(在 begin() 的异步连接内执行同步建表),与下方 _migrate_missing_columns 的 run_sync 一致。
|
|
||
| @override | ||
| async def close(self) -> None: | ||
| if self._owns_storage: | ||
| await self._storage.close() | ||
|
|
||
| @staticmethod |
There was a problem hiding this comment.
问题: CodeActImplementation.create 生成的 version = code_hash[:16],而 _version_key(File/Redis store)用 re.fullmatch(r"[0-9a-f]{16}", version) 校验;sha256 十六进制前 16 个字符恰好全部为数字(prob≈(10/16)^16≈6.16e-3)时,版本号形如 0123456789abcdef 或 1234567890abcdef,正则 [0-9a-f] 实际接受数字与 a-f,但纯数字字符串也通过;真正的失败场景是 version 中所有字符都是数字(如 0123456789012345),此时 re.fullmatch(r"[0-9a-f]{16}") 匹配失败,抛 ValueError,导致 save_candidate/approve 直接崩溃。
触发条件: 任意一次 CodeActImplementation.create 的 code_hash 前 16 字符全为数字(概率≈0.62%);或用户手动调用 approve(version=...) 传入非法版本号。
实际影响: 约每 160 次生成候选中有 1 次 File/Redis store 的 save_candidate 抛 ValueError,Approval 流程失败,生成代码无法落盘;SQL store 不做该校验(无正则),文件与 SQL 行为不一致。
修正方向: 将版本校验改为与生成端一致的契约(如 version 固定 16 位十六进制但允许纯数字;或改用 re.fullmatch(r"[0-9a-fA-F]{16}") 并确保 create 端使用小写十六进制),或直接移除正则校验、以内容寻址保证唯一性。
3edcb09 to
ee71283
Compare
AI Code Review审查结论不通过 审查范围:base_commit 5fac61f 至 head_commit 6cc0b25(feat: 支持codeact特性),44 个文件,+6056/-25 行,含 3 个测试文件。计划符合性:CodeAct 配置/运行时/对象代理/响应处理器/CodeActTool/四种实现存储/示例与文档均已实现并配套测试。主要风险集中在 LlmAgent 循环新增的 CodeAct 控制流:(1) CodeAct 模式下模型主事件永不 yield 且 codeact.result 后直接 return,模型最终文本与 require_result=False 的纯文本响应被整体吞掉,用户无任何自然语言输出(SEVERE);(2) CodeAct 模式下传统 function_call 被永久拒绝并计入 codeact_iterations,对需要 LLM 型工具(如 LLM 即工具)的既有用户构成回归——最终以 RunLimitException 终止整个运行(SEVERE);(3) CodeActObjectStore.wrap 对任意对象全量 json.dumps 序列化以测大小,真实路径上大对象无界物化(MODERATE);(4) require_result=True 时 is_final_response 事件不写 output_key,文本路径状态丢失(LOW);(5) SqlCodeActImplementationStore.approve 并发 IntegrityError 重试窗口在 approval 行尚未提交时会抛 RuntimeError,无测试覆盖(LOW)。测试充分性:新增测试覆盖 result 解析、对象引用、工具调用、approved 复用、存储生命周期等 happy path,但均未覆盖 require_result=False 空响应、function-call 拒绝循环、text-only 重试边界、混合响应、并发存储冲突等失败路径。全部 44 个非生成变更文件均已审查。 发现的问题严重
问题: CodeAct 模式下 触发条件: 任何 CodeAct 模式调用(CodeActConfig 默认 require_result=True);使用 require_result=False 且模型直接以文本回答(未写 python cell)时必然触发空响应;CodeActTool 嵌套 agent 中模型在 return_result 外的文本总结同样被抑制。 实际影响: 用户在 CodeAct 模式下只能收到 修正方向: CodeAct 终态后将最后的模型文本事件(或 codeact.result 事件的序列化值)作为 严重
问题: 本变更在 CHECKPOINT 4 后新增的拒绝分支( 触发条件: LlmAgent 配置 实际影响: 用户的既有工具调用功能在 CodeAct 模式下直接回归——每次调用在浪费 max_iterations 轮次后以 RunLimitException 失败,而不是执行工具,且 修正方向: 在拒绝分支前识别这类必须走 function_call 的工具(如检查工具是否为 LLM 驱动类型),对它们保留工具执行路径,或在校验器( 中等
问题: 触发条件: CodeAct 模式下模型调用工具或访问对象引用,其值超过 inline_threshold 且为 JSON 可序列化类型(list/dict/大字符串场景),属于模型反复探索数据集的常规路径。 实际影响: 大对象场景下每次访问都产生数倍于对象本身的内存峰值(对象 + 完整 JSON 字符串同时存活),可导致 OOM 或明显延迟,与 ObjectRef 设计“大对象不进入模型消息、仅保存本地”的初衷相悖。 修正方向: 用有界采样判断是否 inline(例如对 list/dict 先检查 较低
问题: 本变更将 触发条件: LlmAgent 配置 实际影响: 依赖 修正方向: 对 require_result=True 的文本路径也调用 较低
问题: 触发条件: 多 worker/多进程共享同一 SqlStorage 且同时对同一 (tool_name, contract_hash) 首次 approve,时序落在胜者提交前。 实际影响: 并发部署下的 CodeAct 实现审批偶尔返回 RuntimeError,调用方(CodeActTool.approve)将其作为异常传播,用户审批失败;该路径亦无测试锁定行为。 修正方向: 在 |
| if event.content is not None and not self.code_act: | ||
| yield event | ||
| accumulate_content(event) | ||
|
|
||
| # Then yield code execution events in order. | ||
| for code_event in pending_code_events: | ||
| yield code_event | ||
|
|
||
| if codeact_final_event is not None: | ||
| self._save_output_to_state(ctx, codeact_final_event) | ||
| return |
There was a problem hiding this comment.
问题: CodeAct 模式下 _llm_agent.py 循环对模型主事件添加了 not self.code_act 条件(本变更引入),CodeAct 模式下模型的事件既不 yield 也不 accumulate_content;随后 codeact_final_event 出现时(line 589-591)直接 _save_output_to_state 并 return,跳过 CHECKPOINT 6 的 text-only 重试与 Step 4 的总结循环。因此模型在 cell 之外的最终自然语言文本永远不会输出或入库,require_result=False 时纯文本响应(无代码 cell)的事件既不 yield 也不保存,用户拿不到任何最终事件。
触发条件: 任何 CodeAct 模式调用(CodeActConfig 默认 require_result=True);使用 require_result=False 且模型直接以文本回答(未写 python cell)时必然触发空响应;CodeActTool 嵌套 agent 中模型在 return_result 外的文本总结同样被抑制。
实际影响: 用户在 CodeAct 模式下只能收到 codeact.result 的 JSON,永远收不到模型自然语言总结;require_result=False 时模型文本回答被整体吞掉,用户得到空结果,session 中也无最终模型事件。
修正方向: CodeAct 终态后将最后的模型文本事件(或 codeact.result 事件的序列化值)作为 model 角色事件 yield 并 accumulate;require_result=False 的纯文本路径应走原有的 yield event; accumulate_content(event) 分支,而不是被 not self.code_act 条件屏蔽。
| if self.code_act and collected_tool_calls: | ||
| feedback_event = Event( | ||
| invocation_id=ctx.invocation_id, | ||
| author=self.name, | ||
| branch=ctx.branch, | ||
| content=Content( | ||
| role="user", | ||
| parts=[ | ||
| Part( | ||
| text=( | ||
| "Traditional function calls are disabled in CodeAct mode. " | ||
| "Use a Python cell and call capabilities through await self.<tool>(...)." | ||
| )) | ||
| ], | ||
| ), | ||
| visible=False, | ||
| object="codeact.feedback", | ||
| ) | ||
| yield feedback_event | ||
| accumulate_content(feedback_event) | ||
| logger.debug("CodeAct rejected %s traditional tool calls", len(collected_tool_calls)) | ||
| continue |
There was a problem hiding this comment.
问题: 本变更在 CHECKPOINT 4 后新增的拒绝分支(if self.code_act and collected_tool_calls:)对 CodeAct 模式下模型发出的传统 function_call 一律注入 codeact.feedback 并 continue,没有任何降级或替代路径。由于 _request_processor._add_tools_to_request 在 code_act 存在时直接 return None(不暴露工具 schema),但对依赖 LLM 推理的工具(如 LLM 即工具类 ToolSet,其调用本质是模型发 function_call)用户而言,模型在 CodeAct 指令下仍可能按任务语义发出 function_call;该分支 continue 后每轮都重新计数 codeact_iterations,命中 max_iterations 时抛 RunLimitException 终止整个 agent 运行。
触发条件: LlmAgent 配置 code_act 且 tools 含 LLM 型/需 function_call 驱动的工具时(例如既有的 LlmModelOnTool 类工具),模型在 CodeAct 模式下按任务要求发出 function_call。
实际影响: 用户的既有工具调用功能在 CodeAct 模式下直接回归——每次调用在浪费 max_iterations 轮次后以 RunLimitException 失败,而不是执行工具,且 codeact.feedback 无法指导模型绕过这些工具(它们只能用 function_call 调用)。
修正方向: 在拒绝分支前识别这类必须走 function_call 的工具(如检查工具是否为 LLM 驱动类型),对它们保留工具执行路径,或在校验器(_validate_code_act_configuration)中明确拒绝此配置组合并在文档中说明;至少应在反馈消息中告知模型该工具不可用并避免计入迭代上限。
| def _can_inline(self, value: Any) -> bool: | ||
| try: | ||
| rendered = json.dumps(value, ensure_ascii=False, default=self._reject_json_default) | ||
| except (TypeError, ValueError): | ||
| return False | ||
| return len(rendered) <= self.inline_threshold |
There was a problem hiding this comment.
问题: CodeActObjectStore.wrap 中 _can_inline 对每个非基础类型对象执行完整 json.dumps(value) 序列化来评估大小,仅为了与 inline_threshold(默认 256 字节)比较。工具返回的大对象(如数据表格、长文档)在每次 wrap——包括 CodeActAgentProxy.invoke 返回、CodeActObjectRef.__getitem__/__iter__/__getattr__ 每次访问——都会被整体物化为 JSON 字符串;MemoryError 不在 (TypeError, ValueError) 捕获范围内。
触发条件: CodeAct 模式下模型调用工具或访问对象引用,其值超过 inline_threshold 且为 JSON 可序列化类型(list/dict/大字符串场景),属于模型反复探索数据集的常规路径。
实际影响: 大对象场景下每次访问都产生数倍于对象本身的内存峰值(对象 + 完整 JSON 字符串同时存活),可导致 OOM 或明显延迟,与 ObjectRef 设计“大对象不进入模型消息、仅保存本地”的初衷相悖。
修正方向: 用有界采样判断是否 inline(例如对 list/dict 先检查 len,再对前 N 个元素做小规模 JSON 编码),或直接对超过阈值的非原始类型一律走 ObjectRef,避免全量序列化。
| if event.is_final_response(): | ||
| if self.code_act and self.code_act.require_result: | ||
| codeact_text_only = True | ||
| else: | ||
| self._save_output_to_state(ctx, event) |
There was a problem hiding this comment.
问题: 本变更将 is_final_response 分支改为 if self.code_act and self.code_act.require_result: codeact_text_only = True else: self._save_output_to_state(ctx, event)。默认 require_result=True 时,模型最终文本事件不再写入 output_key 会话状态;而 CodeAct 模式下该事件中的文本事件本身又被前述 SEVERE 问题所述根本不 yield 文本事件,因此 output_key 用户在 CodeAct 模式(尤其 require_result=True + 模型先文本后代码的路径)下丢失中间文本输出,最终仅有 codeact.result 的 JSON 被写入。
触发条件: LlmAgent 配置 output_key 且启用 code_act(默认 require_result=True),模型最终响应不含代码 cell 或含文本部分;require_result=False 时同样不写状态。
实际影响: 依赖 output_key 读取 agent 输出的调用方在 CodeAct 模式下取不到(或只能取到 JSON 串)最终结果,状态值缺失或不符合预期。
修正方向: 对 require_result=True 的文本路径也调用 _save_output_to_state(如原逻辑保持对所有 is_final_response 事件写状态),或让 CodeAct 终态事件统一携带模型最终文本并写入 output_key。
| stdout = io.StringIO() | ||
| stderr = io.StringIO() | ||
| try: | ||
| async with self._execution_lock: | ||
| with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): | ||
| for cell in cells: | ||
| await self._execute_cell(cell, state.globals) | ||
| except CodeActReturnSignal: | ||
| pass | ||
| except Exception: # noqa: BLE001 # pylint: disable=broad-except | ||
| stderr.write(traceback.format_exc()) | ||
|
|
There was a problem hiding this comment.
问题: InProcessCodeActRuntime.execute(_in_process_runtime.py:93-97、145-147)在宿主进程事件循环中直接同步执行模型生成的 Python 单元格(eval(compiled, globals_)),未使用 asyncio.to_thread/worker 进程隔离。这是默认运行时(CodeActConfig().runtime)且文档明确警告"非沙箱,与宿主同权限",因此模型输出可达成任意代码执行;同时同步 eval 阻塞事件循环,使整进程无法响应取消(CancelledError 得不到调度)、无法处理其他请求,time.sleep/while True/重计算可永久挂死整个 Agent 宿主。触发条件: 任何使用默认 CodeActConfig (或 CodeActTool 的生成模式) 的 Agent 收到包含 python 单元格的模型响应——即默认打开功能下任意一次正常调用,均可由模型或注入的 prompt 触发。实际影响: 整进程 DoS/不可取消、任意代码执行(同进程文件、网络、凭据权限);模型的每个单元格都在宿主事件循环内串行运行,期间 runner 的 raise_if_cancelled() 检查点无法执行。修正方向: 提供基于子进程/子解释器(或 asyncio.to_thread + 取消标识)的隔离运行时作为默认,或在文档/配置中明确强制生产环境必须注入沙箱运行时,并对单元格执行增加超时与取消传播(如 asyncio.wait_for + 进程级 kill);同时暴露一个 0 权限代理运行时供无代码 Agent 使用。
| "exec", | ||
| flags=ast.PyCF_ALLOW_TOP_LEVEL_AWAIT, | ||
| dont_inherit=True, | ||
| ) | ||
| result = eval(compiled, globals_) # pylint: disable=eval-used | ||
| if inspect.isawaitable(result): | ||
| await result |
There was a problem hiding this comment.
问题: CodeAct 模型生成代码通过 InProcessCodeActRuntime.execute(_in_process_runtime.py:145 的 eval(compiled, globals_),编译、执行与 __builtins__ 均来自宿主进程)直接获得条件满足时的完整宿主权限:__builtins__ 提供 open/os/import 的完整操作面,且生成的代码可访问 self 代理下所有已配置工具的同步方法(CodeActAgentProxy.__getattr__),无任何白名单、无文件系统/网络限制。文档虽在类 docstring 声明"非沙箱",但这是默认出厂配置,使"不开箱即用即安全"成为默认缺陷。触发条件: 默认配置(CodeActConfig() 不传 runtime)下,模型在响应中输出任意 python 单元格——竞态条件:模型输出本身不可信(prompt 注入、指令冲突),且示例和测试均默认使用此配置。实际影响: 任意代码执行:读取/删除宿主文件(open/shutil/os)、外发网络请求(urllib/requests/socket,构成 SSRF)、访问环境变量与进程凭据,并可将结果经 print 回传给模型进而泄露给客户端;与工具集合并用时还可直接调用 self.<tool> 访问业务能力。修正方向: 将该运行时明确标记为仅限开发/受信模型使用,生产默认改为提供 BaseCodeActRuntime 的沙箱实现(受限解释器/容器),在 CodeActConfig 中增加安全默认(如限制 __builtins__、禁止 __import__/open/网络的原语级选项),并在 AGENTS/docs 中给出强制隔离警告;同时让 CodeActTool/示例默认使用沙箱运行时而非宿主解释器。
| codeact_iterations += 1 | ||
| if codeact_iterations > self.code_act.max_iterations: | ||
| raise RunLimitException( | ||
| agent_name=self.name, | ||
| limit_type=RunLimitType.MAX_ITERATIONS, | ||
| configured_value=self.code_act.max_iterations, | ||
| observed_value=codeact_iterations, | ||
| ) |
There was a problem hiding this comment.
问题: LlmAgent._run_async_impl 中新增的 CodeAct 分支在 codeact_iterations > self.code_act.max_iterations 时抛出的 RunLimitException 会沿用户级 except (RunCancelledException, RunLimitException) 重新抛出,而 finally: 中的 self.code_act.runtime.release(ctx.invocation_id) 虽然执行,但 release 仅弹出状态;ctx.raise_if_limit(RunLimitType.MAX_ITERATIONS)(第 491 行)与新增的 codeact 迭代计数(第 493-500 行)是两个独立计数,前者按 ctx._observed_run_limits 累计,后者不重置——但真正的缺陷在于:当 run 因 codeact 轮次达到 max_iterations 抛出 RunLimitException 后,Runner 的 except RunLimitException(runners.py:630)会执行 update_session 并重新抛出,导致用户看到的是未经 codeact.result 事件包装的异常——即模型已经通过 return_result 提交的最终结果可能丢失。触发条件: 模型在最后一轮(恰好在 max_iterations 边界)调用 return_result 产生 codeact.result 事件后的下一次循环检查——例如单元格执行自身需要 2 轮而 max_iterations=1,或模型先输出文本(触发 text_only 反馈占掉轮次)再输出代码。实际影响: 已通过 return_result 提交的合法结果被 RunLimitException 吞噬,调用方(Runner/API)收到的不是结果而是异常;且 session_service.update_session 已把状态写入,用户端出现"事件流有 codeact.result 但 run 失败"的分裂状态。修正方向: 在迭代计数达到上限时,先检查本轮是否已出现 codeact_final_event(即是否已有合法结果),有则直接 return 提交结果而不抛异常;或将 max_iterations 语义与 MAX_ITERATIONS 统一为同一计数源。
| output_schema: type[BaseModel] | None = None, | ||
| ) -> ParsedCodeActResult: | ||
| """Parse the last ``return_result`` payload from runtime output.""" | ||
| matches = re.findall(re.escape(CODEACT_RESULT_PREFIX) + r"([^\r\n]*)", output or "") | ||
| if not matches: | ||
| return ParsedCodeActResult() | ||
|
|
||
| payload = matches[-1] | ||
| try: | ||
| value = json.loads(payload) | ||
| except json.JSONDecodeError as exc: | ||
| return ParsedCodeActResult(validation_error=f"return_result emitted invalid JSON: {exc}") |
There was a problem hiding this comment.
问题: parse_codeact_result(_config.py:134-147)对模型输出做 re.findall(...) 后取 matches[-1](最后一段 CODEACT_RESULT_PREFIX 后的文本),json.loads 失败时仅形成 validation_error 而不产生 final 事件;同时该解析把标准输出任意模型可打印文本中间的 prefix 段也纳入——例如模型先执行 print(__TRPC_AGENT_CODEACT_RESULT__临时的错误内容) 模拟前缀,或 return_result 正常输出后又执行了打印。关键缺陷:若模型在一次单元内调用了 return_result(触发 CodeActReturnSignal 中断单元),前置单元格中恰好包含 print('__TRPC_AGENT_CODEACT_RESULT__:xxx') 的调试输出……实际触发来自 InProcessCodeActRuntime:单元格 A 打印 payload 前缀的假数据,单元格 B 调用真正的 return_result,matches[-1] 取到最后一段真实 payload——本路径正确;但若真实 return_result 的 payload 含换行(json.dumps 默认不转义换行)?separators 未指定 ensure_ascii=False 之外换行——json.dumps(value, ensure_ascii=False, separators=(",", ":")) 对 value 中的 会转义为 \n,但 \n 在正则 [^\r\n]* 中被吞掉?不会——转义后是字面反斜杠 n,无换行。但若模型调用 return_result 两次(第一次失败值、第二次修正值),matches[-1] 取修正值正确。真正的边界缺陷:payload 内含 或 字面字符时(如 value 为多行字符串 {"text": "line1\nline2"},json.dumps 会输出 \n 转义,无问题)。所以该函数实际缺陷收敛为:return_result 的值含字面换行被 json.dumps 转义为 \n(正确);但 parse_codeact_result 用 re.findall 匹配 [^\r\n]* 无法处理反斜杠+字面 n?反斜杠 n 是两个字符,不匹配 ^\r\n,可匹配——正确。综上 parse_codeact_result 本身对换行安全;真正问题是它只取最后一个匹配:模型在错误尝试后于同一输出中再次 return_result,前一 payload 被忽略且无验证——但这是设计("Call return_result exactly once")。因此本问题降级为:validation_error 分支在 require_result=True 下被模型视为可无限重试的通道,且无任何迭代计数约束(与 max_iterations 无关),模型可产生大量 validation_error 事件填满会话历史/上下文窗口,直至 MAX_ITERATIONS(默认 10)触发 RunLimitException——但 MAX_ITERATIONS 只约束 LLM 轮数,validation_error 走 continue 后每次仍是新 LLM 调用,所以实际仍受 MAX_LLM_CALLS(默认 10)约束;影响为:验证失败的错误文本(含完整 JSON Schema)被注入会话历史,累积 token 消耗。修正方向: 对 validation_error 的反馈文本做长度截断(如仅保留路径级报错),并确保 code_act.max_iterations 同时作为 validation 重试上限。
| if self.code_act and not event.partial and event.usage_metadata: | ||
| codeact_usage_metadata = event.usage_metadata | ||
|
|
||
| # Skip streaming tool calls (partial=True with streaming_tool_call metadata) | ||
| # These events are yielded directly for consumers to handle | ||
| if event.is_streaming_tool_call(): | ||
| pass | ||
| else: | ||
| function_calls = event.get_function_calls() | ||
| if function_calls: | ||
| collected_tool_calls.extend(function_calls) | ||
| logger.debug("Collected %s tool calls from LLM", len(function_calls)) | ||
|
|
||
| if event.is_final_response(): | ||
| self._save_output_to_state(ctx, event) | ||
| if self.code_act and self.code_act.require_result: | ||
| codeact_text_only = True | ||
| else: | ||
| self._save_output_to_state(ctx, event) | ||
|
|
||
| # Process code execution responses if code executor is configured. | ||
| # Route model-generated cells to exactly one execution | ||
| # path. CodeAct and user/Skill code executors are separate. | ||
| # We collect code execution events first (this mutates event.content in place, | ||
| # stripping executable_code parts but keeping text/function_call), then yield | ||
| # the main event BEFORE the code execution events so the causal order in | ||
| # session is preserved: assistant declaration → code execution → result. | ||
| pending_code_events: list[Event] = [] | ||
| if self.code_executor and event.content: | ||
| async for code_event in CodeExecutionResponseProcessor.run_async(ctx, event): | ||
| codeact_final_event: Optional[Event] = None | ||
| code_processor = None | ||
| if self.code_act: | ||
| code_processor = CodeActResponseProcessor | ||
| elif self.code_executor: | ||
| code_processor = CodeExecutionResponseProcessor | ||
| if code_processor and event.content: | ||
| async for code_event in code_processor.run_async(ctx, event): | ||
| if code_event.object == "codeact.result": | ||
| codeact_final_event = code_event | ||
| if code_event.content and code_event.content.parts: | ||
| for part in code_event.content.parts: | ||
| if part.code_execution_result or part.executable_code: | ||
| code_was_executed = True | ||
| break | ||
| pending_code_events.append(code_event) | ||
|
|
||
| # Yield the main LLM response event first (now stripped of executable_code | ||
| # but still carrying text and function_call parts). | ||
| # Skip empty events (content became None after all parts were consumed). | ||
| if event.content is not None: | ||
| # CodeAct model output is control code, not a user-facing | ||
| # answer. Its executable-code and observation events are | ||
| # emitted below; only the validated codeact.result is final. | ||
| allow_codeact_text = bool(self.code_act and not self.code_act.require_result | ||
| and event.is_final_response() and not pending_code_events) | ||
| if event.content is not None and (not self.code_act or allow_codeact_text): | ||
| yield event | ||
| accumulate_content(event) | ||
|
|
||
| # Then yield code execution events in order. | ||
| if pending_code_events and event.usage_metadata: | ||
| pending_code_events[-1].usage_metadata = event.usage_metadata | ||
| codeact_usage_metadata = None |
There was a problem hiding this comment.
问题: LlmAgent 第 542-543 行只在 not event.partial and event.usage_metadata 时记录 codeact_usage_metadata;而当模型响应含函数调用(CodeAct 模式不执行函数调用,走 collected_tool_calls 反馈分支)时,event.usage_metadata 被丢弃——feedback_event(第 614-625 行)以 usage_metadata=None 创建,使计数(prompt/completion tokens)不进入会话状态跟踪;同时第 595-597 行把本轮 LLM 响应的 usage 移到 pending_code_events[-1],但该事件的 usage_metadata 在 yield code_event 后才被 Runner 记录,导致 usage 归到 code-execution 事件而非模型响应事件,可能出现双计/漏计。触发条件: 任何 CodeAct 模式下模型先返回函数调用(被拒绝)后的回合,或文本+代码混合回合。实际影响: 会话/追踪中的 token 统计缺失或错位,影响成本核算和观测;对最终用户为数据准确性问题。修正方向: 将 codeact_usage_metadata 无条件记录(不依赖 not event.partial 之外的条件),并确保 usage 只挂到模型响应事件本身而非后续转换事件上。
| async def approve( | ||
| self, | ||
| tool_name: str, | ||
| contract_hash: str, | ||
| version: str, | ||
| *, | ||
| approved_by: str | None = None, | ||
| ) -> CodeActImplementation: | ||
| await self._ensure_initialized() | ||
| tool_hash = self._tool_hash(tool_name) | ||
| async with self._storage.create_db_session() as db: | ||
| implementation = await self._storage.get( | ||
| db, | ||
| SqlKey( | ||
| key=(tool_hash, contract_hash, version), | ||
| storage_cls=_CodeActImplementationRow, | ||
| ), | ||
| ) | ||
| if implementation is None: | ||
| raise KeyError(f"Unknown CodeAct implementation version {version!r}") | ||
| approval_key = SqlKey( | ||
| key=(tool_hash, contract_hash), | ||
| storage_cls=_CodeActApprovalRow, | ||
| ) | ||
| approval = await self._storage.get_for_update(db, approval_key) | ||
| now = datetime.now(timezone.utc).replace(tzinfo=None) | ||
| if approval is None: | ||
| approval = _CodeActApprovalRow( | ||
| tool_hash=tool_hash, | ||
| contract_hash=contract_hash, | ||
| version=version, | ||
| approved_at=now, | ||
| approved_by=approved_by, | ||
| ) | ||
| await self._storage.add(db, approval) | ||
| else: | ||
| approval.version = version | ||
| approval.approved_at = now | ||
| approval.approved_by = approved_by | ||
| try: | ||
| await self._storage.commit(db) | ||
| except IntegrityError: | ||
| # Two workers can create the first approval row concurrently. | ||
| # Retry as an update after the winning transaction commits. | ||
| pass | ||
| else: | ||
| return CodeActImplementation.model_validate_json(implementation.payload) | ||
| return await self._retry_approval_update( | ||
| tool_hash=tool_hash, | ||
| contract_hash=contract_hash, | ||
| version=version, | ||
| approved_by=approved_by, | ||
| ) |
There was a problem hiding this comment.
问题: SqlCodeActImplementationStore.approve(_sql.py:181-233)在 IntegrityError 竞态分支中 pass 后直接返回 implementation.payload 快照,没有验证当前批准版本是否就是自己写入的版本:两个 worker 并发首次 approve 不同版本时,败者只是把胜者的提交当成"自己已批准"返回,调用方(CodeActTool.approve 的 rollback 语义)会收到与实际批准版本不一致的状态;随后 _retry_approval_update 仅在 IntegrityError 分支执行——但若直接 commit 成功(未发生 IntegrityError),则行级更新语义正确;真正问题是 approve 的并发窗口下get_for_update+commit 之外无版本校验,回滚(approve 更旧版本)时若两个 worker 并发,旧版本可能被新版本覆盖后不报错。触发条件: 多副本/多进程同时 approve 同一 contract 的不同 version(或同时首次 approve)。实际影响: 批准的版本与实际持久化的版本不一致,后续 get_approved 读到的实现可能不是调用方预期的版本,使 rollback/升级语义失效;若实现内容与预期不符,已批准代码被执行(_run_approved_implementation),相当于静默执行了非预期的模型生成代码。修正方向: 在 IntegrityError 分支重新读取当前批准行并校验 version 与请求一致,不一致则抛出冲突错误;或在批准行上加 version 唯一约束,使并发竞争显式失败。
| async def ensure_metadata(self, metadata: MetaData) -> None: | ||
| """Create and forward-migrate another service's tables. | ||
|
|
||
| This allows multiple framework services to share one ``SqlStorage`` | ||
| engine and connection pool while retaining separate SQLAlchemy metadata. | ||
| """ | ||
| await self.create_sql_engine() | ||
| if self._db_engine is None: | ||
| raise RuntimeError("SQL storage engine is not initialized") | ||
| if isinstance(self._db_engine, AsyncEngine): | ||
| async with self._db_engine.begin() as connection: | ||
| await connection.run_sync(metadata.create_all) | ||
| await connection.run_sync(lambda sync_connection: self._migrate_missing_columns( | ||
| sync_connection, | ||
| metadata, | ||
| )) | ||
| return | ||
| metadata.create_all(self._db_engine) | ||
| with self._db_engine.begin() as connection: | ||
| self._migrate_missing_columns(connection, metadata) |
There was a problem hiding this comment.
问题: SqlStorage.ensure_metadata(_sql.py:255-277)打破了 metadata.create_all 的原子语义:create_all 已建表但 _migrate_missing_columns(ALTER TABLE ADD COLUMN)在并发下直接 begin() 执行,且 ensure_metadata 的调用方 SqlCodeActImplementationStore._ensure_initialized 用进程内 asyncio.Lock 防并发——但多进程/多副本下(该 SDK 常用于多 worker 服务),两个进程同时 create_all + 迁移:create_all 内部对已存在的表跳过(安全),但 _migrate_missing_columns 对同一表并发 ALTER TABLE ADD COLUMN,在 MySQL/PostgreSQL 上后发者报 duplicate column name 导致整个 ensure_metadata 失败,store.save_candidate 随即失败;且 ensure_metadata 与 _migrate_missing_columns 是两个事务(不同 begin()),中途失败留下半迁移状态。触发条件: 多副本首次启动(表不存在)时并发初始化同一 SqlStorage。实际影响: 服务启动竞态失败、初始化异常被抛给首个调用方(候选/审批写失败),或半迁移状态导致后续列缺失查询报错;典型"偶发启动失败"。修正方向: 将 create_all 与列迁移放入同一事务(同一 begin() 内执行 run_sync 全部变更),并对 duplicate column 这一特定 IntegrityError 做幂等重试(重查列名集合后跳过)。
| def _extract_first_cell_and_truncate_content(content: Content) -> list[str]: | ||
| """Extract one Python cell and discard speculative trailing content.""" | ||
| for index, part in enumerate(content.parts or []): | ||
| if part.executable_code and not getattr(part, "thought", False): | ||
| language = (part.executable_code.language or "python").lower() | ||
| if language in ("python", "py", "python3"): | ||
| content.parts = [ | ||
| preceding for preceding in content.parts[:index] | ||
| if preceding.text and not getattr(preceding, "thought", False) | ||
| ] | ||
| return [part.executable_code.code or ""] | ||
|
|
||
| text_parts = [part for part in content.parts or [] if part.text and not getattr(part, "thought", False)] | ||
| if not text_parts: |
There was a problem hiding this comment.
问题: CodeActResponseProcessor.run_async(_processor.py:133-146)的 _extract_first_cell_and_truncate_content 在从 executable_code 类型的 part 提取单元格时,会把该 part 之前的部分全部丢弃(content.parts = [preceding ... if preceding.text ...]),只保留文本部分;而 _retain_non_code_parts 则把 function_call parts 保留回 llm_response.content。结果:CodeAct 模式下模型输出中先于代码单元格出现的 function_call(传统函数调用)被 _retain_non_code_parts 保留到 llm_response.content,随后 _llm_agent.py 的 collected_tool_calls 收集逻辑(event.get_function_calls())在 event.content 上读取的是原始事件(未经过处理器截断),因此这些 function_call 仍会被收集并进入 CODEACT_FUNCTION_CALL_RETRY_MESSAGE 反馈分支——语义上一致但产生一次多余的 LLM 轮次。触发条件: 模型在同一响应中先输出工具调用、再输出 python 单元格(CodeAct 模式下模型混淆)。实际影响: 会产生一次无谓的拒绝-重试轮次(消耗一次 LLM 调用额度),但不至于出错;对严格 require_result 场景会浪费一次迭代。修正方向: 在 CodeAct 模式下提前丢弃 function_call parts(_extract_first_cell_and_truncate_content 后不再 _retain_non_code_parts 保留它们),或让 collected_tool_calls 从截断后的内容读取。
| allow_codeact_text = bool(self.code_act and not self.code_act.require_result | ||
| and event.is_final_response() and not pending_code_events) | ||
| if event.content is not None and (not self.code_act or allow_codeact_text): | ||
| yield event | ||
| accumulate_content(event) | ||
|
|
||
| # Then yield code execution events in order. | ||
| if pending_code_events and event.usage_metadata: | ||
| pending_code_events[-1].usage_metadata = event.usage_metadata | ||
| codeact_usage_metadata = None | ||
| for code_event in pending_code_events: | ||
| yield code_event | ||
| accumulate_content(code_event) | ||
|
|
||
| if codeact_final_event is not None: | ||
| self._save_output_to_state(ctx, codeact_final_event) |
There was a problem hiding this comment.
问题: LlmAgent 文本型 CodeAct 结果的保存路径:allow_codeact_text = ... not pending_code_events(即模型本轮无代码时),event 被 yield 前,若 require_result=False,_save_output_to_state(ctx, event)(第 559 行)在 is_final_response() 分支执行——但 allow_codeact_text 为 True 时 event.content 可能是 Part(function_call=...) 与文本混合(event.content.text 部分为模型文本),_save_output_to_state 拼接 part.text 正确;真正的边界:require_result=False 且模型既输出文本又输出代码(如"我先计算:python..."),此时 pending_code_events 非空、allow_codeact_text=False,文本不被保存;随后 codeact.result 未产生、code_was_executed=True 进入 continue——模型文本(用户可能视为回答)在多轮中丢失,仅代码观察事件被积累;若模型不再输出纯文本直接返回,用户得到空结果。触发条件: require_result=False(文本模式)下模型在单条消息中同时给文本和代码。实际影响: 用户可见回答丢失,状态输出缺失;可观测性下降。修正方向: 在 allow_codeact_text 计算中同时保留代码前的文本部分(将 _extract_first_cell_and_truncate_content 截断后的 text_before 也作为内容保存)。
| "user_content": | ||
| Content( | ||
| role="user", | ||
| parts=[ | ||
| Part.from_text(text=self._input_model.model_validate(arguments).model_dump_json( | ||
| exclude_none=True)) | ||
| ], | ||
| ), | ||
| }) | ||
| runtime = code_act.runtime | ||
| try: | ||
| execution = await runtime.execute( | ||
| execution_context, | ||
| [prepare_codeact_code(cell) for cell in approved.code_cells], | ||
| ) | ||
| finally: |
There was a problem hiding this comment.
问题: 测试文件 tests/tools/test_code_act_tool.py、tests/agents/test_codeact.py、tests/tools/test_code_act_implementation.py 覆盖了 CodeAct 协议、ObjectRef、存储后端与拦截语义,但完全未覆盖:① InProcessCodeActRuntime 的线程/进程隔离与取消(CancelledError 在单元格执行期间无法到达——_execution_lock 与阻塞 eval 直接相关);② SqlStorage.ensure_metadata 的多进程并发初始化;③ SqlCodeActImplementationStore.approve 并发冲突;④ LlmAgent 上限 max_iterations 到达时已提交结果是否丢失;⑤ CodeAct 与传统 function_call 混用时的重试循环是否消耗迭代额度。触发条件: 上述路径在生产并发/边界条件下触发。实际影响: 高风险路径(并发 SQL 初始化、运行超时兜底、取消传播、结果完整性)无回归保护,缺陷易在发布后暴露。修正方向: 为上述路径补充针对性测试(并发 ensure_metadata 模拟重复启动、RunLimitException 下 codeact.result 保留、单元格执行期间取消、approve 并发 rollback),并维护现有 3 个测试文件对新增行为的断言。
AI Code Review审查结论不通过 本次审查覆盖 发现的问题严重
问题: 触发条件: 用户对 实际影响: 无限递归使每个调用层新建 Runner、会话与 LLM 请求,直到触发宿主侧超时/资源耗尽;调用栈纵深增长的同时并发占用多个 LLM 循环,可能消耗大量 token 与内存,且无日志上的明显终止原因。 修正方向: 在 中等
问题: 触发条件: 实际影响: 审批路径上的工具调用绕过父请求链的审计、参数改写或限制逻辑,回调预期生效的观测/安全环节缺失;且因该路径不再经过父 修正方向: 在 中等
问题: 触发条件: 模型未生成 实际影响: 动态实现路径上工具调用直接报错,模型无法通过反馈修正,已生成的成功实施也无法保存;与其他收集 修正方向: 对 中等
问题: 模型对 触发条件: 模型输出的首个 python cell 以 实际影响: 修正方向: 将 较低
问题: 实现一致性: 触发条件: 模型在 实际影响: 用户观察端事件流中的 feedback/被拒绝调用事件缺失模型用量信息;同时 修正方向: 在 较低
问题: 触发条件: root 内出现指向 root 外的符号链接文件,或部署在硬链接不可用/不可用的文件系统上执行 实际影响: 修正方向: 较低
问题: 触发条件: 以实例方式传入过滤器(如 实际影响: 修正方向: 在 |
| def _resolve_capabilities( | ||
| self, | ||
| context: InvocationContext, | ||
| ) -> list[BaseTool]: | ||
| if self.capability_tools is not None: | ||
| return list(self.capability_tools) | ||
| parent_tools = getattr(context.agent, "tools", None) or [] | ||
| return [tool for tool in parent_tools if tool is not self and getattr(tool, "name", None) != self.name] | ||
|
|
||
| def _build_instruction(self) -> str: |
There was a problem hiding this comment.
问题: _resolve_capabilities 在配置了 capability_tools 时直接返回该列表而不做任何自排除,嵌套调用链 _run_async_impl(构造嵌套 LlmAgent 并把 capabilities 作为其 tools)与 _run_nested_agent(每次调用新建 Runner/InMemorySessionService/全新 invocation 状态与 run-limit 计数、max_iterations=10 循环重置)之间没有任何深度或环检测。当 capability_tools 中包含工具自身(或 A、B 两个 CodeActTool 互为能力)时,模型选择调用对方就会 A→B→A 无限嵌套,每个 CodeActTool 调用各自生成一次独立 LLM 循环。docs/mkdocs/en/codeact.md:372-375 将 tools=capability_tools 作为公开用法给出,CodeActTool.__init__(第 148 行)接受该参数,因此可达。
触发条件: 用户对 CodeActTool 配置 tools=[该工具本身](或相互引用的两个工具),生成的代码调用自排除分支之外的能力工具,模型在嵌套 agent 中再次选择该工具时触发。
实际影响: 无限递归使每个调用层新建 Runner、会话与 LLM 请求,直到触发宿主侧超时/资源耗尽;调用栈纵深增长的同时并发占用多个 LLM 循环,可能消耗大量 token 与内存,且无日志上的明显终止原因。
修正方向: 在 _resolve_capabilities 中无论是否使用 capability_tools 都按名称(或对象)排除 self,并在 _run_nested_agent/_run_async_impl 中增加固定的嵌套深度上限(例如在 invocation 级传递 code_act_nesting_level 并在超过阈值时抛出明确异常)。
| async def _run_approved_implementation( | ||
| self, | ||
| *, | ||
| approved: CodeActImplementation, | ||
| arguments: dict[str, Any], | ||
| capabilities: list[BaseTool], | ||
| code_act: CodeActConfig, | ||
| parent_context: InvocationContext, | ||
| ) -> _CodeActRun: | ||
| """Execute approved cells directly without an LLM or Runner.""" | ||
| execution_agent = _ApprovedExecutionAgent( | ||
| name=f"approved_code_act_{self.name}", | ||
| description=self.description, | ||
| tools=capabilities, | ||
| ) | ||
| execution_context = parent_context.model_copy( | ||
| update={ | ||
| "invocation_id": | ||
| new_invocation_context_id(), | ||
| "agent": | ||
| execution_agent, | ||
| "branch": | ||
| (f"{parent_context.branch}.{execution_agent.name}" if parent_context.branch else execution_agent.name), | ||
| "user_content": | ||
| Content( | ||
| role="user", | ||
| parts=[ | ||
| Part.from_text(text=self._input_model.model_validate(arguments).model_dump_json( | ||
| exclude_none=True)) | ||
| ], | ||
| ), | ||
| }) | ||
| runtime = code_act.runtime | ||
| try: | ||
| execution = await runtime.execute( | ||
| execution_context, | ||
| [prepare_codeact_code(cell) for cell in approved.code_cells], | ||
| ) | ||
| finally: |
There was a problem hiding this comment.
问题: _run_approved_implementation 为执行已审批单元而新建 _ApprovedExecutionAgent(仅包含 tools,没有任何回调或过滤器)并替换 execution_context.agent(第 272-290 行),随后直接调用 runtime.execute 执行 cell,cell 内对 self.<tool> 的调用经由 proxy.__getattr__ 与 tool.run_async 的 _run_filters 完成,但 BaseTool.run_async 中的 ToolCallbackFilter(before_tool_callback, after_tool_callback) 通过 getattr(tool_context.agent, ...) 读取不存在的属性(见 trpc_agent_sdk/tools/_base_tool.py:201-205),因此父 agent 配置的 before_tool_callback/after_tool_callback 被静默跳过;同时执行上下文使用全新 invocation_id。
触发条件: implementation_policy 非 DYNAMIC 且存在兼容的已审批实施,父 agent 配置了工具回调,生成的已审批代码调用 self.<tool>。
实际影响: 审批路径上的工具调用绕过父请求链的审计、参数改写或限制逻辑,回调预期生效的观测/安全环节缺失;且因该路径不再经过父 LlmAgent,父级 run_context.result_format 与 max_llm_calls 之外的限制(MAX_TOOL_CALLS 计数等)也不在该上下文中生效,产生与生成路径不一致的行为。
修正方向: 在 _ApprovedExecutionAgent 上透传父 parent_context.agent 的 before_tool_callback/after_tool_callback(或从 parent_context 直接读取回调并在 _run_async_impl 中构造 ToolCallbackFilter 传给 _run_filters),并复核该路径下 MAX_TOOL_CALLS 等限制是否与生成路径一致。
| async def _run_nested_agent( | ||
| self, | ||
| *, | ||
| agent: Any, | ||
| arguments: dict[str, Any], | ||
| parent_context: InvocationContext, | ||
| ) -> _CodeActRun: | ||
| from trpc_agent_sdk.runners import Runner | ||
|
|
||
| app_name = f"{agent.name}_code_act_function" | ||
| sessions = InMemorySessionService() | ||
| runner = Runner( | ||
| app_name=app_name, | ||
| agent=agent, | ||
| artifact_service=parent_context.artifact_service, | ||
| session_service=sessions, | ||
| memory_service=InMemoryMemoryService(), | ||
| ) | ||
| session = await sessions.create_session( | ||
| app_name=app_name, | ||
| user_id=parent_context.user_id, | ||
| state=parent_context.state.to_dict(), | ||
| ) | ||
| result_text = "" | ||
| code_cells: list[str] = [] | ||
| try: | ||
| async for event in runner.run_async( | ||
| user_id=session.user_id, | ||
| session_id=session.id, | ||
| new_message=Content( | ||
| role="user", | ||
| parts=[ | ||
| Part.from_text(text=self._input_model.model_validate(arguments).model_dump_json( | ||
| exclude_none=True)) | ||
| ], | ||
| ), | ||
| run_config=parent_context.run_config or RunConfig(), | ||
| ): | ||
| if event.actions.state_delta: | ||
| parent_context.state.update(event.actions.state_delta) | ||
| if event.content: | ||
| code_cells.extend(part.executable_code.code or "" for part in event.content.parts | ||
| if part.executable_code) | ||
| if event.object == "codeact.result": | ||
| result_text = event.get_text() | ||
| finally: | ||
| await runner.close() | ||
|
|
||
| if not result_text: | ||
| raise ValueError(f"CodeActTool {self.name!r} produced no CodeAct result") | ||
| return _CodeActRun( | ||
| result=self._output_model.model_validate_json(result_text).model_dump(mode="python"), | ||
| code_cells=code_cells, |
There was a problem hiding this comment.
问题: _run_async_impl 在 approved is None 时以 code_act = self.code_act or CodeActConfig()(require_result=True 默认)执行 _run_nested_agent,而 require_result=False(或 CodeActTool 未配置 code_act 但嵌套模型给出纯文本最终回答)时,_run_nested_agent(第 330-371 行)只通过 event.object == "codeact.result" 捕获结果,事件流中不存在该事件时于第 368 行抛出 ValueError(f"CodeActTool {self.name!r} produced no CodeAct result"),使整个工具调用失败且 save_candidate(第 246-254 行)被跳过。
触发条件: 模型未生成 ```python cell 直接给出纯文本最终回答(或生成的 cell 因语法/引用错误未执行成功),且没有 codeact.result 事件。
实际影响: 动态实现路径上工具调用直接报错,模型无法通过反馈修正,已生成的成功实施也无法保存;与其他收集 codeact.result 失败的路径相比,该路径将模型行为缺陷暴露为硬失败。
修正方向: 对 require_result=False 场景,将嵌套 agent 的反馈事件(codeact.feedback/codeact.validation_error)转成可重试循环(而非立即 raise),或仅当 require_result=True 时抛出该 ValueError,并为纯文本结果回退为把 event.get_text() 作为结果。
| _PYTHON_FENCE_PATTERN = re.compile( | ||
| r"(?P<fence>`{3,}|~{3,})[ \t]*(?P<language>python|py|python3)[ \t]*\n" | ||
| r"(?P<code>.*?)(?:\n)?(?P=fence)", | ||
| re.IGNORECASE | re.DOTALL, | ||
| ) | ||
|
|
||
|
|
||
| class CodeActResponseProcessor: | ||
| """Extract and execute model-generated CodeAct Python cells.""" | ||
|
|
||
| @staticmethod | ||
| async def run_async( | ||
| invocation_context: InvocationContext, | ||
| llm_response: LlmResponse, | ||
| ) -> AsyncGenerator[Event, None]: |
There was a problem hiding this comment.
问题: 模型对 CodeActTool 直接生成的代码使用英语/自定义语言组织 cell(```python 后跟尾随文本)时,_PYTHON_FENCE_PATTERN 语言匹配包含尾随标记,正则失败;同时注释内多个 ```python 时 _extract_first_cell_and_truncate_content 只执行第一个 cell,后续 cell 被丢弃。
触发条件: 模型输出的首个 python cell 以 ```python <<< 形式书写(尾随文本可能由注释、顺序号或测试目的引入),或注释中先出现 ```python 标记。
实际影响: _processor.py 不产生 executable_code 事件,codeact.result 永远不会生成,_run_nested_agent 在 require_result=True 时持续请求代码直至 max_iterations 后报错,或在 require_result=False 时静默执行失败;用户侧表现为模型明明写了代码却没有执行。
修正方向: 将 _PYTHON_FENCE_PATTERN 语言捕获收紧为仅匹配 python|py|python3、按 fence 分隔严格解析(首个有效 cell),并补充回归测试覆盖尾随文本与多重 ```python 场景。
| if pending_code_events and event.usage_metadata: | ||
| pending_code_events[-1].usage_metadata = event.usage_metadata | ||
| codeact_usage_metadata = None |
There was a problem hiding this comment.
问题: 实现一致性:codeact.feedback 的 usage metadata 在两种路径上被置空——codeact_usage_metadata = None 在代码事件存在时清空(第 594-595 行),在不存在代码事件时(not pending_code_events 的 collected_tool_calls 分支)也直接使用 None。
触发条件: 模型在 require_result=True 时返回 return_result 前的纯文本最终响应(第 553 行 codeact_text_only=True),或返回传统工具调用(被拒绝),此时 feedback/被拒绝工具调用事件携带 usage_metadata=None。
实际影响: 用户观察端事件流中的 feedback/被拒绝调用事件缺失模型用量信息;同时 pending_code_events[-1].usage_metadata = event.usage_metadata 覆盖最后一个代码事件(执行结果)的用量,语义偏差但影响有界。
修正方向: 在 collected_tool_calls 分支显式使用 codeact_usage_metadata(而非直接 None),避免该分支的 usage 信息丢失。
| return sorted(path.relative_to(self.root.resolve()) for path in directory.glob(pattern) if path.is_file()) | ||
|
|
||
| return await asyncio.to_thread(_list) | ||
|
|
||
| async def close(self) -> None: | ||
| """Release resources. File storage does not retain open handles.""" | ||
|
|
||
| def _resolve(self, key: str | Path) -> Path: | ||
| root = self.root.resolve() | ||
| path = (root / key).resolve() | ||
| if path != root and root not in path.parents: | ||
| raise ValueError(f"File storage key escapes root: {key!s}") | ||
| return path | ||
|
|
||
| def _write_text_sync( | ||
| self, | ||
| path: Path, | ||
| content: str, | ||
| overwrite: bool, | ||
| ) -> bool: | ||
| path.parent.mkdir(parents=True, exist_ok=True) | ||
| descriptor, temporary_name = tempfile.mkstemp( | ||
| prefix=f".{path.name}.", | ||
| dir=path.parent, | ||
| ) | ||
| try: | ||
| with os.fdopen(descriptor, "w", encoding=self.encoding) as file: | ||
| file.write(content) | ||
| file.flush() | ||
| os.fsync(file.fileno()) | ||
| if overwrite: | ||
| os.replace(temporary_name, path) | ||
| else: | ||
| try: | ||
| os.link(temporary_name, path) | ||
| except FileExistsError: |
There was a problem hiding this comment.
问题: FileStorage.list_files 对匹配到的 path 直接 relative_to(self.root.resolve()),当 directory 中存在指向 root 之外(如符号链接)的文件时 relative_to 抛出 ValueError,导致整个列表操作失败;_write_text_sync 的 overwrite=False 依赖 os.link 抛 FileExistsError,在不支持硬链接的文件系统(FAT/NFS/Windows 无权限)上会以未捕获的 OSError 失败。
触发条件: root 内出现指向 root 外的符号链接文件,或部署在硬链接不可用/不可用的文件系统上执行 save_candidate。
实际影响: FileCodeActImplementationStore.list_implementations 遇到符号链接时抛异常导致审批列表不可用;save_candidate 在硬链接不可用环境下整体失败(略过写入),使 CodeActTool 无法保存生成实施。
修正方向: list_files 改为仅收集 root 内的路径(例如 path.resolve() 后判断是否在 root 内),_write_text_sync 捕获 OSError 并在目标不存在时报错重试为 os.open(..., O_CREAT|O_EXCL) 的单写尝试。
| self._filters = self._convert_filters() | ||
| self._filters_name = [filter.name for filter in self._filters] | ||
|
|
There was a problem hiding this comment.
问题: _init_filters 现在把 self._filters_name 重写为 [filter.name for filter in self._filters],其中 _convert_filters 先追加解析后的注册过滤器实例(name 为注册名),随后保留传入的过滤器实例——FilterABC 的默认 _name 为 ""(trpc_agent_sdk/abc/_filter.py:129),实例过滤器拿到空名。
触发条件: 以实例方式传入过滤器(如 tool.add_filters([MyFilter()])、FunctionTool(filters=[...]))且使用 filter.name 的 filters_name。
实际影响: filters_name 现在对实例过滤器返回 [""],get_filter(type, "") 返回 None,依赖该值的下游(如子 agent 工具、模型过滤)可能触发 ValueError("Filter not found") 或语义漂移的过滤解析。
修正方向: 在 _convert_filters 中为真正的实例过滤器保留用户传入的 filters_name(按位置映射)或在 FilterABC.name 中用具名注册名回填,避免覆盖。
ee71283 to
1467dcc
Compare
- CodeAct Tools: 支持工具返回的大对象可以保留在本地,节省token - Generation Method: 可生成用户定义的函数
1467dcc to
a228552
Compare
AI Code Review审查结论不通过 本变更引入 CodeAct 特性(trpc_agent_sdk/codeact/ 新模块、LlmAgent 集成、CodeActTool 及 4 种实现存储后端、FileStorage/SqlStorage 扩展、3 个测试文件)。审查覆盖 base_commit b80d43a 至 head_commit a228552 全部 43 个非生成变更文件,重点追踪 CodeAct 处理器→运行时→结果解析→Agent 主循环→工具回放→持久化存储的完整链路,并通过子代理对候选问题逐一验证。 计划符合性:核心协议(return_result 标记、Python cell 提取、ObjectRef 大对象保留、max_iterations 循环限制、text-only 重试、usage_metadata 透传、独立于 code_executor 的路由)均已实现且有对应单测覆盖;请求构建(指令注入、屏蔽 provider 工具 schema)、Agent 终态返回、运行时释放均有 finally 兜底。测试较充分(tests/agents/test_codeact.py 599 行 + tests/tools 两个文件 487 行),覆盖成功路径、文本重试、schema 校验、路径独立性、释放失败容错等。 主要风险与门禁结论:发现 2 个高置信 SEVERE 缺陷——(1) 结果判定将任何 stderr 输出视为执行失败,导致合法 return_result 结果被丢弃、Agent 跑满迭代上限后以 RunLimitException 终止;(2) CodeActTool 把含失败轮次的全部 cell 持久化进实现存储,批准回放时必然整体失败且后续 cell 永不执行,批准工作流在常见多轮场景下不可用。另有 1 个 MODERATE(进程级 stdout/stderr 重定向与全局锁造成跨调用输出污染与串行化)和 3 个 LOW(CodeAct 模式丢弃 partial 流式事件、usage_metadata 归属错位、require_result=False 下 output_key 状态与下发内容不一致)。其余候选(多轮 cell 内 return_result 提前中断、marker 解析竞争、filter 重构语义差异等)经证据验证后未达报告门槛。门禁结论:FAILED。 发现的问题严重
问题: 触发条件: 模型生成的 Python cell 向 stderr 写入任意内容(warnings/异常 traceback/显式 print 到 stderr),随后在同一 cell 中调用 实际影响: 合法最终结果被误判为执行失败, 修正方向: 在 严重
问题: 触发条件: 生成过程经过多轮(例如第一轮 cell 报错、第二轮修复后 实际影响: 批准的实现每次执行都必然失败:失败 cell 重放抛异常使整批中止( 修正方向: 中等
问题: 触发条件: 同一进程内并发运行多个 CodeAct 调用(共享同一 实际影响: 一方面所有并发调用被全局锁串行化,单个长 cell 会阻塞其他调用的全部 cell 执行;另一方面其他协程的输出会被捕获进当前 cell 的 stdout/stderr 缓冲区,污染模型可见的执行输出,极端情况下甚至可能把其他调用方打印的内容混入本调用的输出流,干扰 修正方向: 改为每调用独立的锁或直接移除全局锁(各调用已有独立的 较低
问题: 触发条件: 实际影响: 客户端在整个推理期间收不到任何增量文本,只在终态事件到达时才一次性收到全文,流式体验退化为非流式;对需要边生成边展示/边中断的用户场景影响明显,且与 修正方向: 对 较低
问题: 触发条件: CodeAct Agent 单轮执行后 cell 未调用 return_result,或结果校验失败需要反馈下一轮(这是常见多轮路径)。 实际影响: 重试/反馈事件的 usage 统计丢失,调用方可观察的 token 用量在 CodeAct 多轮流程中断档,与 修正方向: 仅当 较低
问题: 触发条件: 实际影响: 修正方向: 在 |
| """Build a model-visible CodeAct execution result.""" | ||
| output = "" | ||
| outcome = Outcome.OUTCOME_OK | ||
| if stderr: | ||
| output = f"Code execution error:\n{stderr}\n" | ||
| outcome = Outcome.OUTCOME_FAILED | ||
| if stdout: | ||
| output += f"Code execution result:\n{stdout}\n" | ||
| return CodeActExecutionResult(outcome=outcome, output=output) |
There was a problem hiding this comment.
问题: create_codeact_execution_result(trpc_agent_sdk/codeact/_runtime.py:56-61)把 "stderr 非空" 直接映射为 OUTCOME_FAILED,而 CodeActResponseProcessor.run_async(trpc_agent_sdk/codeact/_processor.py:90)只在 outcome == OUTCOME_OK 时才解析 return_result 结果标记。二者组合导致:只要 cell 输出中出现任何 stderr 内容(如 warnings.warn(...)、被捕获异常打印的 traceback、print(..., file=sys.stderr)),即使同一 cell 随后调用 return_result 并成功输出结果,该结果也会被当作执行失败而丢弃。本缺陷由本次新增的 stderr→FAILED 判定与新的结果解析流程共同引入。
触发条件: 模型生成的 Python cell 向 stderr 写入任意内容(warnings/异常 traceback/显式 print 到 stderr),随后在同一 cell 中调用 return_result(value);InProcessCodeActRuntime.execute 使用 redirect_stderr 全量捕获 stderr(_in_process_runtime.py:94)。
实际影响: 合法最终结果被误判为执行失败,codeact.result 事件不会产生,Agent 落入执行观察/重试循环(_llm_agent.py 中 code_was_executed 分支),直至耗尽 max_iterations 以 RunLimitException 终止整次调用,用户拿不到本已成功计算的答案。
修正方向: 在 create_codeact_execution_result 或结果解析处将"结果标记存在"与 stderr 解耦:优先检测输出中是否含有效的 return_result 标记(成功则视为 OK,stderr 仅作为附加信息保留在输出文本中),或在 _processor.py 中先扫描结果标记再决定 outcome 分支,而不是仅凭 stderr 非空即判失败。
| if event.content: | ||
| code_cells.extend(part.executable_code.code or "" for part in event.content.parts | ||
| if part.executable_code) |
There was a problem hiding this comment.
问题: CodeActTool._run_nested_agent(trpc_agent_sdk/tools/_code_act_tool.py:359-361)把嵌套 Agent 每轮事件中出现的全部 executable_code cell(含执行失败的轮次)累积进 code_cells,随后在 _run_async_impl(约 241-252 行)整批写入 implementation_store;而 _run_approved_implementation(约 296-299 行)批准回放时把存储的全部 cell 按顺序在同一 globals 命名空间内一次性执行,任何一个早期 cell 重放抛错都会中止后续 cell。
触发条件: 生成过程经过多轮(例如第一轮 cell 报错、第二轮修复后 return_result 成功),且配置了非 DYNAMIC 策略与实现存储;此后管理员批准该候选并复用时,早先失败轮次的 cell 被重放。
实际影响: 批准的实现每次执行都必然失败:失败 cell 重放抛异常使整批中止(execute() 捕获后 stderr 非空 → outcome FAILED,见 _runtime.py:56-61),即使后续 cell 能产出结果标记也会被拒绝(_run_approved_implementation 直接 raise ValueError);这是"生成成功但批准后永远失败"的永久性故障,且同一机制下任何一个已批准 cell 的良性 stderr 输出也会令批准实现不可用,用户只能手动清空存储。
修正方向: _run_nested_agent 只保存成功产出 codeact.result 的最小 cell 序列(例如只收集最后一个成功回合的 cell,或在失败回合后清空累积列表),并让 _run_approved_implementation 对批量执行容忍孤立 cell 失败(逐 cell 独立执行并把 stderr 内容仅作为附注,结果标记存在即成功)。
| stdout = io.StringIO() | ||
| stderr = io.StringIO() | ||
| try: | ||
| async with self._execution_lock: | ||
| with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): | ||
| for cell in cells: | ||
| await self._execute_cell(cell, state.globals) | ||
| except CodeActReturnSignal: | ||
| pass | ||
| except Exception: # noqa: BLE001 # pylint: disable=broad-except | ||
| stderr.write(traceback.format_exc()) | ||
|
|
||
| return create_codeact_execution_result(stdout=stdout.getvalue(), stderr=stderr.getvalue()) |
There was a problem hiding this comment.
问题: InProcessCodeActRuntime.execute(trpc_agent_sdk/codeact/_in_process_runtime.py:93-96)使用实例级单一 _execution_lock 串行化所有调用方,并在持锁期间执行 contextlib.redirect_stdout/redirect_stderr——该重定向替换的是进程全局 sys.stdout/sys.stderr,而 _execute_cell 内部存在 await(顶层 await 及 self.<tool> 调用),事件循环会在重定向生效期间切换到其他协程。
触发条件: 同一进程内并发运行多个 CodeAct 调用(共享同一 CodeActConfig/runtime 实例时持锁串行等待整批执行);任意 cell 在执行中 await 时,另一协程中有 print/日志输出到 stdout(该 SDK 自身 logger 也挂载了 StreamHandler(sys.stdout))。
实际影响: 一方面所有并发调用被全局锁串行化,单个长 cell 会阻塞其他调用的全部 cell 执行;另一方面其他协程的输出会被捕获进当前 cell 的 stdout/stderr 缓冲区,污染模型可见的执行输出,极端情况下甚至可能把其他调用方打印的内容混入本调用的输出流,干扰 return_result 标记解析与观察结果。
修正方向: 改为每调用独立的锁或直接移除全局锁(各调用已有独立的 globals 与 StringIO),并对 cell 执行期间被其他协程污染的输出去重/剥离(例如仅收集持锁期间的输出,或改用 contextvars 感知的流重定向方案)。
| allow_codeact_text = bool(self.code_act and not self.code_act.require_result | ||
| and event.is_final_response() and not pending_code_events) | ||
| if event.content is not None and (not self.code_act or allow_codeact_text): | ||
| yield event | ||
| accumulate_content(event) |
There was a problem hiding this comment.
问题: _llm_agent.py 的 CodeAct 分支(trpc_agent_sdk/agents/_llm_agent.py:586-590)把 allow_codeact_text 收敛到"非 require_result 且无 pending 代码事件",导致在 CodeAct 模式下所有 partial 流式事件被静默丢弃:call_llm_async 在流式循环里只把 partial 事件逐个 yield,这些事件带真实文本 token,但被此处条件拦截。
触发条件: run_config.streaming=True 的 CodeAct Agent(尤其 require_result=False 时模型以纯文本作答)。
实际影响: 客户端在整个推理期间收不到任何增量文本,只在终态事件到达时才一次性收到全文,流式体验退化为非流式;对需要边生成边展示/边中断的用户场景影响明显,且与 require_result=False 模式下"允许文本"的预期相悖。
修正方向: 对 require_result=False 且事件为纯文本 partial 的场景放行 partial 文本事件(仅在存在待处理代码事件时抑制),或在文档/示例中明确说明 CodeAct 模式不支持流式文本输出。
| if pending_code_events and event.usage_metadata: | ||
| pending_code_events[-1].usage_metadata = event.usage_metadata | ||
| codeact_usage_metadata = None |
There was a problem hiding this comment.
问题: _llm_agent.py:593-595 把 event.usage_metadata 附着到 pending_code_events[-1](批内最后一个代码事件)上并随后把 codeact_usage_metadata 置为 None;当该轮未产出 codeact.result(cell 只是产生观察结果)时,usage 元数据被挂到 observation 事件上,而后续构造的 codeact.feedback / 校验错误重试事件(若需要)只能以 usage_metadata=None 发出。
触发条件: CodeAct Agent 单轮执行后 cell 未调用 return_result,或结果校验失败需要反馈下一轮(这是常见多轮路径)。
实际影响: 重试/反馈事件的 usage 统计丢失,调用方可观察的 token 用量在 CodeAct 多轮流程中断档,与 test_llm_agent_codeact_preserves_usage_metadata 保障的单轮行为不一致。
修正方向: 仅当 codeact_final_event(codeact.result)存在时才将 usage 附着到其上,否则保留 codeact_usage_metadata 供后续反馈事件使用,而不是先置 None;或为 feedback 事件单独携带最近一次模型响应的 usage。
| if event.is_final_response(): | ||
| self._save_output_to_state(ctx, event) | ||
| if self.code_act and self.code_act.require_result: | ||
| codeact_text_only = True | ||
| else: | ||
| self._save_output_to_state(ctx, event) |
There was a problem hiding this comment.
问题: require_result=False 模式下,模型在最终响应中同时给出代码 cell 与结尾文本时,_processor.py 的 _retain_non_code_parts 会保留文本并写回 llm_response.content,但 _llm_agent.py:553-557 对该最终响应调用了 _save_output_to_state(把保留文本写入 output_key 状态),而同一文本因 allow_codeact_text 需要 not pending_code_events 而不会被 yield、也不会 accumulate 进会话历史——状态与对外可见输出不一致。
触发条件: require_result=False + 配置了 output_key 的 CodeAct Agent,模型最终回合输出"代码 + 文字结尾"(CodeAct 指令鼓励的混合输出形态,实践中常见)。
实际影响: ctx.state[output_key] 中的最终结果包含模型结尾文字,但用户从事件流中看不到该段文字(只见代码执行观察),下游依赖 output_key 的应用拿到与用户所见不一致的内容。
修正方向: 在 require_result=False 的 CodeAct 分支中,仅当事件确实被 yield/accumulate 时才调用 _save_output_to_state(例如把保存动作移到 allow_codeact_text 分支内,或在代码事件场景下将保留文本附加到 observation 事件一并下发)。