""" Agent Runner - Agent 执行引擎 核心职责： 1. 执行 Agent 任务（循环调用 LLM + 工具） 2. 记录执行轨迹（Trace + Messages + GoalTree） 3. 加载和注入技能（Skill） 4. 管理执行计划（GoalTree） 5. 支持续跑（continue）和回溯重跑（rewind） 参数分层： - Infrastructure: AgentRunner 构造时设置（trace_store, llm_call 等） - RunConfig: 每次 run 时指定（model, trace_id, after_sequence 等） - Messages: OpenAI SDK 格式的任务消息 """ import asyncio from copy import deepcopy import json import logging import os import uuid from dataclasses import dataclass, field from datetime import datetime from hashlib import sha256 from typing import AsyncIterator, Optional, Dict, Any, List, Callable, Literal, Tuple, Union from cyber_agent.trace.models import Trace, Message from cyber_agent.trace.protocols import TraceStore from cyber_agent.trace.trace_id import generate_sub_trace_id from cyber_agent.trace.goal_models import GoalTree from cyber_agent.trace.compaction import ( CompressionConfig, compress_completed_goals, estimate_tokens, needs_level2_compression, build_compression_prompt, ) from cyber_agent.skill.models import Skill from cyber_agent.skill.skill_loader import load_skills_from_dir from cyber_agent.tools import ToolRegistry, get_tool_registry from cyber_agent.tools.approval import ( ToolApprovalBatchV1, ToolApprovalCallV1, approval_grant, tool_argument_hash, ) from cyber_agent.tools.builtin.knowledge import KnowledgeConfig from cyber_agent.core.memory import ( MEMORY_IDENTITY_CONTEXT_KEY, MemoryConfig, compute_memory_identity, ) from cyber_agent.core.dream import DreamScope from cyber_agent.core.run_snapshot import ( RUN_CONFIG_SNAPSHOT_CONTEXT_KEY, RunConfigSnapshotError, RunConfigSnapshotV1, RunConfigSnapshotV2, load_run_config_snapshot, persist_run_config_snapshot, ) from cyber_agent.core.agent_mode import ( AGENT_MODE_CONTEXT_KEY, AgentMode, RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY, RECURSIVE_CHILD_EXECUTION_MODE_CONTEXT_KEY, RECURSIVE_MAX_PARALLEL_CHILDREN_CONTEXT_KEY, apply_policy_to_context, assert_removed_config_absent, policy_from_context, CURRENT_RECURSIVE_REVISION, policy_from_environment, require_mutable_trace_policy, validate_recursive_child_execution, ) from cyber_agent.core.task_protocol import ( RootTaskAnchor, ensure_task_protocol, initialize_task_progress, task_progress_artifact_refs, task_progress_at_revision, task_progress_readiness_error, rewind_task_progress, protocol_error_report, rebuild_pending_replans, ) from cyber_agent.core.task_protocol_service import TaskProtocolService from cyber_agent.core.context_policy import ( canonical_json, ContextPolicyError, normalize_root_task_anchor, persist_root_task_anchor, prune_context_access, render_recursive_context, get_authorized_context_snapshot, require_matching_root_task_anchor, require_root_task_anchor, replace_context_access, ) from cyber_agent.core.resource_budget import ( RESOURCE_BUDGET_CONTEXT_KEY, ResourceBudget, ResourceBudgetController, ResourceBudgetExceeded, ResourceBudgetStateError, ) from cyber_agent.core.artifacts import ( ArtifactRef, ArtifactResolver, MaterialIssue, ValidationMaterial, extract_artifact_refs, material_chars, resolve_artifact_refs, ) from cyber_agent.core.validation import ( LLMValidator, ScopeValidationResult, ValidationPolicy, ValidationResult, ValidationRun, ValidationScope, ValidatorSettings, persist_validation_policy, require_validation_policy, ) from cyber_agent.core.validator_web import ( PageFetcher, SerperWebSearchProvider, ValidatorToolLimits, ValidatorToolSession, ValidatorWebSearchProvider, ) from cyber_agent.core.prompts import ( DEFAULT_SYSTEM_PREFIX, TRUNCATION_HINT, TOOL_INTERRUPTED_MESSAGE, AGENT_INTERRUPTED_SUMMARY, AGENT_CONTINUE_HINT_TEMPLATE, TASK_NAME_GENERATION_SYSTEM_PROMPT, TASK_NAME_FALLBACK, SUMMARY_HEADER_TEMPLATE, build_summary_header, build_tool_interrupted_message, build_agent_continue_hint, ) logger = logging.getLogger(__name__) @dataclass class ContextUsage: """Context 使用情况""" trace_id: str message_count: int token_count: int max_tokens: int usage_percent: float image_count: int = 0 @dataclass class SideBranchContext: """侧分支上下文（压缩/反思/知识评估）""" type: Literal["compression", "reflection", "knowledge_eval"] branch_id: str start_head_seq: int # 侧分支起点的 head_seq start_sequence: int # 侧分支第一条消息的 sequence start_history_length: int # 侧分支起点的 history 长度 start_iteration: int # 旧 Trace 兼容字段；不再用于轮次预算 max_turns: int = 5 # 最大轮次 turns_used: int = 0 # 已持久化的 assistant 轮次 def to_dict(self) -> Dict[str, Any]: """转换为字典（用于持久化和传递给工具）""" return { "type": self.type, "branch_id": self.branch_id, "start_head_seq": self.start_head_seq, "start_sequence": self.start_sequence, "start_iteration": self.start_iteration, "max_turns": self.max_turns, "turns_used": self.turns_used, "is_side_branch": True, "started_at": datetime.now().isoformat(), } # ===== 运行配置 ===== @dataclass class RunConfig: """ 运行参数 — 控制 Agent 如何执行 分为模型层参数（由上游 agent 或用户决定）和框架层参数（由系统注入）。 """ # --- 模型层参数 --- model: str = "gpt-4o" temperature: float = 0.3 max_iterations: int = 200 tools: Optional[List[str]] = None # None = 按 tool_groups 过滤；显式列表 = 精确指定 tool_groups: Optional[List[str]] = field(default_factory=lambda: ["core"]) # 工具分组白名单；默认仅 core，项目按需追加 exclude_tools: List[str] = field(default_factory=list) # 从 tools / tool_groups 结果中再排除的工具名（如远程 agent 禁用 agent/evaluate） side_branch_max_turns: int = 5 # 侧分支最大轮次（压缩/反思） goal_compression: Literal["none", "on_complete", "on_overflow"] = "on_overflow" # Goal 压缩模式 # --- 强制侧分支（用于 API 手动触发或自动压缩流程）--- # 使用列表作为侧分支队列，每次完成一个侧分支后 pop(0) 取下一个 force_side_branch: Optional[List[Literal["compression", "reflection"]]] = None # --- 框架层参数 --- agent_type: str = "default" uid: Optional[str] = None system_prompt: Optional[str] = None # None = 从 skills 自动构建 skills: Optional[List[str]] = None # 注入 system prompt 的 skill 名称列表；None = 按 preset 决定 enable_memory: bool = True auto_execute_tools: bool = True name: Optional[str] = None # 显示名称（空则由 utility_llm 自动生成） enable_prompt_caching: bool = True # 启用 Anthropic Prompt Caching（仅 Claude 模型有效） parallel_tool_execution: bool = False # 是否启用并发 Tool Call 执行（慎用，需确保无资源冲突） child_execution_mode: Literal["sequential", "parallel"] = "sequential" max_parallel_children: int = 2 root_task_anchor: Optional[RootTaskAnchor] = None # --- Trace 控制 --- trace_id: Optional[str] = None # None = 新建 parent_trace_id: Optional[str] = None # 子 Agent 专用 parent_goal_id: Optional[str] = None # --- 续跑控制 --- after_sequence: Optional[int] = None # 从哪条消息后续跑（message sequence） # --- 额外 LLM 参数（传给 llm_call 的 **kwargs）--- extra_llm_params: Dict[str, Any] = field(default_factory=dict) # --- 自定义元数据上下文 --- context: Dict[str, Any] = field(default_factory=dict) # --- 研究流程控制 --- enable_research_flow: bool = True # 是否启用自动研究流程（知识检索→经验检索→调研→计划） # --- 知识管理配置 --- knowledge: KnowledgeConfig = field(default_factory=KnowledgeConfig) # --- Memory 配置（见 cyber_agent/docs/memory.md） --- # None = 默认 Agent（无长期记忆）；赋值 MemoryConfig 使该 Agent 成为 memory-bearing Agent memory: Optional["MemoryConfig"] = None # --- 一次性恢复动作（不进入持久化 RunConfigSnapshot） --- approval_batch_id: Optional[str] = None # --- ApplicationRuntime 固化身份（只由框架装配） --- application_ref: Any = None role_id: Optional[str] = None role_hash: Optional[str] = None effective_run_limits: Dict[str, Any] = field(default_factory=dict) def apply_snapshot( self, snapshot: RunConfigSnapshotV1 | RunConfigSnapshotV2, ) -> None: """Restore all persisted behavior fields while retaining invocation routing.""" self.model = snapshot.model self.temperature = snapshot.temperature self.max_iterations = snapshot.max_iterations self.extra_llm_params = dict(snapshot.extra_llm_params) self.tools = list(snapshot.tools) if snapshot.tools is not None else None self.tool_groups = ( list(snapshot.tool_groups) if snapshot.tool_groups is not None else None ) self.exclude_tools = list(snapshot.exclude_tools) self.auto_execute_tools = snapshot.auto_execute_tools self.agent_type = snapshot.agent_type self.uid = snapshot.uid self.skills = list(snapshot.skills) if snapshot.skills is not None else None self.enable_memory = snapshot.enable_memory self.memory = MemoryConfig(**snapshot.memory) if snapshot.memory else None self.knowledge = KnowledgeConfig(**snapshot.knowledge) self.parallel_tool_execution = snapshot.parallel_tool_execution self.child_execution_mode = snapshot.child_execution_mode self.max_parallel_children = snapshot.max_parallel_children self.side_branch_max_turns = snapshot.side_branch_max_turns self.goal_compression = snapshot.goal_compression self.enable_prompt_caching = snapshot.enable_prompt_caching self.enable_research_flow = snapshot.enable_research_flow self.context = dict(snapshot.custom_context) if isinstance(snapshot, RunConfigSnapshotV2): self.application_ref = dict(snapshot.application_ref) self.role_id = snapshot.role_id self.role_hash = snapshot.role_hash self.effective_run_limits = dict(snapshot.effective_run_limits) # BUILTIN_TOOLS 硬编码列表已移除（2026-04）。 # 工具可用性现在由 @tool(groups=[...]) 声明 + RunConfig.tool_groups 过滤控制。 @dataclass class CallResult: """单次调用结果""" reply: str tool_calls: Optional[List[Dict]] = None trace_id: Optional[str] = None step_id: Optional[str] = None tokens: Optional[Dict[str, int]] = None cost: float = 0.0 # ===== 执行引擎 ===== CONTEXT_INJECTION_INTERVAL = 5 # 每 N 轮注入一次 GoalTree + Collaborators + IM 通知 class AgentRunner: """ Agent 执行引擎 支持三种运行模式（通过 RunConfig 区分）： 1. 新建：trace_id=None 2. 续跑：trace_id=已有ID, after_sequence=None 或 == head 3. 回溯：trace_id=已有ID, after_sequence=N（N < head_sequence） """ def __init__( self, trace_store: Optional[TraceStore] = None, tool_registry: Optional[ToolRegistry] = None, llm_call: Optional[Callable] = None, utility_llm_call: Optional[Callable] = None, skills_dir: Optional[str] = None, goal_tree: Optional[GoalTree] = None, debug: bool = False, logger_name: Optional[str] = None, validation_policy: Optional[ValidationPolicy] = None, validator_search_provider: Optional[ValidatorWebSearchProvider] = None, validator_page_fetcher: Optional[PageFetcher] = None, artifact_resolver: Optional[ArtifactResolver] = None, application_binding: Any = None, context_provider: Any = None, ): """ 初始化 AgentRunner Args: trace_store: Trace 存储 tool_registry: 工具注册表（默认使用全局注册表） llm_call: 主 LLM 调用函数 utility_llm_call: 轻量 LLM（用于生成任务标题等），可选 skills_dir: Skills 目录路径 goal_tree: 初始 GoalTree（可选） debug: 保留参数（已废弃） logger_name: 自定义日志名称（如 "agents.knowledge_manager"），默认用模块名 validation_policy: Recursive 根 Trace 固化的可信验收策略 validator_search_provider: Validator 私有网页搜索适配器 validator_page_fetcher: Validator 受控页面读取适配器 artifact_resolver: Validator 只读产物解析器 """ self.trace_store = trace_store self.tools = tool_registry or get_tool_registry() self.llm_call = llm_call self.utility_llm_call = utility_llm_call self.skills_dir = skills_dir self.goal_tree = goal_tree self.debug = debug self.log = logging.getLogger(logger_name) if logger_name else logger self.validation_policy = validation_policy or ValidationPolicy() self.validator_search_provider = validator_search_provider self.validator_page_fetcher = validator_page_fetcher self.artifact_resolver = artifact_resolver self.application_binding = application_binding self.context_provider = context_provider self.stdin_check: Optional[Callable] = None # 由外部设置，用于子 agent 执行期间检查 stdin self._cancel_events: Dict[str, asyncio.Event] = {} # trace_id → cancel event self._recursive_active_traces: Dict[str, asyncio.Event] = {} self._active_children: Dict[str, set[str]] = {} self._active_parents: Dict[str, str] = {} # 知识保存跟踪（每个 trace 独立） self._saved_knowledge_ids: Dict[str, List[str]] = {} # trace_id → [knowledge_ids] # Context 使用跟踪 self._context_warned: Dict[str, set] = {} # trace_id → {30, 50, 80} 已警告过的阈值 self._context_usage: Dict[str, ContextUsage] = {} # trace_id → 当前用量快照 # 图片优化缓存（避免重复处理） # key: 图片内容的 hash, value: {"downscaled": ..., "description": ...} self._image_opt_cache: Dict[str, Dict[str, Any]] = {} self.resource_budget = ( ResourceBudgetController(trace_store) if trace_store else None ) self.task_protocol_service = ( TaskProtocolService(trace_store) if trace_store else None ) # ===== 核心公开方法 ===== def get_context_usage(self, trace_id: str) -> Optional[ContextUsage]: """获取指定 trace 的 context 使用情况""" return self._context_usage.get(trace_id) async def _resource_budget_for_trace( self, trace_id: str, ) -> tuple[str, ResourceBudget] | None: """解析本地 Trace 所属 Recursive 根树的不可变预算快照。 由模型调用、工具用量记账和子 Agent 创建入口调用；Legacy Trace 直接返回无预算。 """ if not self.trace_store: return None trace = await self.trace_store.get_trace(trace_id) if not trace or not policy_from_context(trace.context).requires_task_protocol: return None root_trace_id = trace.context.get("root_trace_id") or trace.trace_id root = trace if root_trace_id == trace.trace_id else await self.trace_store.get_trace(root_trace_id) if not root: raise ResourceBudgetStateError( f"Recursive root Trace not found: {root_trace_id}" ) snapshot = root.context.get(RESOURCE_BUDGET_CONTEXT_KEY) if snapshot is None: raise ResourceBudgetStateError( "Recursive tree has no persisted resource budget; create a new trace" ) return root_trace_id, ResourceBudget.from_dict(snapshot) async def call_recursive_llm( self, trace_id: str, *, purpose: Literal["ordinary", "root_validator"] = "ordinary", call: Optional[Callable] = None, fail_on_post_response_exhaustion: bool = False, **kwargs: Any, ) -> Dict[str, Any]: """Recursive 树中统一的 LLM 调用和资源记账入口。 Agent 主循环、上下文压缩、图片描述和 Validator 共用；请求前预留次数，响应后记账。 """ llm = call or self.llm_call if not llm: raise ValueError("llm_call function not provided") resolved = await self._resource_budget_for_trace(trace_id) if resolved is None: return await llm(**kwargs) root_trace_id, budget = resolved if not self.resource_budget: raise ResourceBudgetStateError("ResourceBudgetController is unavailable") await self.resource_budget.reserve_llm_call( root_trace_id, budget, purpose=purpose, ) result = await llm(**kwargs) try: await self.resource_budget.record_llm_usage( root_trace_id, budget, prompt_tokens=int(result.get("prompt_tokens", 0) or 0), completion_tokens=int(result.get("completion_tokens", 0) or 0), cost_usd=float(result.get("cost", 0) or 0), ) except ResourceBudgetExceeded as exc: if fail_on_post_response_exhaustion: raise result = dict(result) result["_resource_budget_exceeded"] = exc.dimension return result async def record_recursive_tool_usage( self, trace_id: str, tool_usage: Dict[str, Any], ) -> None: """登记工具内部自行发起的模型用量。 Agent 主循环在 Tool Result 携带 ``tool_usage`` 时调用，并计入同一棵 Recursive 树。 """ resolved = await self._resource_budget_for_trace(trace_id) if resolved is None: return root_trace_id, budget = resolved if not self.resource_budget: raise ResourceBudgetStateError("ResourceBudgetController is unavailable") await self.resource_budget.record_external_llm_usage( root_trace_id, budget, prompt_tokens=int(tool_usage.get("prompt_tokens", 0) or 0), completion_tokens=int(tool_usage.get("completion_tokens", 0) or 0), cost_usd=float(tool_usage.get("cost", 0) or 0), ) async def record_recursive_validation_usage( self, trace_id: str, *, tool_calls: int = 0, material_chars_count: int = 0, provider_cost_usd: float = 0.0, ) -> None: """把 Validator网页工具和真实材料字符计入同一棵树的预算。""" resolved = await self._resource_budget_for_trace(trace_id) if resolved is None: return root_trace_id, budget = resolved if not self.resource_budget: raise ResourceBudgetStateError("ResourceBudgetController is unavailable") await self.resource_budget.record_validation_usage( root_trace_id, budget, tool_calls=tool_calls, material_chars=material_chars_count, provider_cost_usd=provider_cost_usd, ) async def validate_recursive_trace( self, evaluated_trace_id: str, *, scope: Optional[ValidationScope] = None, task_brief: Optional[Dict[str, Any]] = None, task_report: Optional[Dict[str, Any]] = None, completion_criteria: Optional[List[str]] = None, expected_outputs: Optional[List[str]] = None, candidate_output: Optional[str] = None, deterministic_failure: Optional[Dict[str, Any]] = None, root_validator: bool = False, ) -> ValidationRun: """编译Plan、解析材料、顺序运行Scope并持久化聚合缓存。""" if not self.trace_store or not self.llm_call: raise RuntimeError("Validator requires trace_store and llm_call") evaluated = await self.trace_store.get_trace(evaluated_trace_id) if not evaluated: raise ValueError(f"Trace not found: {evaluated_trace_id}") root_trace_id = evaluated.context.get("root_trace_id") or evaluated.trace_id root = ( evaluated if root_trace_id == evaluated.trace_id else await self.trace_store.get_trace(root_trace_id) ) if not root: raise ContextPolicyError(f"Recursive root Trace not found: {root_trace_id}") root_anchor = require_matching_root_task_anchor( root.context, evaluated.context, ) policy, settings = require_validation_policy(root.context) state = ensure_task_protocol(evaluated.context) runtime_policy = policy_from_context(evaluated.context) authoritative_brief = state.get("task_brief") if authoritative_brief is not None: task_brief = authoritative_brief task_brief_version = int(state.get("task_brief_version", 0) or 0) progress_revision = ( state.get("task_progress_head_revision") if root_validator or task_report is None else state.get("task_report_progress_revision") ) task_progress = task_progress_at_revision(state, progress_revision) if runtime_policy.requires_task_progress and task_progress is None: raise ValueError("Recursive revision 3 validation requires TaskProgress") trajectory = await self.trace_store.get_main_path_messages( evaluated_trace_id, evaluated.head_sequence or evaluated.last_sequence, ) refs: list[ArtifactRef] = [] materials: list[ValidationMaterial] = [] source_urls: list[str] = [] material_issues: list[MaterialIssue] = [] try: if isinstance(task_report, dict): refs.extend(extract_artifact_refs(task_report)) source_urls = list(task_report.get("source_urls") or []) refs.extend(task_progress_artifact_refs(task_progress)) if task_progress is not None: for item in ( *task_progress.questions, *task_progress.blockers, *task_progress.findings, *task_progress.hypotheses, *task_progress.work_items, ): for ref in item.context_refs: snapshot = get_authorized_context_snapshot( evaluated.context, ref_id=ref.ref_id, version=ref.version, root_trace_id=root_trace_id, uid=evaluated.uid, ) materials.append(ValidationMaterial( artifact_id=ref.ref_id, version=snapshot.version, content_hash=snapshot.version, kind=f"context.{snapshot.kind}", mime_type="application/json", root_trace_id=root_trace_id, uid=evaluated.uid, content=snapshot.content, )) if root_validator: for message in trajectory: if message.role == "tool" and isinstance(message.content, dict): refs.extend(extract_artifact_refs(message.content)) except Exception as exc: material_issues.append(MaterialIssue( artifact_id="artifact_metadata", outcome="error", reason=f"Invalid artifact metadata: {exc}", )) unique_refs = { (item.artifact_id, item.version, item.content_hash): item for item in refs } resolved_materials, resolved_issues = await resolve_artifact_refs( list(unique_refs.values()), resolver=self.artifact_resolver, root_trace_id=root_trace_id, uid=evaluated.uid, ) materials.extend(resolved_materials) material_issues.extend(resolved_issues) total_material_chars = sum(material_chars(item) for item in materials) if deterministic_failure: material_issues.append(MaterialIssue( artifact_id="execution", outcome=deterministic_failure.get("outcome", "error"), reason=deterministic_failure.get("reason", "Task did not complete"), )) default_model = settings.validator_model or evaluated.model or "" root_model = settings.root_validator_model or default_model model_by_scope = { item: (root_model if item == "root" else default_model) for item in ("evidence", "hypothesis", "output", "task", "root") } if scope == "root" or root_validator: root_validator = True elif scope and task_brief is None: task_brief = { "completion_criteria": completion_criteria or [], "expected_outputs": expected_outputs or [], "validation_scopes": [] if scope == "task" else [scope], } plan = policy.compile_plan( task_brief=task_brief, task_brief_version=task_brief_version, root_task_anchor=root_anchor, task_report=task_report, candidate_output=candidate_output, evaluated_head_sequence=evaluated.head_sequence or evaluated.last_sequence, materials=materials, material_issues=material_issues, model_by_scope=model_by_scope, root=root_validator, task_progress=task_progress, ) cached = state.get("task_report_validation") validation_cache: dict[str, Any] resume_scope_results: list[ScopeValidationResult] = [] if isinstance(cached, dict) and cached.get("plan_hash") == plan.plan_hash: validation_cache = cached try: aggregate_raw = cached.get("aggregate_result") if aggregate_raw: aggregate = ValidationResult.model_validate(aggregate_raw) if ( aggregate.evaluated_trace_id == evaluated_trace_id and aggregate.plan_hash == plan.plan_hash ): return ValidationRun( result=aggregate, trace_ids=[ item.validator_trace_id for item in aggregate.scope_results ], cached=True, ) resume_scope_results = [ ScopeValidationResult.model_validate(item) for item in cached.get("scope_results", []) ] except Exception: resume_scope_results = [] else: validation_cache = { "validation_plan": plan.model_dump(mode="json"), "plan_hash": plan.plan_hash, "scope_results": [], "aggregate_result": None, "validated_at_sequence": plan.evaluated_head_sequence, "material_usage_recorded": total_material_chars == 0, } state["task_report_validation"] = validation_cache await self.trace_store.update_trace( evaluated_trace_id, context=evaluated.context, ) if ( total_material_chars and not validation_cache.get("material_usage_recorded", False) ): await self.record_recursive_validation_usage( evaluated_trace_id, material_chars_count=total_material_chars, ) fresh = await self.trace_store.get_trace(evaluated_trace_id) if not fresh: raise ValueError(f"Trace not found: {evaluated_trace_id}") fresh_state = ensure_task_protocol(fresh.context) fresh_cache = fresh_state.get("task_report_validation") if ( not isinstance(fresh_cache, dict) or fresh_cache.get("plan_hash") != plan.plan_hash ): raise ValueError("Validation cache changed while recording materials") fresh_cache["material_usage_recorded"] = True await self.trace_store.update_trace( evaluated_trace_id, context=fresh.context, ) lineage_event = None if ( evaluated.parent_trace_id and evaluated_trace_id not in self._recursive_active_traces ): lineage_event = self.register_recursive_child( evaluated.parent_trace_id, evaluated_trace_id, ) async def validator_llm_call(**kwargs: Any) -> Dict[str, Any]: result = await self.call_recursive_llm( evaluated_trace_id, purpose="root_validator" if root_validator else "ordinary", **kwargs, ) dimension = result.get("_resource_budget_exceeded") if dimension: raise RuntimeError( f"Validator exceeded tree resource budget: {dimension}" ) return result provider: ValidatorWebSearchProvider | None if settings.search_provider == "disabled": provider = None else: provider = self.validator_search_provider or SerperWebSearchProvider() def tool_session_factory( validation_scope: ValidationScope, allowed_urls: set[str], validator_trace_id: str, ) -> ValidatorToolSession | None: limits_by_scope = { "evidence": ValidatorToolLimits(5, 10, 15), "hypothesis": ValidatorToolLimits(2, 4, 6), "root": ValidatorToolLimits(2, 5, 7), } limits = limits_by_scope.get(validation_scope) if limits is None: return None async def record_usage( tool_calls: int, chars: int, provider_cost_usd: float, ) -> None: await self.record_recursive_validation_usage( evaluated_trace_id, tool_calls=tool_calls, material_chars_count=chars, provider_cost_usd=provider_cost_usd, ) session_kwargs: Dict[str, Any] = {} if self.validator_page_fetcher is not None: session_kwargs["page_fetcher"] = self.validator_page_fetcher return ValidatorToolSession( provider=provider, allowed_urls=allowed_urls, limits=limits, usage_recorder=record_usage, **session_kwargs, ) validator = LLMValidator( llm_call=validator_llm_call, trace_store=self.trace_store, policy=policy, tool_session_factory=tool_session_factory, cancel_check=self.is_cancel_requested, trace_register=self.register_recursive_child, trace_release=self.release_recursive_trace, ) try: async def persist_scope(result: ScopeValidationResult) -> None: fresh = await self.trace_store.get_trace(evaluated_trace_id) if not fresh: raise ValueError(f"Trace not found: {evaluated_trace_id}") fresh_state = ensure_task_protocol(fresh.context) cache = fresh_state.get("task_report_validation") if not isinstance(cache, dict) or cache.get("plan_hash") != plan.plan_hash: raise ValueError("Validation cache changed while scopes were running") by_scope = { item.get("scope"): item for item in cache.get("scope_results", []) if isinstance(item, dict) } by_scope[result.scope] = result.model_dump(mode="json") cache["scope_results"] = [ by_scope[item] for item in plan.effective_scopes if item in by_scope ] await self.trace_store.update_trace( evaluated_trace_id, context=fresh.context, ) run = await validator.validate_plan( evaluated_trace=evaluated, trajectory=trajectory, plan=plan, root_task_anchor=root_anchor, task_brief=task_brief, task_report=task_report, task_progress=task_progress, candidate_output=candidate_output, materials=materials, material_issues=material_issues, model_by_scope=model_by_scope, source_urls=source_urls, resume_scope_results=resume_scope_results, on_scope_result=persist_scope, ) fresh = await self.trace_store.get_trace(evaluated_trace_id) if not fresh: raise ValueError(f"Trace not found: {evaluated_trace_id}") fresh_state = ensure_task_protocol(fresh.context) cache = fresh_state.get("task_report_validation") if not isinstance(cache, dict) or cache.get("plan_hash") != plan.plan_hash: raise ValueError("Validation cache changed before aggregation") cache["aggregate_result"] = run.result.model_dump(mode="json") cache["scope_results"] = [ item.model_dump(mode="json") for item in run.result.scope_results ] await self.trace_store.update_trace( evaluated_trace_id, context=fresh.context, ) return run finally: if lineage_event is not None: self.release_recursive_trace(evaluated_trace_id, lineage_event) async def dream( self, memory_config: MemoryConfig, *, uid: Optional[str], agent_type: str, reflect_model: str = "gpt-4o-mini", dream_model: str = "gpt-4o", ) -> "DreamReport": """执行 dream（整理长期记忆）——外部调度入口。 Agent 主动调用走 dream 工具；外部调度（定时器、CLI）走这个方法。 Args: memory_config: 记忆配置 uid/agent_type: 与 MemoryConfig 一起形成强制 Dream 身份边界 reflect_model: per-trace 反思模型 dream_model: 跨 trace 整合模型 """ from cyber_agent.core.dream import DreamScope, run_dream if not self.trace_store or not self.llm_call: raise RuntimeError("dream 需要 trace_store 和 llm_call 均已配置") return await run_dream( store=self.trace_store, llm_call=self.llm_call, memory_config=memory_config, dream_scope=DreamScope( uid=uid, agent_type=agent_type, memory_identity=compute_memory_identity(memory_config), ), reflect_model=reflect_model, dream_model=dream_model, ) async def run( self, messages: List[Dict], config: Optional[RunConfig] = None, inject_skills: Optional[List[str]] = None, skill_recency_threshold: int = 10, ) -> AsyncIterator[Union[Trace, Message]]: """ Agent 模式执行（核心方法） Args: messages: OpenAI SDK 格式的输入消息 新建: 初始任务消息 [{"role": "user", "content": "..."}] 续跑: 追加的新消息 回溯: 在插入点之后追加的消息 config: 运行配置 inject_skills: 本次调用需要指定注入的 skill 列表（skill 名称） skill_recency_threshold: 最近 N 条消息内有该 skill 就不重复注入 Yields: Union[Trace, Message]: Trace 对象（状态变化）或 Message 对象（执行过程） """ if not self.llm_call: raise ValueError("llm_call function not provided") config = config or RunConfig() trace = None run_cancel_event: Optional[asyncio.Event] = None try: # Phase 1: PREPARE TRACE trace, goal_tree, sequence = await self._prepare_trace(messages, config) # 子 Trace 可能已在排队阶段收到停止信号，不能覆盖既有 Event。 run_cancel_event = self._cancel_events.setdefault( trace.trace_id, asyncio.Event(), ) if policy_from_context(trace.context).mode is AgentMode.RECURSIVE: self._recursive_active_traces[trace.trace_id] = run_cancel_event if trace.parent_trace_id: self._active_children.setdefault( trace.parent_trace_id, set(), ).add(trace.trace_id) self._active_parents[trace.trace_id] = trace.parent_trace_id yield trace # 检查是否有未完成的侧分支（用于用户追加消息场景） side_branch_ctx_for_build: Optional[SideBranchContext] = None if trace.context.get("active_side_branch") and messages: side_branch_data = trace.context["active_side_branch"] # 创建侧分支上下文（用于标记用户追加的消息） side_branch_ctx_for_build = SideBranchContext( type=side_branch_data["type"], branch_id=side_branch_data["branch_id"], start_head_seq=side_branch_data["start_head_seq"], start_sequence=side_branch_data["start_sequence"], start_history_length=0, start_iteration=side_branch_data.get("start_iteration", 0), max_turns=side_branch_data.get("max_turns", config.side_branch_max_turns), turns_used=side_branch_data.get("turns_used", 0), ) # Phase 2: BUILD HISTORY history, sequence, created_messages, head_seq = await self._build_history( trace.trace_id, messages, goal_tree, config, sequence, side_branch_ctx_for_build ) # Update trace's head_sequence in memory trace.head_sequence = head_seq for msg in created_messages: yield msg # Phase 3: AGENT LOOP async for event in self._agent_loop( trace, history, goal_tree, config, sequence, inject_skills=inject_skills, skill_recency_threshold=skill_recency_threshold, ): yield event except asyncio.CancelledError: if trace and config.approval_batch_id: await asyncio.shield( self._mark_approval_execution_unknown(trace.trace_id) ) raise except Exception as e: self.log.error(f"Agent run failed: {e}") # Preparation rejections (for example, attempting to resume an # immutable Validator Trace) must not rewrite the existing record. tid = trace.trace_id if trace else None if self.trace_store and tid: approval_unknown = ( await self._mark_approval_execution_unknown(tid) if config.approval_batch_id else False ) # 读取当前 last_sequence 作为 head_sequence，确保续跑时能加载完整历史 current = await self.trace_store.get_trace(tid) head_seq = current.last_sequence if current else None updates: Dict[str, Any] = { "status": "failed", "head_sequence": head_seq, "error_message": ( "Tool execution outcome is unknown; automatic retry was refused" if approval_unknown else str(e) ), "completed_at": datetime.now(), } if isinstance(e, ResourceBudgetExceeded): current_context = dict(current.context if current else {}) current_context["termination_reason"] = ( f"budget_exhausted:{e.dimension}" ) updates["context"] = current_context await self.trace_store.update_trace( tid, **updates, ) trace_obj = await self.trace_store.get_trace(tid) if trace_obj: yield trace_obj raise finally: if trace and run_cancel_event is not None: self.release_recursive_trace(trace.trace_id, run_cancel_event) async def run_result( self, messages: List[Dict], config: Optional[RunConfig] = None, on_event: Optional[Callable] = None, inject_skills: Optional[List[str]] = None, ) -> Dict[str, Any]: """ 结果模式 — 消费 run()，返回结构化结果。 主要用于 agent/evaluate 工具内部。 Args: on_event: 可选回调，每个 Trace/Message 事件触发一次，用于实时输出子 Agent 执行过程。 inject_skills: 本次调用需要指定注入的 skill 列表（透传给 run()）。 """ last_assistant_text = "" final_trace: Optional[Trace] = None async for item in self.run(messages=messages, config=config, inject_skills=inject_skills): if on_event: on_event(item) if isinstance(item, Message) and item.role == "assistant": content = item.content text = "" if isinstance(content, dict): text = content.get("text", "") or "" elif isinstance(content, str): text = content if text and text.strip(): last_assistant_text = text elif isinstance(item, Trace): final_trace = item config = config or RunConfig() if not final_trace and config.trace_id and self.trace_store: final_trace = await self.trace_store.get_trace(config.trace_id) status = final_trace.status if final_trace else "unknown" error = final_trace.error_message if final_trace else None summary = last_assistant_text stopped_recursively = bool( final_trace and status == "stopped" and policy_from_context(final_trace.context).mode is AgentMode.RECURSIVE ) if not summary and stopped_recursively: summary = "Agent execution stopped." elif not summary and status == "waiting_confirmation": summary = "Agent is waiting for local tool approval." elif not summary: status = "failed" error = error or "Agent 没有产生 assistant 文本结果" # 获取保存的知识 ID trace_id = final_trace.trace_id if final_trace else config.trace_id saved_knowledge_ids = self._saved_knowledge_ids.get(trace_id, []) return { "status": status, "summary": summary, "trace_id": trace_id, "error": error, "saved_knowledge_ids": saved_knowledge_ids, # 新增：返回保存的知识 ID "stats": { "total_messages": final_trace.total_messages if final_trace else 0, "total_tokens": final_trace.total_tokens if final_trace else 0, "total_cost": final_trace.total_cost if final_trace else 0.0, }, } async def stop(self, trace_id: str) -> bool: """ 停止运行中的 Trace Trace API 定位实际 Runner 后调用本方法；Legacy 只停当前 Trace， Recursive 由 ``request_stop`` 把信号传给当前进程内已登记的子树。 Returns: True 如果成功发送停止信号，False 如果该 trace 不在运行中 """ return self.request_stop(trace_id) def request_stop(self, trace_id: str) -> bool: """同步设置停止信号，供 API 与 stdin 回调共用。 Recursive Trace 会沿进程内父子登记表向下遍历，不影响父级或兄弟分支。 """ if trace_id not in self._cancel_events: return False if trace_id not in self._recursive_active_traces: self._cancel_events[trace_id].set() return True pending = [trace_id] visited = set() while pending: current = pending.pop() if current in visited: continue visited.add(current) event = self._cancel_events.get(current) if event: event.set() pending.extend(tuple(self._active_children.get(current, ()))) return True def register_recursive_child( self, parent_trace_id: str, child_trace_id: str, ) -> asyncio.Event: """登记已创建但可能仍在排队的 Recursive 直属孩子。 ``agent`` 工具预创建孩子后、Runner 启动 Validator 前调用，使父级停止能传递到后代。 """ event = self._cancel_events.setdefault(child_trace_id, asyncio.Event()) self._recursive_active_traces[child_trace_id] = event self._active_children.setdefault(parent_trace_id, set()).add(child_trace_id) self._active_parents[child_trace_id] = parent_trace_id parent_event = self._cancel_events.get(parent_trace_id) if parent_event and parent_event.is_set(): event.set() return event def is_cancel_requested(self, trace_id: str) -> bool: event = self._cancel_events.get(trace_id) return bool(event and event.is_set()) def unregister_recursive_trace( self, trace_id: str, event: Optional[asyncio.Event] = None, ) -> None: """幂等清理运行登记；event 防止旧运行误删续跑状态。""" current_event = self._recursive_active_traces.get(trace_id) if event is not None and current_event is not event: return self._recursive_active_traces.pop(trace_id, None) parent_id = self._active_parents.pop(trace_id, None) if parent_id: children = self._active_children.get(parent_id) if children: children.discard(trace_id) if not children: self._active_children.pop(parent_id, None) if not self._active_children.get(trace_id): self._active_children.pop(trace_id, None) def release_recursive_trace( self, trace_id: str, event: Optional[asyncio.Event] = None, ) -> None: """完成一次运行并按对象身份释放取消 Event 与父子登记。""" current_event = self._cancel_events.get(trace_id) if event is not None and current_event is not event: return self.unregister_recursive_trace(trace_id, event) if self._cancel_events.get(trace_id) is current_event: self._cancel_events.pop(trace_id, None) async def _mark_trace_stopped( self, trace_id: str, head_sequence: Optional[int], ) -> Optional[Trace]: if not self.trace_store: return None await self.trace_store.update_trace( trace_id, status="stopped", head_sequence=head_sequence, completed_at=datetime.now(), ) try: from cyber_agent.trace.websocket import broadcast_trace_status_changed await broadcast_trace_status_changed(trace_id, "stopped") except Exception: pass return await self.trace_store.get_trace(trace_id) # ===== 单次调用（保留）===== async def call( self, messages: List[Dict], model: str = "gpt-4o", tools: Optional[List[str]] = None, uid: Optional[str] = None, trace: bool = True, **kwargs ) -> CallResult: """ 单次 LLM 调用（无 Agent Loop） """ if not self.llm_call: raise ValueError("llm_call function not provided") trace_id = None message_id = None tool_schemas = self._get_tool_schemas(tools) if trace and self.trace_store: trace_obj = Trace.create(mode="call", uid=uid, model=model, tools=tool_schemas, llm_params=kwargs) trace_id = await self.trace_store.create_trace(trace_obj) result = await self.llm_call(messages=messages, model=model, tools=tool_schemas, **kwargs) if trace and self.trace_store and trace_id: msg = Message.create( trace_id=trace_id, role="assistant", sequence=1, goal_id=None, content={"text": result.get("content", ""), "tool_calls": result.get("tool_calls")}, prompt_tokens=result.get("prompt_tokens", 0), completion_tokens=result.get("completion_tokens", 0), finish_reason=result.get("finish_reason"), cost=result.get("cost", 0), ) message_id = await self.trace_store.add_message(msg) await self.trace_store.update_trace(trace_id, status="completed", completed_at=datetime.now()) return CallResult( reply=result.get("content", ""), tool_calls=result.get("tool_calls"), trace_id=trace_id, step_id=message_id, tokens={"prompt": result.get("prompt_tokens", 0), "completion": result.get("completion_tokens", 0)}, cost=result.get("cost", 0) ) # ===== Phase 1: PREPARE TRACE ===== async def _prepare_trace( self, messages: List[Dict], config: RunConfig, ) -> Tuple[Trace, Optional[GoalTree], int]: """ 准备 Trace：为 ``run`` 选择新建、续跑或回溯路径。 在 Agent 主循环前调用，并在任何 Trace 操作前校验已废弃的配置项。 Returns: (trace, goal_tree, next_sequence) """ assert_removed_config_absent() if self.application_binding is not None and any( message.get("role") == "system" for message in messages ): raise ValueError( "Application runs do not allow caller-provided system messages" ) if config.trace_id: return await self._prepare_existing_trace(config) else: return await self._prepare_new_trace(messages, config) @staticmethod def _application_ref_dict(value: Any) -> dict[str, Any] | None: if value is None: return None if hasattr(value, "model_dump"): return value.model_dump(mode="json") return dict(value) if isinstance(value, dict) else None def _validate_application_config(self, config: RunConfig) -> None: """Reject a new application RunConfig that diverges from its Binding.""" binding = self.application_binding if binding is None: raise ValueError("Application binding is unavailable") expected_ref = binding.application_ref.model_dump(mode="json") if self._application_ref_dict(config.application_ref) != expected_ref: raise ValueError("RunConfig ApplicationRef does not match Runner binding") if not config.role_id: raise ValueError("Application role_id is required") role = binding.role(config.role_id) if config.role_hash != role.role_hash: raise ValueError("RunConfig role hash does not match Runner binding") if ( config.model != role.role.model or config.temperature != role.role.temperature or config.extra_llm_params != role.role.model_parameters or not set(config.tools or []).issubset(set(role.tool_names)) or config.system_prompt != role.system_prompt or config.skills != [] ): raise ValueError("RunConfig behavior does not match the bound application role") limits = dict(config.effective_run_limits) if not limits: raise ValueError("Application effective_run_limits are required") allowed_limits = role.effective_limits.model_dump(mode="json") for name, allowed in allowed_limits.items(): value = limits.get(name) if value is None or value > allowed: raise ValueError( f"Application run limit exceeds the bound role: {name}" ) if ( config.max_iterations != limits["max_iterations"] or config.max_parallel_children != limits["max_parallel_children"] ): raise ValueError("RunConfig limits do not match effective_run_limits") def _validate_application_snapshot( self, snapshot: RunConfigSnapshotV1 | RunConfigSnapshotV2, trace: Trace, ) -> None: """Perform the Runner-side binding gate before any resume mutation.""" if isinstance(snapshot, RunConfigSnapshotV1) and not isinstance( snapshot, RunConfigSnapshotV2, ): if self.application_binding is not None: raise ValueError("Application-bound Runner cannot resume a V1 Trace") return if self.application_binding is None: raise ValueError( "Application Trace requires an ApplicationRuntime-bound Runner" ) expected_ref = self.application_binding.application_ref.model_dump(mode="json") if snapshot.application_ref != expected_ref: raise ValueError("ApplicationRef does not match Runner binding") role = self.application_binding.role(snapshot.role_id) if snapshot.role_hash != role.role_hash: raise ValueError("Application role hash does not match Runner binding") if ( trace.context.get("application_ref") != snapshot.application_ref or trace.context.get("application_role_id") != snapshot.role_id or trace.context.get("application_role_hash") != snapshot.role_hash or trace.context.get("effective_run_limits") != snapshot.effective_run_limits ): raise ValueError("Application snapshot does not match Trace context") restored = RunConfig() restored.apply_snapshot(snapshot) restored.system_prompt = role.system_prompt restored.skills = [] self._validate_application_config(restored) async def _prepare_new_trace( self, messages: List[Dict], config: RunConfig, ) -> Tuple[Trace, Optional[GoalTree], int]: """创建并持久化一个新根 Trace。 ``run`` 首次执行时调用；Recursive 会在此固化根任务锚点和树级预算。 """ if self.application_binding is not None: self._validate_application_config(config) elif config.application_ref is not None: raise ValueError( "Application runs require an ApplicationRuntime-bound Runner" ) # 在任何标题生成/LLM 调用前完成模式校验。 policy = policy_from_environment( recursive_revision=CURRENT_RECURSIVE_REVISION, ) if policy.mode is AgentMode.RECURSIVE: validate_recursive_child_execution( config.child_execution_mode, config.max_parallel_children, ) if config.root_task_anchor is None: raise ValueError( "New Recursive root traces require root_task_anchor" ) try: root_task_anchor = normalize_root_task_anchor( config.root_task_anchor ) except ContextPolicyError as exc: raise ValueError(str(exc)) from exc deployment_budget = ResourceBudget.from_environment() if self.application_binding is not None: limits = config.effective_run_limits budget = ResourceBudget( enabled=deployment_budget.enabled, max_total_agents=int(limits["max_total_agents"]), max_llm_calls=int(limits["max_llm_calls"]), max_total_tokens=int(limits["max_total_tokens"]), max_total_cost_usd=float(limits["max_total_cost_usd"]), max_duration_seconds=int(limits["max_duration_seconds"]), reserved_final_calls=min( deployment_budget.reserved_final_calls, int(limits["max_llm_calls"]) - 1, ), max_validation_tool_calls=int( limits["max_validation_tool_calls"] ), max_validation_material_chars=int( limits["max_validation_material_chars"] ), ) else: budget = deployment_budget validator_settings = ValidatorSettings.from_environment() trace_id = str(uuid.uuid4()) # 生成任务名称 task_name = config.name or ( self._fallback_task_name(messages) if policy.mode is AgentMode.RECURSIVE else await self._generate_task_name(messages) ) # 准备工具 Schema tool_schemas = self._get_tool_schemas(config.tools, config.tool_groups, config.exclude_tools) trace_context = apply_policy_to_context(config.context, policy) trace_context.setdefault("agent_depth", 0) trace_context.setdefault("root_trace_id", trace_id) memory_identity = ( compute_memory_identity(config.memory) if config.memory else None ) if memory_identity is not None: trace_context[MEMORY_IDENTITY_CONTEXT_KEY] = memory_identity if self.application_binding is not None: run_snapshot = RunConfigSnapshotV2.from_run_config( config, memory_identity=memory_identity, ) trace_context.update({ "application_ref": run_snapshot.application_ref, "application_role_id": run_snapshot.role_id, "application_role_hash": run_snapshot.role_hash, "effective_run_limits": run_snapshot.effective_run_limits, }) else: run_snapshot = RunConfigSnapshotV1.from_run_config( config, memory_identity=memory_identity, ) persist_run_config_snapshot(trace_context, run_snapshot) if policy.requires_task_protocol: persist_root_task_anchor(trace_context, root_task_anchor) persist_validation_policy( trace_context, self.validation_policy, validator_settings, ) trace_context[RESOURCE_BUDGET_CONTEXT_KEY] = budget.to_dict() state = ensure_task_protocol(trace_context) if policy.requires_task_progress: initialize_task_progress( state, root_task_anchor_hash=trace_context.get( "root_task_anchor_hash" ), ) replace_context_access( trace_context, [], root_task_anchor=root_task_anchor, task_brief=state.get("task_brief"), ) if self.application_binding is not None and self.context_provider is not None: from cyber_agent.application.context import load_application_context from cyber_agent.application.ports import ContextRequest await load_application_context( self.application_binding, trace_context, ContextRequest( application_ref=self.application_binding.application_ref, root_trace_id=trace_id, trace_id=trace_id, uid=config.uid, role_id=config.role_id, task_brief=None, task_brief_revision=0, ), root_task_anchor=root_task_anchor, task_brief=None, granted_at_sequence=0, ) trace_obj = Trace( trace_id=trace_id, mode="agent", task=task_name, agent_type=config.agent_type, parent_trace_id=config.parent_trace_id, parent_goal_id=config.parent_goal_id, uid=config.uid, model=config.model, tools=tool_schemas, llm_params={"temperature": config.temperature, **config.extra_llm_params}, context=trace_context, status="running", ) goal_tree = self.goal_tree or GoalTree(mission=task_name) if self.trace_store: await self.trace_store.create_trace(trace_obj) await self.trace_store.update_goal_tree(trace_id, goal_tree) assert self.resource_budget is not None if policy.mode is AgentMode.RECURSIVE: await self.resource_budget.initialize( trace_id, budget, initial_agents=1, ) return trace_obj, goal_tree, 1 async def _prepare_existing_trace( self, config: RunConfig, ) -> Tuple[Trace, Optional[GoalTree], int]: """加载已有 Trace，决定续跑或回溯。 ``run`` 携带 ``trace_id`` 时调用；Recursive 只信任持久化模式、预算和协议状态。 """ if not self.trace_store: raise ValueError("trace_store required for continue/rewind") trace_obj = await self.trace_store.get_trace(config.trace_id) if not trace_obj: raise ValueError(f"Trace not found: {config.trace_id}") if ( trace_obj.agent_type == "validator" or trace_obj.context.get("created_by_tool") == "validator" ): raise ValueError( "Validator traces cannot be continued or rewound" ) require_mutable_trace_policy(trace_obj.context) # Resume/rewind behavior is always recovered from the immutable snapshot. # Old Recursive traces cannot be reconstructed safely; Legacy traces get # one explicitly-marked, conservative inferred snapshot for compatibility. if RUN_CONFIG_SNAPSHOT_CONTEXT_KEY in trace_obj.context: try: snapshot = load_run_config_snapshot(trace_obj.context) except RunConfigSnapshotError as exc: raise ValueError(str(exc)) from exc else: existing_policy = policy_from_context(trace_obj.context) if existing_policy.mode is AgentMode.RECURSIVE: raise ValueError( "This Recursive trace predates RunConfig snapshots; create a new trace" ) inferred = RunConfig( model=trace_obj.model or "gpt-4o", temperature=float((trace_obj.llm_params or {}).get("temperature", 0.3)), tools=[ item.get("function", {}).get("name") for item in (trace_obj.tools or []) if item.get("function", {}).get("name") ], tool_groups=None, agent_type=trace_obj.agent_type or "default", uid=trace_obj.uid, extra_llm_params={ key: value for key, value in (trace_obj.llm_params or {}).items() if key != "temperature" }, context={ "project_name": trace_obj.context.get("project_name") } if trace_obj.context.get("project_name") else {}, ) snapshot = RunConfigSnapshotV1.from_run_config( inferred, memory_identity=None, legacy_inferred=True, ) persist_run_config_snapshot(trace_obj.context, snapshot) await self.trace_store.update_trace( trace_obj.trace_id, context=trace_obj.context, ) if snapshot.uid != trace_obj.uid or snapshot.agent_type != (trace_obj.agent_type or "default"): raise ValueError("run config snapshot identity does not match Trace metadata") self._validate_application_snapshot(snapshot, trace_obj) config.apply_snapshot(snapshot) if isinstance(snapshot, RunConfigSnapshotV2): role = self.application_binding.role(snapshot.role_id) config.system_prompt = role.system_prompt config.skills = [] persisted_memory_identity = trace_obj.context.get( MEMORY_IDENTITY_CONTEXT_KEY ) if persisted_memory_identity != snapshot.memory_identity: raise ValueError( "run config snapshot memory identity does not match Trace context" ) if config.memory is not None and ( compute_memory_identity(config.memory) != snapshot.memory_identity ): raise ValueError( "restored MemoryConfig does not match the persisted memory identity" ) # Historical traces predate AGENT_MODE. They resume in safe Legacy mode # and the choice is persisted immediately; environment changes never # mutate an existing tree. if AGENT_MODE_CONTEXT_KEY not in trace_obj.context: legacy_policy = policy_from_context(None) trace_obj.context = apply_policy_to_context(trace_obj.context, legacy_policy) await self.trace_store.update_trace( config.trace_id, context=trace_obj.context, ) else: policy = policy_from_context(trace_obj.context) if policy.mode is AgentMode.RECURSIVE: validate_recursive_child_execution( config.child_execution_mode, config.max_parallel_children, ) if policy.requires_task_protocol: root_trace_id = trace_obj.context.get("root_trace_id") root = ( trace_obj if root_trace_id == trace_obj.trace_id else await self.trace_store.get_trace(root_trace_id) ) if not root: raise ValueError( f"Recursive root Trace not found: {root_trace_id}" ) try: require_matching_root_task_anchor( root.context, trace_obj.context, ) require_validation_policy(root.context) except ContextPolicyError as exc: raise ValueError(str(exc)) from exc except ValueError as exc: raise ValueError(str(exc)) from exc if RESOURCE_BUDGET_CONTEXT_KEY not in root.context: raise ValueError( "This experimental Recursive trace predates tree resource " "budgets; create a new trace" ) ResourceBudget.from_dict( root.context[RESOURCE_BUDGET_CONTEXT_KEY] ) if root_trace_id == trace_obj.trace_id: state = ensure_task_protocol(trace_obj.context) if state["root_validation_attempts"] >= 2: raise ValueError( "Root task already used its two independent " "validation attempts; create a new trace" ) if policy.requires_task_progress: state = ensure_task_protocol(trace_obj.context) had_report = state.get("task_report") is not None report_progress_revision = state.get( "task_report_progress_revision" ) if had_report and report_progress_revision is None: raise ValueError( "TaskReport is missing its TaskProgress revision binding" ) previous_head = state.get("task_progress_head_revision") rewind_task_progress(state, trace_obj.last_sequence) if ( had_report and state.get("task_report_progress_revision") is None ): raise ValueError( "TaskReport references TaskProgress outside the " "current revision ancestry" ) if state.get("task_progress_head_revision") != previous_head: await self.trace_store.update_trace( trace_obj.trace_id, context=trace_obj.context, ) assert self.resource_budget is not None await self.resource_budget.get_usage(root.trace_id) goal_tree = await self.trace_store.get_goal_tree(config.trace_id) if goal_tree is None: # 防御性兜底：trace 存在但 goal.json 丢失时，创建空树 goal_tree = GoalTree(mission=trace_obj.task or "Agent task") await self.trace_store.update_goal_tree(config.trace_id, goal_tree) # 自动判断行为：after_sequence 为 None 或 == head → 续跑；< head → 回溯 after_seq = config.after_sequence # 如果 after_seq > head_sequence，说明 generator 被强制关闭时 store 的 # head_sequence 未来得及更新（仍停在 Phase 2 写入的初始值）。 # 用 last_sequence 修正 head_sequence，确保续跑时能看到完整历史。 if after_seq is not None and after_seq > trace_obj.head_sequence: trace_obj.head_sequence = trace_obj.last_sequence await self.trace_store.update_trace( config.trace_id, head_sequence=trace_obj.head_sequence ) if after_seq is not None and after_seq < trace_obj.head_sequence: # 回溯模式 sequence = await self._rewind(config.trace_id, after_seq, goal_tree) else: # 续跑模式：从 last_sequence + 1 开始 sequence = trace_obj.last_sequence + 1 # 状态置为 running await self.trace_store.update_trace( config.trace_id, status="running", completed_at=None, ) trace_obj.status = "running" # 广播状态变化给前端 try: from cyber_agent.trace.websocket import broadcast_trace_status_changed await broadcast_trace_status_changed(config.trace_id, "running") except Exception: pass return trace_obj, goal_tree, sequence # ===== Phase 2: BUILD HISTORY ===== async def _build_history( self, trace_id: str, new_messages: List[Dict], goal_tree: Optional[GoalTree], config: RunConfig, sequence: int, side_branch_ctx: Optional[SideBranchContext] = None, ) -> Tuple[List[Dict], int, List[Message], int]: """ 构建完整的 LLM 消息历史 1. 从 head_sequence 沿 parent chain 加载主路径消息（续跑/回溯场景） 2. 构建 system prompt（新建时注入 skills） 3. 新建时：在第一条 user message 末尾注入当前经验 4. 追加 input messages（设置 parent_sequence 链接到当前 head） 5. 如果在侧分支中，追加的消息自动标记为侧分支消息 Returns: (history, next_sequence, created_messages, head_sequence) created_messages: 本次新创建并持久化的 Message 列表，供 run() yield 给调用方 head_sequence: 当前主路径头节点的 sequence """ history: List[Dict] = [] created_messages: List[Message] = [] head_seq: Optional[int] = None # 当前主路径的头节点 sequence # 1. 加载已有 messages（通过主路径遍历） if config.trace_id and self.trace_store: trace_obj = await self.trace_store.get_trace(trace_id) if trace_obj and trace_obj.head_sequence > 0: main_path = await self.trace_store.get_main_path_messages( trace_id, trace_obj.head_sequence ) # An approved pending batch deliberately has orphaned calls; # they are executed before the next model turn. if not config.approval_batch_id: main_path, sequence = await self._heal_orphaned_tool_calls( main_path, trace_id, goal_tree, sequence, ) history = [msg.to_llm_dict() for msg in main_path] if main_path: head_seq = main_path[-1].sequence snapshot = load_run_config_snapshot(trace_obj.context) if snapshot.system_prompt_hash: persisted_system = next( ( message.get("content") for message in history if message.get("role") == "system" and isinstance(message.get("content"), str) ), None, ) if persisted_system is None or sha256( persisted_system.encode("utf-8") ).hexdigest() != snapshot.system_prompt_hash: raise ValueError("persisted system prompt does not match RunConfig snapshot") current_trace = ( await self.trace_store.get_trace(trace_id) if self.trace_store else None ) if ( current_trace and current_trace.head_sequence == 0 and policy_from_context(current_trace.context).requires_task_protocol ): anchor = require_root_task_anchor(current_trace.context) anchor_text = ( "# Root Task Anchor\n\n" + canonical_json(anchor.model_dump(mode="json")) ) anchored_messages = [] injected = False for message in new_messages: if not injected and message.get("role") == "user": content = message.get("content") or "" if isinstance(content, str): anchored_content = f"{anchor_text}\n\n{content}" elif isinstance(content, list): anchored_content = [ {"type": "text", "text": anchor_text}, *content, ] else: raise ValueError( "Recursive root anchor requires text or multimodal user content" ) anchored_messages.append({ **message, "content": anchored_content, }) injected = True else: anchored_messages.append(message) if not injected: anchored_messages.append({"role": "user", "content": anchor_text}) new_messages = anchored_messages # 2. 构建/注入 skills 到 system prompt has_system = any(m.get("role") == "system" for m in history) has_system_in_new = any(m.get("role") == "system" for m in new_messages) if not has_system: if has_system_in_new: # 入参消息已含 system，将 skills 注入其中（在 step 4 持久化之前） augmented = [] for msg in new_messages: if msg.get("role") == "system": base = msg.get("content") or "" enriched = await self._build_system_prompt(config, base_prompt=base) augmented.append({**msg, "content": enriched or base}) else: augmented.append(msg) new_messages = augmented else: # 没有 system，自动构建并插入历史 system_prompt = await self._build_system_prompt(config) if system_prompt: history = [{"role": "system", "content": system_prompt}] + history if self.trace_store: system_msg = Message.create( trace_id=trace_id, role="system", sequence=sequence, goal_id=None, content=system_prompt, parent_sequence=None, # system message 是 root ) await self.trace_store.add_message(system_msg) created_messages.append(system_msg) head_seq = sequence sequence += 1 # 3. 追加新 messages（设置 parent_sequence 链接到当前 head） for msg_dict in new_messages: history.append(msg_dict) if self.trace_store: # 如果在侧分支中，标记为侧分支消息 if side_branch_ctx: stored_msg = Message.create( trace_id=trace_id, role=msg_dict["role"], sequence=sequence, goal_id=goal_tree.current_id if goal_tree else None, parent_sequence=head_seq, branch_type=side_branch_ctx.type, branch_id=side_branch_ctx.branch_id, content=msg_dict.get("content"), ) self.log.info(f"用户在侧分支 {side_branch_ctx.type} 中追加消息") else: stored_msg = Message.from_llm_dict( msg_dict, trace_id=trace_id, sequence=sequence, goal_id=None, parent_sequence=head_seq, ) await self.trace_store.add_message(stored_msg) created_messages.append(stored_msg) head_seq = sequence sequence += 1 # 5. 更新 trace 的 head_sequence if self.trace_store and head_seq is not None: await self.trace_store.update_trace(trace_id, head_sequence=head_seq) persisted = await self.trace_store.get_trace(trace_id) system_content = next( ( message.get("content") for message in history if message.get("role") == "system" and isinstance(message.get("content"), str) ), None, ) if persisted and system_content is not None: snapshot = load_run_config_snapshot(persisted.context) # Pre-created Recursive children enter through the resume path # on their very first execution. Bind their actual persisted # system prompt once, just like a newly-created root Trace. if snapshot.system_prompt_hash is None: snapshot = snapshot.model_copy(update={ "system_prompt_hash": sha256( system_content.encode("utf-8") ).hexdigest() }) persist_run_config_snapshot(persisted.context, snapshot) await self.trace_store.update_trace( trace_id, context=persisted.context, ) return history, sequence, created_messages, head_seq or 0 # ===== Phase 3: AGENT LOOP ===== async def _manage_context_usage( self, trace_id: str, history: List[Dict], goal_tree: Optional[GoalTree], config: RunConfig, sequence: int, head_seq: int, ) -> Tuple[List[Dict], int, int, bool]: """ 管理 context 用量：检查、预警、压缩 Returns: (updated_history, new_head_seq, next_sequence, needs_enter_compression_branch) """ compression_config = CompressionConfig() token_count = estimate_tokens(history) max_tokens = compression_config.get_max_tokens(config.model) # 计算使用率 progress_pct = (token_count / max_tokens * 100) if max_tokens > 0 else 0 msg_count = len(history) img_count = sum( 1 for msg in history if isinstance(msg.get("content"), list) for part in msg["content"] if isinstance(part, dict) and part.get("type") in ("image", "image_url") ) # 更新 context usage 快照 self._context_usage[trace_id] = ContextUsage( trace_id=trace_id, message_count=msg_count, token_count=token_count, max_tokens=max_tokens, usage_percent=progress_pct, image_count=img_count, ) # 阈值警告（30%, 50%, 80%） if trace_id not in self._context_warned: self._context_warned[trace_id] = set() for threshold in [30, 50, 80]: if progress_pct >= threshold and threshold not in self._context_warned[trace_id]: self._context_warned[trace_id].add(threshold) self.log.warning( f"Context 使用率达到 {threshold}%: {token_count:,} / {max_tokens:,} tokens ({msg_count} 条消息)" ) # 检查是否需要压缩（仅基于 token 数量） needs_compression = token_count > max_tokens if not needs_compression: return history, head_seq, sequence, False # 检查是否有待评估知识（压缩前必须先评估） if self.trace_store and not config.force_side_branch: pending = await self.trace_store.get_pending_knowledge_entries(trace_id) if pending: # 设置侧分支队列：反思 → 知识评估 → 压缩 # 反思放在前面，确保反思期间完成的 goal 产生的新知识也能在压缩前被评估 if config.knowledge.enable_extraction: config.force_side_branch = ["reflection", "knowledge_eval", "compression"] else: config.force_side_branch = ["knowledge_eval", "compression"] # 在 trace.context 中设置触发事件 trace = await self.trace_store.get_trace(trace_id) if trace: if not trace.context: trace.context = {} trace.context["knowledge_eval_trigger"] = "compression" await self.trace_store.update_trace(trace_id, context=trace.context) self.log.info(f"[Knowledge Eval] 压缩前触发知识评估，待评估: {len(pending)} 条") return history, head_seq, sequence, True # 知识提取：在任何压缩发生前，用完整 history 做反思（进入反思侧分支） if config.knowledge.enable_extraction and not config.force_side_branch: # 设置侧分支队列：先反思，再压缩 config.force_side_branch = ["reflection", "compression"] return history, head_seq, sequence, True # 以下为未启用反思、需要压缩的情况，直接进行level 1压缩，并检查是否需要进行level 2压缩（进入侧分支） # Level 1 压缩：Goal 完成压缩 if config.goal_compression != "none" and self.trace_store and goal_tree: if head_seq > 0: main_path_msgs = await self.trace_store.get_main_path_messages( trace_id, head_seq ) compressed_msgs = compress_completed_goals(main_path_msgs, goal_tree) if len(compressed_msgs) < len(main_path_msgs): self.log.info( "Level 1 压缩: %d -> %d 条消息", len(main_path_msgs), len(compressed_msgs), ) history = [msg.to_llm_dict() for msg in compressed_msgs] else: self.log.info( "Level 1 压缩: 无可过滤消息 (%d 条全部保留)", len(main_path_msgs), ) elif needs_compression: self.log.warning( "Token 数 (%d) 超过阈值，但无法执行 Level 1 压缩（缺少 store 或 goal_tree，或 goal_compression=none）", token_count, ) # Level 2 压缩：检查 Level 1 后是否仍超阈值 # 注意：Level 1 压缩后需要重新优化图片并计算 token optimized_history_after = await self._optimize_images( history, config.model, trace_id=trace_id, ) token_count_after = estimate_tokens(optimized_history_after) needs_level2 = token_count_after > max_tokens if needs_level2: self.log.info( "Level 1 后仍超阈值 (token=%d/%d)，需要进入压缩侧分支", token_count_after, max_tokens, ) # 如果还没有设置侧分支（说明没有启用知识提取），直接进入压缩 if not config.force_side_branch: config.force_side_branch = ["compression"] # 返回标志，让主循环进入侧分支 return history, head_seq, sequence, True # 压缩完成后，输出最终发给模型的消息列表 self.log.info("Level 1 压缩完成，发送给模型的消息列表:") for idx, msg in enumerate(history): role = msg.get("role", "unknown") content = msg.get("content", "") if isinstance(content, str): preview = content[:100] + ("..." if len(content) > 100 else "") elif isinstance(content, list): preview = f"[{len(content)} blocks]" else: preview = str(content)[:100] self.log.info(f" [{idx}] {role}: {preview}") return history, head_seq, sequence, False async def _build_knowledge_eval_prompt( self, trace_id: str, goal_tree: Optional[GoalTree] ) -> str: """构建知识评估 prompt""" if not self.trace_store: return "" pending = await self.trace_store.get_pending_knowledge_entries(trace_id) if not pending: return "" # 获取mission trace = await self.trace_store.get_trace(trace_id) mission = trace.task if trace else "未知任务" # 获取当前Goal current_goal = goal_tree.find(goal_tree.current_id) if goal_tree and goal_tree.current_id else None goal_desc = current_goal.description if current_goal else "无当前目标" # 构建知识列表 knowledge_list = [] for idx, entry in enumerate(pending, 1): knowledge_list.append( f"### 知识 {idx}: {entry['knowledge_id']}\n" f"- task: {entry['task']}\n" f"- content: {entry['content']}\n" f"- 注入于: sequence {entry['injected_at_sequence']}, goal {entry['goal_id']}" ) prompt = f"""你是知识评估助手。请评估以下知识在本次任务执行中的实际效果。 ## 当前任务（Mission） {mission} ## 当前 Goal {goal_desc} ## 待评估知识列表 {chr(10).join(knowledge_list)} ## 评估维度 1. **helpfulness**: 知识内容是否对完成任务有实质帮助？ 2. **relevance**: 执行过程中是否体现了该知识的内容？ ## 评估分类 - irrelevant: task与当前任务无关 - unused: 相关但未使用 - helpful: 有帮助 - harmful: 有负面作用 - neutral: 无明显作用 ## 输出格式 请直接输出评估结果，使用JSON格式： {{ "evaluations": [ {{ "knowledge_id": "knowledge-xxx", "eval_status": "helpful", "reason": "1-2句评估理由" }} ] }} """ return prompt async def _single_turn_compress( self, trace_id: str, history: List[Dict], goal_tree: Optional[GoalTree], config: RunConfig, ) -> str: """单次 LLM 调用生成压缩摘要，返回 summary 文本""" self.log.info("执行单次 LLM 压缩") # 构建压缩 prompt（使用 SINGLE_TURN_PROMPT） from cyber_agent.core.prompts import build_single_turn_prompt goal_prompt = goal_tree.to_prompt(include_summary=True) if goal_tree else "" compress_prompt = build_single_turn_prompt(goal_prompt) compress_messages = list(history) + [ {"role": "user", "content": compress_prompt} ] # 应用 Prompt Caching compress_messages = self._add_cache_control( compress_messages, config.model, config.enable_prompt_caching ) # 单次 LLM 调用（无工具） result = await self.call_recursive_llm( trace_id, purpose="ordinary", messages=compress_messages, model=config.model, tools=[], # 不提供工具 temperature=config.temperature, fail_on_post_response_exhaustion=True, **config.extra_llm_params, ) summary_text = result.get("content", "").strip() # 提取 [[SUMMARY]] 块 if "[[SUMMARY]]" in summary_text: summary_text = summary_text[ summary_text.index("[[SUMMARY]]") + len("[[SUMMARY]]"): ].strip() return summary_text @staticmethod def _try_fix_json(s: str) -> Optional[dict]: """尝试修复常见的 JSON 截断/格式问题，返回 dict 或 None""" import re fixed = s.strip() # 1. 修复值中未转义的引号（如 "key": "he said "hello" to me"） # 策略：找到 key-value 模式中值字符串内部的裸引号并转义 def _fix_inner_quotes(text: str) -> str: # 匹配 ": "..." 模式，修复值内部的未转义引号 result = [] i = 0 while i < len(text): # 找到 ": " 后面的值字符串开头 if text[i] == '"': # 找到这个引号对应的字符串结束位置 j = i + 1 while j < len(text): if text[j] == '\\': j += 2 # 跳过转义字符 continue if text[j] == '"': break j += 1 # 检查引号后面是否是合法的 JSON 分隔符 if j < len(text): after = j + 1 # 跳过空白 while after < len(text) and text[after] in ' \t\n\r': after += 1 if after < len(text) and text[after] not in ':,}]\n\r': # 这个引号不是真正的结束引号，继续往后找 # 找到下一个后面跟合法分隔符的引号 k = j + 1 found_end = False while k < len(text): if text[k] == '"': peek = k + 1 while peek < len(text) and text[peek] in ' \t\n\r': peek += 1 if peek >= len(text) or text[peek] in ':,}]': # 这才是真正的结束引号，转义中间的引号 inner = text[i+1:k].replace('"', '\\"') result.append('"' + inner + '"') i = k + 1 found_end = True break k += 1 if found_end: continue result.append(text[i]) i += 1 return ''.join(result) fixed = _fix_inner_quotes(fixed) # 2. 去掉尾部多余逗号 fixed = re.sub(r',\s*([}\]])', r'\1', fixed) # 3. 尝试补全截断的字符串和括号 for suffix in ['', '"', '"}', '"]', '"}]', '"}}']: try: attempt = fixed + suffix open_braces = attempt.count('{') - attempt.count('}') open_brackets = attempt.count('[') - attempt.count(']') attempt += '}' * max(0, open_braces) + ']' * max(0, open_brackets) result = json.loads(attempt) if isinstance(result, dict): self.log.info(f"[JSON Fix] 成功修复 JSON (suffix={repr(suffix)})") return result except json.JSONDecodeError: continue return None async def _resume_approved_tool_batch( self, *, trace: Trace, history: List[Dict], goal_tree: Optional[GoalTree], config: RunConfig, sequence: int, head_seq: int, side_branch_ctx: Optional[SideBranchContext], runtime_tool_names: set[str], ) -> Tuple[List[Dict], int, int]: """Execute one fully-decided persisted approval batch exactly once.""" if not config.approval_batch_id: return history, sequence, head_seq if not self.trace_store or not hasattr( self.trace_store, "get_tool_approval_batch" ): raise RuntimeError("TraceStore does not support tool approvals") batch = await self.trace_store.get_tool_approval_batch(trace.trace_id) if batch is None or batch.batch_id != config.approval_batch_id: raise RuntimeError("approved tool batch was not found") if batch.status != "decided": raise RuntimeError(f"tool approval batch is not executable: {batch.status}") if any(call.decision == "pending" for call in batch.calls): raise RuntimeError("tool approval batch still has pending decisions") if any(call.execution_status == "executing" for call in batch.calls): batch.status = "execution_unknown" for call in batch.calls: if call.execution_status == "executing": call.execution_status = "execution_unknown" batch.updated_at = datetime.now().isoformat() await self.trace_store.replace_tool_approval_batch(trace.trace_id, batch) raise RuntimeError("tool execution outcome is unknown; refusing automatic retry") batch.status = "executing" batch.updated_at = datetime.now().isoformat() await self.trace_store.replace_tool_approval_batch(trace.trace_id, batch) current_goal_id = goal_tree.current_id if goal_tree and goal_tree.current_id else None trigger_event = None if side_branch_ctx and side_branch_ctx.type == "knowledge_eval": trigger_event = (trace.context.get("active_side_branch") or {}).get( "trigger_event", "unknown", ) for call in batch.calls: if call.execution_status in {"executed", "rejected"}: continue if call.decision == "rejected": tool_result: Any = json.dumps({ "status": "rejected", "error": "Tool call rejected by the local user", }, ensure_ascii=False) call.execution_status = "rejected" elif call.decision in {"approved", "auto_approved"}: call.execution_status = "executing" batch.updated_at = datetime.now().isoformat() await self.trace_store.replace_tool_approval_batch( trace.trace_id, batch, ) tool_result = await self.tools.execute( call.tool_name, call.effective_arguments, uid=config.uid or "", context=self._build_tool_context( config=config, trace=trace, trace_id=trace.trace_id, goal_id=current_goal_id, goal_tree=goal_tree, sequence=sequence, side_branch_ctx=side_branch_ctx, trigger_event=trigger_event, ), allowed_tool_names=runtime_tool_names, tool_call_id=call.tool_call_id, approval_grant=approval_grant(batch, call), ) else: raise RuntimeError(f"unsupported tool approval decision: {call.decision}") if isinstance(tool_result, str): normalized = {"text": tool_result} elif isinstance(tool_result, dict): normalized = tool_result else: normalized = {"text": str(tool_result)} tool_text = normalized.get("text", str(normalized)) tool_images = normalized.get("images", []) artifact_refs = normalized.get("artifact_refs", []) tool_content: Any = tool_text if tool_images: tool_content = [{"type": "text", "text": tool_text}] for image in tool_images: if image.get("type") == "base64" and image.get("data"): media_type = image.get("media_type", "image/png") tool_content.append({ "type": "image_url", "image_url": { "url": f"data:{media_type};base64,{image['data']}" }, }) elif image.get("type") == "url" and image.get("url"): tool_content.append({ "type": "image_url", "image_url": {"url": image["url"]}, }) tool_msg = Message.create( trace_id=trace.trace_id, role="tool", sequence=sequence, goal_id=current_goal_id, parent_sequence=head_seq, tool_call_id=call.tool_call_id, branch_type=side_branch_ctx.type if side_branch_ctx else None, branch_id=side_branch_ctx.branch_id if side_branch_ctx else None, content={ "tool_name": call.tool_name, "result": tool_content, "artifact_refs": artifact_refs, }, ) await self.trace_store.add_message(tool_msg) if tool_images and hasattr(self.trace_store, "write_message_attachment"): import base64 as b64mod for image in tool_images: if image.get("data"): await self.trace_store.write_message_attachment( trace.trace_id, tool_msg.message_id, suffix=".png", data=b64mod.b64decode(image["data"]), ) break call.result_message_id = tool_msg.message_id if call.execution_status != "rejected": call.execution_status = "executed" batch.updated_at = datetime.now().isoformat() await self.trace_store.replace_tool_approval_batch(trace.trace_id, batch) history.append({ "role": "tool", "tool_call_id": call.tool_call_id, "name": call.tool_name, "content": tool_content, "_message_id": tool_msg.message_id, }) head_seq = sequence sequence += 1 batch.status = "completed" batch.updated_at = datetime.now().isoformat() await self.trace_store.replace_tool_approval_batch(trace.trace_id, batch) await self.trace_store.update_trace( trace.trace_id, status="running", head_sequence=head_seq, ) config.approval_batch_id = None return history, sequence, head_seq async def _mark_approval_execution_unknown(self, trace_id: str) -> bool: """Fail closed if an approved batch stopped after execution began.""" if not self.trace_store or not hasattr( self.trace_store, "get_tool_approval_batch" ): return False batch = await self.trace_store.get_tool_approval_batch(trace_id) if batch is None or batch.status != "executing": return False batch.status = "execution_unknown" batch.updated_at = datetime.now().isoformat() for call in batch.calls: if call.execution_status == "executing": call.execution_status = "execution_unknown" await self.trace_store.replace_tool_approval_batch(trace_id, batch) await self.trace_store.update_trace( trace_id, status="failed", error_message=( "Tool execution outcome is unknown; automatic retry was refused" ), completed_at=datetime.now(), ) return True async def _agent_loop( self, trace: Trace, history: List[Dict], goal_tree: Optional[GoalTree], config: RunConfig, sequence: int, inject_skills: Optional[List[str]] = None, skill_recency_threshold: int = 10, ) -> AsyncIterator[Union[Trace, Message]]: """执行 Agent 的 ReAct 主循环。 ``run`` 在 Trace 准备后调用；Recursive 的预算、停止、工具门禁和根验收都在此串联。 """ trace_id = trace.trace_id runtime_tool_names = self._get_runtime_tool_names(config, trace) tool_schemas = self._get_runtime_tool_schemas( config, trace, runtime_tool_names=runtime_tool_names, ) completion_status = "completed" # 当前主路径头节点的 sequence（用于设置 parent_sequence） head_seq = trace.head_sequence # 侧分支状态（None = 主路径） side_branch_ctx: Optional[SideBranchContext] = None # 检查是否有未完成的侧分支需要恢复 if trace.context.get("active_side_branch"): side_branch_data = trace.context["active_side_branch"] branch_id = side_branch_data["branch_id"] start_sequence = side_branch_data["start_sequence"] # 只恢复同一持久分支；sequence 范围会把相邻分支混进来。 if self.trace_store: all_messages = await self.trace_store.get_trace_messages(trace_id) side_messages = [ m for m in all_messages if ( m.sequence >= start_sequence and m.branch_id == branch_id and m.branch_type == side_branch_data["type"] and m.status == "active" ) ] reconstructed_turns = sum( 1 for message in side_messages if message.role == "assistant" ) persisted_turns = int(side_branch_data.get("turns_used", 0)) turns_used = max(persisted_turns, reconstructed_turns) if turns_used != persisted_turns or "turns_used" not in side_branch_data: side_branch_data["turns_used"] = turns_used await self.trace_store.update_trace( trace_id, context=trace.context, ) # 恢复侧分支上下文 side_branch_ctx = SideBranchContext( type=side_branch_data["type"], branch_id=branch_id, start_head_seq=side_branch_data["start_head_seq"], start_sequence=side_branch_data["start_sequence"], start_history_length=0, # 稍后重新计算 start_iteration=side_branch_data.get("start_iteration", 0), max_turns=side_branch_data.get("max_turns", config.side_branch_max_turns), turns_used=int(turns_used), ) self.log.info( f"恢复未完成的侧分支: {side_branch_ctx.type}, " f"max_turns={side_branch_ctx.max_turns}" ) # 将侧分支消息追加到 history for m in side_messages: history.append(m.to_llm_dict()) # 重新计算 start_history_length side_branch_ctx.start_history_length = len(history) - len(side_messages) # The assistant Message is the durable turn record. A crash can # happen after that Message is stored but before Trace.context is # updated. If reconstruction shows the budget was already # exhausted, fail the branch closed instead of buying one extra # model turn after restart. if side_branch_ctx.turns_used >= side_branch_ctx.max_turns: self.log.warning( "恢复的侧分支 %s 已用尽轮次 %d/%d，直接返回主路径", side_branch_ctx.type, side_branch_ctx.turns_used, side_branch_ctx.max_turns, ) main_path_messages = await self.trace_store.get_main_path_messages( trace_id, side_branch_ctx.start_head_seq, ) history = [message.to_llm_dict() for message in main_path_messages] head_seq = side_branch_ctx.start_head_seq trace.context.pop("active_side_branch", None) if config.force_side_branch: if config.force_side_branch[0] == side_branch_ctx.type: config.force_side_branch.pop(0) if not config.force_side_branch: config.force_side_branch = None await self.trace_store.update_trace( trace_id, context=trace.context, head_sequence=head_seq, ) side_branch_ctx = None if config.approval_batch_id: history, sequence, head_seq = await self._resume_approved_tool_batch( trace=trace, history=history, goal_tree=goal_tree, config=config, sequence=sequence, head_seq=head_seq, side_branch_ctx=side_branch_ctx, runtime_tool_names=runtime_tool_names, ) break_after_side_branch = False # 侧分支退出后是否 break 主循环 for iteration in range(config.max_iterations): # 更新活动时间（表明trace正在活跃运行） if self.trace_store: await self.trace_store.update_trace( trace_id, last_activity_at=datetime.now() ) # 检查取消信号 cancel_event = self._cancel_events.get(trace_id) if cancel_event and cancel_event.is_set(): self.log.info(f"Trace {trace_id} stopped by user") trace_obj = await self._mark_trace_stopped(trace_id, head_seq) if trace_obj: yield trace_obj return # 检查Goal完成触发的知识评估 if not side_branch_ctx and self.trace_store: trace = await self.trace_store.get_trace(trace_id) if trace and trace.context and trace.context.get("pending_knowledge_eval"): # 清除标志 trace.context.pop("pending_knowledge_eval", None) await self.trace_store.update_trace(trace_id, context=trace.context) # 设置侧分支队列 config.force_side_branch = ["knowledge_eval"] self.log.info("[Knowledge Eval] 检测到Goal完成触发，进入知识评估侧分支") # Context 管理（仅主路径） needs_enter_side_branch = False if not side_branch_ctx: # 侧分支退出后需要 break 主循环 if break_after_side_branch and not config.force_side_branch: break # 检查是否强制进入侧分支（API 手动触发或自动压缩流程） if config.force_side_branch: needs_enter_side_branch = True self.log.info(f"强制进入侧分支: {config.force_side_branch}") else: # 正常的 context 管理逻辑 history, head_seq, sequence, needs_enter_side_branch = await self._manage_context_usage( trace_id, history, goal_tree, config, sequence, head_seq ) # 进入侧分支 if needs_enter_side_branch and not side_branch_ctx: # 刷新 trace，获取 _manage_context_usage 可能写入 DB 的 knowledge_eval_trigger if self.trace_store: fresh = await self.trace_store.get_trace(trace_id) if fresh: trace = fresh # 从队列中取出第一个侧分支类型 branch_type: Literal["compression", "reflection", "knowledge_eval"] if config.force_side_branch and isinstance(config.force_side_branch, list) and len(config.force_side_branch) > 0: branch_type = config.force_side_branch.pop(0) # type: ignore self.log.info(f"从队列取出侧分支: {branch_type}, 剩余队列: {config.force_side_branch}") elif config.knowledge.enable_extraction: # 兼容旧的单值模式（如果 force_side_branch 是字符串） branch_type = "reflection" else: # 自动触发：压缩 branch_type = "compression" branch_id = f"{branch_type}_{uuid.uuid4().hex[:8]}" side_branch_ctx = SideBranchContext( type=branch_type, branch_id=branch_id, start_head_seq=head_seq, start_sequence=sequence, start_history_length=len(history), start_iteration=iteration, max_turns=config.side_branch_max_turns, turns_used=0, ) # 持久化侧分支状态 if self.trace_store: # 获取触发事件（如果是 knowledge_eval 分支） trigger_event = trace.context.get("knowledge_eval_trigger", "unknown") if branch_type == "knowledge_eval" else None trace.context["active_side_branch"] = { "type": side_branch_ctx.type, "branch_id": side_branch_ctx.branch_id, "start_head_seq": side_branch_ctx.start_head_seq, "start_sequence": side_branch_ctx.start_sequence, "start_iteration": side_branch_ctx.start_iteration, "max_turns": side_branch_ctx.max_turns, "turns_used": side_branch_ctx.turns_used, "started_at": datetime.now().isoformat(), } # 如果是 knowledge_eval 分支，添加 trigger_event if trigger_event: trace.context["active_side_branch"]["trigger_event"] = trigger_event # 清除触发事件标记 trace.context.pop("knowledge_eval_trigger", None) await self.trace_store.update_trace( trace_id, context=trace.context ) # 追加侧分支 prompt if branch_type == "reflection": # 完成场景用全局复盘 prompt，压缩场景用阶段性反思 prompt if break_after_side_branch: prompt = config.knowledge.get_completion_reflect_prompt() else: prompt = config.knowledge.get_reflect_prompt() elif branch_type == "knowledge_eval": prompt = await self._build_knowledge_eval_prompt(trace_id, goal_tree) else: # compression from cyber_agent.trace.compaction import build_compression_prompt prompt = build_compression_prompt(goal_tree) branch_user_msg = Message.create( trace_id=trace_id, role="user", sequence=sequence, parent_sequence=head_seq, goal_id=goal_tree.current_id if goal_tree else None, branch_type=branch_type, branch_id=branch_id, content=prompt, ) if self.trace_store: await self.trace_store.add_message(branch_user_msg) history.append(branch_user_msg.to_llm_dict()) head_seq = sequence sequence += 1 self.log.info(f"进入侧分支: {branch_type}, branch_id={branch_id}") continue # 跳过本轮，下一轮开始侧分支 if self.trace_store: fresh_trace = await self.trace_store.get_trace(trace_id) if fresh_trace: trace = fresh_trace runtime_policy = policy_from_context(trace.context) runtime_tool_names = self._get_runtime_tool_names( config, trace, in_side_branch=side_branch_ctx is not None, ) tool_schemas = self._get_runtime_tool_schemas( config, trace, in_side_branch=side_branch_ctx is not None, runtime_tool_names=runtime_tool_names, ) dispatch_allowlist = ( runtime_tool_names if runtime_policy.mode is AgentMode.RECURSIVE else None ) # 构建 LLM messages（注入上下文，移除内部字段） llm_messages = [{k: v for k, v in msg.items() if not k.startswith("_")} for msg in history] # 优化已处理的图片（分级处理：保留/压缩/描述） llm_messages = await self._optimize_images( llm_messages, config.model, trace_id=trace_id, ) # 对历史消息应用 Prompt Caching llm_messages = self._add_cache_control( llm_messages, config.model, config.enable_prompt_caching ) # 调用 LLM（等待完成后再检查 cancel 信号，不中断正在进行的调用） result = await self.call_recursive_llm( trace_id, purpose="ordinary", messages=llm_messages, model=config.model, tools=tool_schemas, temperature=config.temperature, **config.extra_llm_params, ) if ( runtime_policy.mode is AgentMode.RECURSIVE and self.is_cancel_requested(trace_id) ): trace_obj = await self._mark_trace_stopped(trace_id, head_seq) if trace_obj: yield trace_obj return response_content = result.get("content", "") reasoning_content = result.get("reasoning_content", "") tool_calls = result.get("tool_calls") finish_reason = result.get("finish_reason") prompt_tokens = result.get("prompt_tokens", 0) completion_tokens = result.get("completion_tokens", 0) step_cost = result.get("cost", 0) cache_creation_tokens = result.get("cache_creation_tokens") cache_read_tokens = result.get("cache_read_tokens") budget_exceeded_dimension = result.get("_resource_budget_exceeded") if budget_exceeded_dimension: tool_calls = None finish_reason = "budget_exhausted" lifecycle_tools = { "agent", "submit_task_report", "review_task_result", "update_task_progress", } lifecycle_call_count = sum( 1 for tc in (tool_calls or []) if tc.get("function", {}).get("name") in lifecycle_tools ) protocol_batch_error = None if ( runtime_policy.requires_task_protocol and lifecycle_call_count and len(tool_calls or []) != 1 ): protocol_batch_error = ( "Recursive lifecycle tools must be the only tool call in an LLM turn" ) runtime_protocol_state = ( ensure_task_protocol(trace.context) if runtime_policy.requires_task_protocol else None ) protocol_lifecycle_required = bool( runtime_protocol_state and ( runtime_protocol_state["pending_reviews"] or runtime_protocol_state["next_actions"] ) ) # 周期性自动注入上下文（仅主路径） if ( not side_branch_ctx and not budget_exceeded_dimension and iteration % CONTEXT_INJECTION_INTERVAL == 0 ): # 检查是否已经调用了 get_current_context if tool_calls: has_context_call = any( tc.get("function", {}).get("name") == "get_current_context" for tc in tool_calls ) else: has_context_call = False tool_calls = [] if ( not has_context_call and not lifecycle_call_count and not protocol_lifecycle_required and "get_current_context" in runtime_tool_names ): # 手动添加 get_current_context 工具调用 context_call_id = f"call_context_{uuid.uuid4().hex[:8]}" tool_calls.append({ "id": context_call_id, "type": "function", "function": {"name": "get_current_context", "arguments": "{}"} }) self.log.info(f"[周期性注入] 自动添加 get_current_context 工具调用 (iteration={iteration})") # Skill 指定注入（仅主路径，首轮 iteration==0 时执行） if ( not side_branch_ctx and not budget_exceeded_dimension and inject_skills and iteration == 0 and not lifecycle_call_count and not protocol_lifecycle_required and "skill" in runtime_tool_names ): skills_to_inject = self._check_skills_need_injection( trace, inject_skills, history, skill_recency_threshold ) if skills_to_inject: if not tool_calls: tool_calls = [] for skill_name in skills_to_inject: skill_call_id = f"call_skill_{skill_name}_{uuid.uuid4().hex[:8]}" tool_calls.append({ "id": skill_call_id, "type": "function", "function": { "name": "skill", "arguments": json.dumps({"skill_name": skill_name}) } }) self.log.info(f"[Skill 指定注入] 自动添加 skill(\"{skill_name}\") 工具调用") # 按需自动创建 root goal（仅主路径） if not side_branch_ctx and goal_tree and not goal_tree.goals and tool_calls: has_goal_call = any( tc.get("function", {}).get("name") == "goal" for tc in tool_calls ) self.log.debug(f"[Auto Root Goal] Before tool execution: goal_tree.goals={len(goal_tree.goals)}, has_goal_call={has_goal_call}, tool_calls={[tc.get('function', {}).get('name') for tc in tool_calls]}") if not has_goal_call: mission = goal_tree.mission root_desc = mission[:200] if len(mission) > 200 else mission goal_tree.add_goals( descriptions=[root_desc], reasons=["系统自动创建：Agent 未显式创建目标"], parent_id=None ) if self.trace_store: await self.trace_store.add_goal(trace_id, goal_tree.goals[0]) await self.trace_store.update_goal_tree(trace_id, goal_tree) self.log.info(f"自动创建 root goal: {goal_tree.goals[0].id}（未自动 focus，等待模型决定）") else: self.log.debug(f"[Auto Root Goal] 检测到 goal 工具调用，跳过自动创建") # 获取当前 goal_id current_goal_id = goal_tree.current_id if (goal_tree and goal_tree.current_id) else None # 记录 assistant Message（parent_sequence 指向当前 head） assistant_msg = Message.create( trace_id=trace_id, role="assistant", sequence=sequence, goal_id=current_goal_id, parent_sequence=head_seq if head_seq > 0 else None, branch_type=side_branch_ctx.type if side_branch_ctx else None, branch_id=side_branch_ctx.branch_id if side_branch_ctx else None, content={"text": response_content, "tool_calls": tool_calls, "reasoning_content": reasoning_content or None}, prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, cache_creation_tokens=cache_creation_tokens, cache_read_tokens=cache_read_tokens, finish_reason=finish_reason, cost=step_cost, ) if self.trace_store: await self.trace_store.add_message(assistant_msg) # 记录模型使用 await self.trace_store.record_model_usage( trace_id=trace_id, sequence=sequence, role="assistant", model=config.model, prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, cache_read_tokens=cache_read_tokens or 0, ) # 知识评估侧分支：即时检测并写入评估结果 if side_branch_ctx and side_branch_ctx.type == "knowledge_eval": text = response_content if isinstance(response_content, str) else "" eval_results = None try: eval_results = json.loads(text.strip()) if "evaluations" not in eval_results: eval_results = None except json.JSONDecodeError: import re json_match = re.search(r'```json\s*(\{.*?\})\s*```', text, re.DOTALL) if json_match: try: eval_results = json.loads(json_match.group(1)) except json.JSONDecodeError: pass if not eval_results: json_match = re.search(r'\{[^{]*"evaluations"[^}]*\[[^\]]*\][^}]*\}', text, re.DOTALL) if json_match: try: eval_results = json.loads(json_match.group(0)) except json.JSONDecodeError: pass if eval_results and self.trace_store: current_trace = await self.trace_store.get_trace(trace_id) trigger_event = current_trace.context.get("active_side_branch", {}).get("trigger_event", "unknown") for eval_item in eval_results.get("evaluations", []): await self.trace_store.update_knowledge_evaluation( trace_id=trace_id, knowledge_id=eval_item["knowledge_id"], eval_result={ "eval_status": eval_item["eval_status"], "reason": eval_item.get("reason", "") }, trigger_event=trigger_event ) self.log.info(f"[Knowledge Eval] 已写入 {len(eval_results.get('evaluations', []))} 条评估结果") # 一轮以成功持久化的侧分支 assistant 回复为准；先落盘计数， # 再 yield，确保停止/重启不会重新获得已消费轮次。 if side_branch_ctx: side_branch_ctx.turns_used += 1 active_branch = trace.context.get("active_side_branch") if isinstance(active_branch, dict): active_branch["turns_used"] = side_branch_ctx.turns_used if self.trace_store: await self.trace_store.update_trace( trace_id, context=trace.context, ) yield assistant_msg head_seq = sequence sequence += 1 if budget_exceeded_dimension: completion_status = "failed" trace.context["termination_reason"] = ( f"budget_exhausted:{budget_exceeded_dimension}" ) if self.trace_store: await self.trace_store.update_trace( trace_id, context=trace.context, error_message=trace.context["termination_reason"], ) break # 检查侧分支是否应该退出 if side_branch_ctx: turns_in_branch = side_branch_ctx.turns_used should_exit = turns_in_branch >= side_branch_ctx.max_turns or not tool_calls if turns_in_branch >= side_branch_ctx.max_turns: self.log.warning( f"侧分支 {side_branch_ctx.type} 达到最大轮次 " f"{side_branch_ctx.max_turns}，强制退出" ) if should_exit and side_branch_ctx.type == "compression": # === 压缩侧分支退出（超时 + 正常完成统一处理）=== summary_text = "" # 1. 从当前回复提取 if response_content: if "[[SUMMARY]]" in response_content: summary_text = response_content[ response_content.index("[[SUMMARY]]") + len("[[SUMMARY]]"): ].strip() elif response_content.strip(): summary_text = response_content.strip() # 2. 从持久化存储按 sequence 范围查询 if not summary_text and self.trace_store: all_messages = await self.trace_store.get_trace_messages(trace_id) side_messages = [ m for m in all_messages if ( m.sequence >= side_branch_ctx.start_sequence and m.branch_id == side_branch_ctx.branch_id and m.branch_type == side_branch_ctx.type and m.status == "active" ) ] for msg in reversed(side_messages): if msg.role == "assistant" and isinstance(msg.content, dict): text = msg.content.get("text", "") if "[[SUMMARY]]" in text: summary_text = text[text.index("[[SUMMARY]]") + len("[[SUMMARY]]"):].strip() break elif text: summary_text = text break # 3. 单次 LLM 调用 if not summary_text: self.log.warning("侧分支未生成有效 summary，fallback 到单次 LLM 压缩") pre_branch_history = history[:side_branch_ctx.start_history_length] summary_text = await self._single_turn_compress( trace_id, pre_branch_history, goal_tree, config, ) # 创建主路径 summary 消息并重建 history if summary_text: # 清理侧分支指令，防止泄露到主分支 summary_text = summary_text.replace( "**生成摘要后立即停止，不要继续执行原有任务。**", "" ).strip() from cyber_agent.core.prompts import build_summary_header summary_content = build_summary_header(summary_text) if goal_tree and goal_tree.goals: goal_tree_detail = goal_tree.to_prompt(include_summary=True) summary_content += f"\n\n## Current Plan\n\n{goal_tree_detail}" # 找第一条 user message 的 sequence 作为 parent # 续跑时 get_main_path_messages 沿 parent 链回溯， # 指向 first_user 可以跳过所有被压缩的中间消息 first_user_seq = None if self.trace_store: all_msgs = await self.trace_store.get_trace_messages(trace_id) for m in all_msgs: if m.role == "user": first_user_seq = m.sequence break summary_msg = Message.create( trace_id=trace_id, role="user", sequence=sequence, parent_sequence=first_user_seq, branch_type=None, content=summary_content, ) if self.trace_store: await self.trace_store.add_message(summary_msg) history = self._rebuild_history_after_compression( history, summary_msg.to_llm_dict(), label="压缩侧分支" ) head_seq = sequence sequence += 1 else: self.log.error("所有压缩方案均未生成有效 summary，跳过压缩") # 回退 history 到侧分支开始前，防止侧分支指令泄露到主分支 history = history[:side_branch_ctx.start_history_length] head_seq = side_branch_ctx.start_head_seq # 清理 trace.context.pop("active_side_branch", None) config.force_side_branch = None if self.trace_store: await self.trace_store.update_trace( trace_id, context=trace.context, head_sequence=head_seq, ) side_branch_ctx = None continue elif should_exit and side_branch_ctx.type == "reflection": # === 反思侧分支退出（超时 + 正常完成统一处理）=== self.log.info("反思侧分支退出") # auto-commit hook：默认 pending 要等人工 review， # 但 reflect_auto_commit=True 时视作全部 approved，直接批量 upload。 if ( self.trace_store and getattr(config.knowledge, "reflect_auto_commit", False) ): try: from cyber_agent.trace.extraction_review import auto_commit_branch report = await auto_commit_branch( self.trace_store, trace_id, side_branch_ctx.branch_id, ) if report.committed or report.failed: self.log.info( f"[auto-commit] committed={len(report.committed)} " f"failed={len(report.failed)} skipped={len(report.skipped)}" ) except Exception as e: self.log.error(f"[auto-commit] 反思分支自动提交失败: {e}") # 恢复主路径 if self.trace_store: main_path_messages = await self.trace_store.get_main_path_messages( trace_id, side_branch_ctx.start_head_seq ) history = [m.to_llm_dict() for m in main_path_messages] head_seq = side_branch_ctx.start_head_seq # 清理 trace.context.pop("active_side_branch", None) if not config.force_side_branch or len(config.force_side_branch) == 0: config.force_side_branch = None self.log.info("反思完成，队列为空") if self.trace_store: await self.trace_store.update_trace( trace_id, context=trace.context, head_sequence=head_seq, ) side_branch_ctx = None continue elif should_exit and side_branch_ctx.type == "knowledge_eval": # === 知识评估侧分支退出 === self.log.info("知识评估侧分支退出") # 恢复主路径 if self.trace_store: main_path_messages = await self.trace_store.get_main_path_messages( trace_id, side_branch_ctx.start_head_seq ) history = [m.to_llm_dict() for m in main_path_messages] head_seq = side_branch_ctx.start_head_seq # 清理 trace.context.pop("active_side_branch", None) if not config.force_side_branch or len(config.force_side_branch) == 0: config.force_side_branch = None self.log.info("知识评估完成，队列为空") if self.trace_store: await self.trace_store.update_trace( trace_id, context=trace.context, head_sequence=head_seq, ) side_branch_ctx = None continue # 处理工具调用 # 截断兜底：finish_reason == "length" 说明响应被 max_tokens 截断， # tool call 参数很可能不完整，不应执行，改为提示模型分批操作 if tool_calls and finish_reason == "length": self.log.warning( "[Runner] 响应被 max_tokens 截断，跳过 %d 个不完整的 tool calls", len(tool_calls), ) truncation_hint = TRUNCATION_HINT history.append({ "role": "assistant", "content": response_content, "tool_calls": tool_calls, }) # 为每个被截断的 tool call 返回错误结果 for tc in tool_calls: history.append({ "role": "tool", "tool_call_id": tc["id"], "content": truncation_hint, }) continue if tool_calls and not config.approval_batch_id: needs_confirmation = ( not config.auto_execute_tools or self.tools.check_confirmation_required(tool_calls) ) if needs_confirmation: approval_calls: List[ToolApprovalCallV1] = [] for tool_call in tool_calls: tool_name = tool_call.get("function", {}).get("name", "") raw_arguments = tool_call.get("function", {}).get( "arguments", {} ) if isinstance(raw_arguments, str): try: parsed_arguments = ( json.loads(raw_arguments) if raw_arguments.strip() else {} ) except json.JSONDecodeError: parsed_arguments = {"_raw": raw_arguments} elif isinstance(raw_arguments, dict): parsed_arguments = dict(raw_arguments) else: parsed_arguments = {} policy = self.tools.get_runtime_policy(tool_name) requires_confirmation = ( not config.auto_execute_tools or bool(policy.get("requires_confirmation")) ) decision = ( "pending" if requires_confirmation else "auto_approved" ) call_id = str(tool_call.get("id") or "") approval_calls.append(ToolApprovalCallV1( tool_call_id=call_id, tool_name=tool_name, original_arguments=parsed_arguments, effective_arguments=parsed_arguments, argument_hash=tool_argument_hash( tool_call_id=call_id, tool_name=tool_name, arguments=parsed_arguments, ), editable_params=list(policy.get("editable_params", [])), requires_confirmation=requires_confirmation, decision=decision, )) batch = ToolApprovalBatchV1.create( trace_id=trace_id, assistant_message_id=assistant_msg.message_id, assistant_sequence=assistant_msg.sequence, calls=approval_calls, ) if not self.trace_store or not hasattr( self.trace_store, "replace_tool_approval_batch" ): raise RuntimeError( "persistent TraceStore tool approval support is required" ) await self.trace_store.replace_tool_approval_batch( trace_id, batch, ) await self.trace_store.update_trace( trace_id, status="waiting_confirmation", head_sequence=head_seq, ) trace.status = "waiting_confirmation" try: from cyber_agent.trace.websocket import broadcast_trace_status_changed await broadcast_trace_status_changed( trace_id, "waiting_confirmation", ) except Exception: pass yield trace return if tool_calls and config.auto_execute_tools: if ( runtime_policy.mode is AgentMode.RECURSIVE and self.is_cancel_requested(trace_id) ): trace_obj = await self._mark_trace_stopped(trace_id, head_seq) if trace_obj: yield trace_obj return history.append({ "role": "assistant", "content": response_content, "tool_calls": tool_calls, }) if config.parallel_tool_execution: # === 并发执行 === current_goal_id = goal_tree.current_id if (goal_tree and goal_tree.current_id) else None async def _execute_single_tool(tc: dict) -> tuple: tool_name = tc["function"]["name"] tool_args = tc["function"]["arguments"] if protocol_batch_error: return ( tc, {}, json.dumps({ "status": "failed", "error": protocol_batch_error, }, ensure_ascii=False), ) if isinstance(tool_args, str): if not tool_args.strip(): tool_args = {} else: try: tool_args = json.loads(tool_args) except json.JSONDecodeError: tool_args = self._try_fix_json(tool_args) if tool_args is None: self.log.warning(f"[Tool Call] JSON 解析失败: {tc['function']['arguments'][:200]}") tc["function"]["arguments"] = json.dumps({"_error": "JSON parse failed", "_raw": tc["function"]["arguments"][:200]}, ensure_ascii=False) return (tc, None, f"Error: 工具参数 JSON 格式错误，无法解析。原始参数: {tc['function']['arguments'][:200]}") elif tool_args is None: tool_args = {} args_str = json.dumps(tool_args, ensure_ascii=False) args_display = args_str[:100] + "..." if len(args_str) > 100 else args_str self.log.info(f"[Tool Call] {tool_name}({args_display})") trigger_event_for_tool = None if side_branch_ctx and side_branch_ctx.type == "knowledge_eval" and self.trace_store: current_trace = await self.trace_store.get_trace(trace_id) if current_trace: trigger_event_for_tool = current_trace.context.get("active_side_branch", {}).get("trigger_event", "unknown") if tool_name in ("toolhub_call", "toolhub_search", "toolhub_health"): try: from cyber_agent.tools.builtin.toolhub import set_trace_context set_trace_context(trace_id) except ImportError: pass try: tool_result = await self.tools.execute( tool_name, tool_args, uid=config.uid or "", context=self._build_tool_context( config=config, trace=trace, trace_id=trace_id, goal_id=current_goal_id, goal_tree=goal_tree, sequence=sequence, side_branch_ctx=side_branch_ctx, trigger_event=trigger_event_for_tool, ), allowed_tool_names=dispatch_allowlist, ) return (tc, tool_args, tool_result) except Exception as e: import traceback return (tc, tool_args, f"Error executing tool {tool_name}: {str(e)}\n{traceback.format_exc()}") tasks = [_execute_single_tool(tc) for tc in tool_calls] results = await asyncio.gather(*tasks) for res in results: tc, tool_args, tool_result = res tool_name = tc["function"]["name"] if tool_args is None: history.append({"role": "tool", "tool_call_id": tc["id"], "name": tool_name, "content": tool_result}) yield Message.create(trace_id=trace_id, role="tool", sequence=sequence, parent_sequence=head_seq, tool_call_id=tc["id"], content=tool_result) head_seq = sequence sequence += 1 continue if tool_name == "goal" and goal_tree: self.log.debug(f"[Goal Tool] After execution: goal_tree.goals={len(goal_tree.goals)}, current_id={goal_tree.current_id}") if tool_name == "upload_knowledge" and isinstance(tool_result, dict): self.log.info(f"[Knowledge Tracking] 知识已上传") if isinstance(tool_result, str): tool_result = {"text": tool_result} elif not isinstance(tool_result, dict): tool_result = {"text": str(tool_result)} tool_text = tool_result.get("text", str(tool_result)) tool_images = tool_result.get("images", []) tool_usage = tool_result.get("tool_usage") artifact_refs = tool_result.get("artifact_refs", []) if tool_images: tool_result_text = tool_text tool_content_for_llm = [{"type": "text", "text": tool_text}] for img in tool_images: if img.get("type") == "base64" and img.get("data"): media_type = img.get("media_type", "image/png") tool_content_for_llm.append({"type": "image_url", "image_url": {"url": f"data:{media_type};base64,{img['data']}"}}) elif img.get("type") == "url" and img.get("url"): tool_content_for_llm.append({"type": "image_url", "image_url": {"url": img["url"]}}) else: tool_result_text = tool_text tool_content_for_llm = tool_text tool_msg = Message.create(trace_id=trace_id, role="tool", sequence=sequence, goal_id=current_goal_id, parent_sequence=head_seq, tool_call_id=tc["id"], branch_type=side_branch_ctx.type if side_branch_ctx else None, branch_id=side_branch_ctx.branch_id if side_branch_ctx else None, content={"tool_name": tool_name, "result": tool_content_for_llm, "artifact_refs": artifact_refs}) if self.trace_store: await self.trace_store.add_message(tool_msg) if tool_usage: await self.trace_store.record_model_usage(trace_id=trace_id, sequence=sequence, role="tool", tool_name=tool_name, model=tool_usage.get("model"), prompt_tokens=tool_usage.get("prompt_tokens", 0), completion_tokens=tool_usage.get("completion_tokens", 0), cache_read_tokens=tool_usage.get("cache_read_tokens", 0)) await self.record_recursive_tool_usage( trace_id, tool_usage, ) if tool_images: import base64 as b64mod for img in tool_images: if img.get("data"): await self.trace_store.write_message_attachment( trace_id, tool_msg.message_id, suffix=".png", data=b64mod.b64decode(img["data"]), ) break yield tool_msg head_seq = sequence sequence += 1 history.append({"role": "tool", "tool_call_id": tc["id"], "name": tool_name, "content": tool_content_for_llm, "_message_id": tool_msg.message_id}) if tool_name == "skill" and tc["id"].startswith("call_skill_"): try: skill_args = json.loads(tc["function"]["arguments"]) if isinstance(tc["function"]["arguments"], str) else tc["function"]["arguments"] injected_skill_name = skill_args.get("skill_name", "") if injected_skill_name: await self._update_skill_injection_record(trace_id, trace, injected_skill_name, tool_msg.message_id, tool_msg.sequence) self.log.info(f"[Skill 指定注入] 已记录 {injected_skill_name} → msg={tool_msg.message_id}") except Exception as e: self.log.warning(f"[Skill 指定注入] 记录追踪失败: {e}") else: for tc in tool_calls: current_goal_id = goal_tree.current_id if (goal_tree and goal_tree.current_id) else None tool_name = tc["function"]["name"] tool_args = tc["function"]["arguments"] if isinstance(tool_args, str): if not tool_args.strip(): tool_args = {} else: try: tool_args = json.loads(tool_args) except json.JSONDecodeError: # 尝试修复常见的截断/格式问题 tool_args = self._try_fix_json(tool_args) if tool_args is None: self.log.warning(f"[Tool Call] JSON 解析失败，跳过工具调用 {tool_name}: {tc['function']['arguments'][:200]}") # 修复 history 中 assistant message 里的残缺 JSON， # 避免 Qwen API 拒绝 "function.arguments must be in JSON format" tc["function"]["arguments"] = json.dumps( {"_error": "JSON parse failed", "_raw": tc["function"]["arguments"][:200]}, ensure_ascii=False, ) history.append({ "role": "tool", "tool_call_id": tc["id"], "content": f"Error: 工具参数 JSON 格式错误，无法解析。请重新生成正确的 JSON 参数调用此工具。原始参数: {tc['function']['arguments'][:200]}", }) # 注意：这里不 yield Message，因为缺少必需参数会导致错误 # yield Message 应该由 trace_store 统一管理 continue elif tool_args is None: tool_args = {} # 记录工具调用（INFO 级别，显示参数） args_str = json.dumps(tool_args, ensure_ascii=False) args_display = args_str[:100] + "..." if len(args_str) > 100 else args_str self.log.info(f"[Tool Call] {tool_name}({args_display})") # 获取trigger_event（如果在knowledge_eval侧分支中） trigger_event_for_tool = None if side_branch_ctx and side_branch_ctx.type == "knowledge_eval" and self.trace_store: current_trace = await self.trace_store.get_trace(trace_id) if current_trace: trigger_event_for_tool = current_trace.context.get("active_side_branch", {}).get("trigger_event", "unknown") # 设置 trace_id 上下文供 toolhub 使用（图片保存到 outputs/{trace_id}/） if tool_name in ("toolhub_call", "toolhub_search", "toolhub_health"): try: from cyber_agent.tools.builtin.toolhub import set_trace_context set_trace_context(trace_id) except ImportError: pass if protocol_batch_error: tool_result = json.dumps({ "status": "failed", "error": protocol_batch_error, }, ensure_ascii=False) else: tool_result = await self.tools.execute( tool_name, tool_args, uid=config.uid or "", context=self._build_tool_context( config=config, trace=trace, trace_id=trace_id, goal_id=current_goal_id, goal_tree=goal_tree, sequence=sequence, side_branch_ctx=side_branch_ctx, trigger_event=trigger_event_for_tool, ), allowed_tool_names=dispatch_allowlist, ) # 如果是 goal 工具，记录执行后的状态 if tool_name == "goal" and goal_tree: self.log.debug(f"[Goal Tool] After execution: goal_tree.goals={len(goal_tree.goals)}, current_id={goal_tree.current_id}") # 跟踪上传的知识（通过 upload_knowledge） if tool_name == "upload_knowledge" and isinstance(tool_result, dict): metadata = tool_result.get("metadata", {}) # upload_knowledge 返回的是统计信息，不是单个 knowledge_id # 这里只记录上传动作，不跟踪具体 ID self.log.info(f"[Knowledge Tracking] 知识已上传到 Knowledge Manager") # --- 支持多模态工具反馈 --- # execute() 返回 dict{"text","images","tool_usage"} 或 str # 统一为dict格式 if isinstance(tool_result, str): tool_result = {"text": tool_result} tool_text = tool_result.get("text", str(tool_result)) tool_images = tool_result.get("images", []) tool_usage = tool_result.get("tool_usage") # 新增：提取tool_usage artifact_refs = tool_result.get("artifact_refs", []) # 处理多模态消息 if tool_images: tool_result_text = tool_text # 构建多模态消息格式 tool_content_for_llm = [{"type": "text", "text": tool_text}] for img in tool_images: if img.get("type") == "base64" and img.get("data"): media_type = img.get("media_type", "image/png") tool_content_for_llm.append({ "type": "image_url", "image_url": { "url": f"data:{media_type};base64,{img['data']}" } }) elif img.get("type") == "url" and img.get("url"): tool_content_for_llm.append({ "type": "image_url", "image_url": { "url": img["url"] } }) img_count = len(tool_content_for_llm) - 1 # 减去 text 块 print(f"[Runner] 多模态工具反馈: tool={tool_name}, images={img_count}, text_len={len(tool_result_text)}") else: tool_result_text = tool_text tool_content_for_llm = tool_text tool_msg = Message.create( trace_id=trace_id, role="tool", sequence=sequence, goal_id=current_goal_id, parent_sequence=head_seq, tool_call_id=tc["id"], branch_type=side_branch_ctx.type if side_branch_ctx else None, branch_id=side_branch_ctx.branch_id if side_branch_ctx else None, # 存储完整内容：有图片时保留 list（含 image_url），纯文本时存字符串 content={"tool_name": tool_name, "result": tool_content_for_llm, "artifact_refs": artifact_refs}, ) if self.trace_store: await self.trace_store.add_message(tool_msg) # 记录工具的模型使用 if tool_usage: await self.trace_store.record_model_usage( trace_id=trace_id, sequence=sequence, role="tool", tool_name=tool_name, model=tool_usage.get("model"), prompt_tokens=tool_usage.get("prompt_tokens", 0), completion_tokens=tool_usage.get("completion_tokens", 0), cache_read_tokens=tool_usage.get("cache_read_tokens", 0), ) await self.record_recursive_tool_usage( trace_id, tool_usage, ) # 截图单独存为同名 PNG 文件 if tool_images: import base64 as b64mod for img in tool_images: if img.get("data"): png_path = await self.trace_store.write_message_attachment( trace_id, tool_msg.message_id, suffix=".png", data=b64mod.b64decode(img["data"]), ) print(f"[Runner] 截图已保存: {png_path.name}") break # 只存第一张 # 如果在侧分支，tool_msg 已持久化（不需要额外维护） yield tool_msg head_seq = sequence sequence += 1 history.append({ "role": "tool", "tool_call_id": tc["id"], "name": tool_name, "content": tool_content_for_llm, "_message_id": tool_msg.message_id, }) # 更新 skill 注入追踪记录 if tool_name == "skill" and tc["id"].startswith("call_skill_"): try: skill_args = json.loads(tc["function"]["arguments"]) if isinstance(tc["function"]["arguments"], str) else tc["function"]["arguments"] injected_skill_name = skill_args.get("skill_name", "") if injected_skill_name: await self._update_skill_injection_record( trace_id, trace, injected_skill_name, tool_msg.message_id, tool_msg.sequence, ) self.log.info(f"[Skill 指定注入] 已记录 {injected_skill_name} → msg={tool_msg.message_id}") except Exception as e: self.log.warning(f"[Skill 指定注入] 记录追踪失败: {e}") # on_complete 模式：goal(done=...) 后立即压缩该 goal 的消息 if ( not side_branch_ctx and config.goal_compression == "on_complete" and self.trace_store and goal_tree ): has_goal_done = False for tc in tool_calls: if tc["function"]["name"] != "goal": continue try: raw = tc["function"]["arguments"] args = json.loads(raw) if isinstance(raw, str) and raw.strip() else {} except (json.JSONDecodeError, TypeError): args = {} if args.get("done") is not None: has_goal_done = True break if has_goal_done: main_path_msgs = await self.trace_store.get_main_path_messages( trace_id, head_seq ) compressed_msgs = compress_completed_goals(main_path_msgs, goal_tree) if len(compressed_msgs) < len(main_path_msgs): self.log.info( "on_complete 压缩: %d -> %d 条消息", len(main_path_msgs), len(compressed_msgs), ) history = [msg.to_llm_dict() for msg in compressed_msgs] continue # 继续循环 # 无工具调用 # 如果在侧分支中，已经在上面处理过了（不会走到这里） # 主路径无工具调用 → 任务完成，检查是否需要完成后反思或知识评估 if not side_branch_ctx and self.trace_store: fresh_trace = await self.trace_store.get_trace(trace_id) if fresh_trace: trace = fresh_trace policy = policy_from_context(trace.context) if policy.requires_task_protocol: state = ensure_task_protocol(trace.context) missing_report = bool( trace.parent_trace_id and state.get("task_report") is None ) pending_reviews = bool(state["pending_reviews"]) pending_actions = bool(state["next_actions"]) progress_error = ( task_progress_readiness_error(state) if policy.requires_task_progress and not trace.parent_trace_id else None ) if ( missing_report or pending_reviews or pending_actions or progress_error ): attempts = state["protocol_correction_attempts"] if attempts < 2: state["protocol_correction_attempts"] = attempts + 1 await self.trace_store.update_trace( trace_id, context=trace.context, ) if pending_reviews: required_action = ( "review every pending child report with " "review_task_result" ) elif pending_actions: required_action = ( "execute the approved next action with agent" ) elif missing_report: required_action = ( "submit a valid TaskReport with submit_task_report" ) else: required_action = ( "update TaskProgress to a ready_to_submit snapshot " f"({progress_error})" ) history.append({ "role": "user", "content": ( "Protocol gate: you cannot finish yet. You must " f"{required_action}." ), }) continue if missing_report: report = protocol_error_report( trace_id, "No valid TaskReport after two correction attempts", ) state["task_report"] = report.model_dump() state["task_report_submitted_at_sequence"] = sequence state["task_report_progress_revision"] = state.get( "task_progress_head_revision" ) completion_status = "failed" await self.trace_store.update_trace( trace_id, context=trace.context, error_message="Recursive task protocol gate failed", ) break # 检查是否有待评估的知识 if not side_branch_ctx and self.trace_store: pending = await self.trace_store.get_pending_knowledge_entries(trace_id) if pending: self.log.info(f"任务即将结束，但仍有 {len(pending)} 条知识未评估，强制触发评估") config.force_side_branch = ["knowledge_eval"] trace = await self.trace_store.get_trace(trace_id) if trace: trace.context["knowledge_eval_trigger"] = "task_completion" await self.trace_store.update_trace(trace_id, context=trace.context) continue if not side_branch_ctx and config.knowledge.enable_completion_extraction and not break_after_side_branch: config.force_side_branch = ["reflection"] break_after_side_branch = True self.log.info("任务完成，进入完成后反思侧分支") continue if ( not side_branch_ctx and self.trace_store and policy_from_context(trace.context).requires_task_protocol and not trace.parent_trace_id ): state = ensure_task_protocol(trace.context) if ( state["root_validation_attempts"] >= 2 ): raise ValueError( "Root task already used its two independent validation " "attempts; create a new trace" ) # The Validator must read the exact persisted main path that # produced this candidate, including real Tool Results. await self.trace_store.update_trace( trace_id, head_sequence=head_seq, ) trace.head_sequence = head_seq validation_run = await self.validate_recursive_trace( trace_id, scope="root", completion_criteria=require_root_task_anchor( trace.context ).completion_criteria, candidate_output=response_content, root_validator=True, ) refreshed_trace = await self.trace_store.get_trace(trace_id) if not refreshed_trace: raise ValueError(f"Trace not found after root validation: {trace_id}") trace = refreshed_trace state = ensure_task_protocol(trace.context) validation_cache = state.get("task_report_validation") or {} validation_record = { **validation_run.result.model_dump(), "validation_plan": validation_cache.get("validation_plan"), "validated_at_sequence": head_seq, } state["root_validation_history"].append(validation_record) state["root_validation_attempts"] += 1 state["root_validation_passed"] = ( validation_run.result.outcome == "passed" ) await self.trace_store.update_trace( trace_id, context=trace.context, ) if not state["root_validation_passed"]: if state["root_validation_attempts"] < 2: validation_feedback = { "role": "user", "content": ( "Root validation did not pass. Revise the current " "answer using this framework-owned ValidationResult:\n" + json.dumps( validation_run.result.model_dump(), ensure_ascii=False, ) ), } history.append(validation_feedback) feedback_msg = Message.from_llm_dict( validation_feedback, trace_id=trace_id, sequence=sequence, goal_id=(goal_tree.current_id if goal_tree else None), parent_sequence=head_seq, ) await self.trace_store.add_message(feedback_msg) head_seq = sequence sequence += 1 await self.trace_store.update_trace( trace_id, head_sequence=head_seq, ) trace.head_sequence = head_seq if goal_tree and goal_tree.current_id: goal = goal_tree.find(goal_tree.current_id) if goal: goal.status = "in_progress" await self.trace_store.update_goal_tree( trace_id, goal_tree, ) continue completion_status = "failed" await self.trace_store.update_trace( trace_id, error_message="Root task did not pass independent validation", ) break if ( policy_from_context(trace.context).mode is AgentMode.RECURSIVE and self.is_cancel_requested(trace_id) ): trace_obj = await self._mark_trace_stopped(trace_id, head_seq) if trace_obj: yield trace_obj return # max_iterations 等非正常退出也必须经过同一完成门禁。 if self.trace_store: fresh_trace = await self.trace_store.get_trace(trace_id) if fresh_trace: trace = fresh_trace policy = policy_from_context(trace.context) if policy.requires_task_protocol: state = ensure_task_protocol(trace.context) if trace.parent_trace_id and state.get("task_report") is None: report = protocol_error_report( trace_id, "Agent loop ended without a valid TaskReport", ) state["task_report"] = report.model_dump() state["task_report_submitted_at_sequence"] = sequence state["task_report_progress_revision"] = state.get( "task_progress_head_revision" ) completion_status = "failed" await self.trace_store.update_trace( trace_id, context=trace.context, error_message="Recursive task protocol gate failed", ) if state["pending_reviews"]: completion_status = "failed" if state["next_actions"]: completion_status = "failed" if ( not trace.parent_trace_id and not state.get("root_validation_passed") ): completion_status = "failed" # 清理 trace 相关的跟踪数据 self._context_warned.pop(trace_id, None) self._context_usage.pop(trace_id, None) self._saved_knowledge_ids.pop(trace_id, None) # 更新 head_sequence 并完成 Trace if self.trace_store: await self.trace_store.update_trace( trace_id, status=completion_status, head_sequence=head_seq, completed_at=datetime.now(), ) trace_obj = await self.trace_store.get_trace(trace_id) if trace_obj: yield trace_obj # ===== 压缩辅助方法 ===== def _rebuild_history_after_compression( self, history: List[Dict], summary_msg_dict: Dict, label: str = "压缩", ) -> List[Dict]: """ 压缩后重建 history：system prompt + 第一条 user message + summary Args: history: 压缩前的 history summary_msg_dict: summary 消息的 LLM dict label: 日志标签 Returns: 新的 history """ system_msg = None first_user_msg = None for msg in history: if msg.get("role") == "system" and not system_msg: system_msg = msg elif msg.get("role") == "user" and not first_user_msg: first_user_msg = msg if system_msg and first_user_msg: break new_history = [] if system_msg: new_history.append(system_msg) if first_user_msg: new_history.append(first_user_msg) new_history.append(summary_msg_dict) self.log.info(f"{label}完成: {len(history)} → {len(new_history)} 条消息") for idx, msg in enumerate(new_history): role = msg.get("role", "unknown") content = msg.get("content", "") if isinstance(content, str): preview = content elif isinstance(content, list): preview = f"[{len(content)} blocks]" else: preview = str(content) self.log.info(f" {label}后[{idx}] {role}: {preview}") return new_history # ===== 回溯（Rewind）===== async def _rewind( self, trace_id: str, after_sequence: int, goal_tree: Optional[GoalTree], ) -> int: """ 回溯 Trace：快照 GoalTree，重建干净树并设置 ``head_sequence``。 ``_prepare_existing_trace`` 判定为回溯时调用；Recursive 同时重建待审核和待重规划状态。 Returns: 下一个可用的 sequence 号 """ if not self.trace_store: raise ValueError("trace_store required for rewind") # 1. 加载所有 messages（用于 safe cutoff 和 max sequence） all_messages = await self.trace_store.get_trace_messages(trace_id) if not all_messages: return 1 # 2. 找到安全截断点（确保不截断在 tool_call 和 tool response 之间） cutoff = self._find_safe_cutoff(all_messages, after_sequence) # 3. 快照并重建 GoalTree if goal_tree: # 获取截断点消息的 created_at 作为时间界限 cutoff_msg = None for msg in all_messages: if msg.sequence == cutoff: cutoff_msg = msg break cutoff_time = cutoff_msg.created_at if cutoff_msg else datetime.now() # 快照到 events（含 head_sequence 供前端感知分支切换） await self.trace_store.append_event(trace_id, "rewind", { "after_sequence": cutoff, "head_sequence": cutoff, "goal_tree_snapshot": goal_tree.to_dict(), }) # 按时间重建干净的 GoalTree new_tree = goal_tree.rebuild_for_rewind(cutoff_time) await self.trace_store.update_goal_tree(trace_id, new_tree) # 更新内存中的引用 goal_tree.goals = new_tree.goals goal_tree.current_id = new_tree.current_id trace = await self.trace_store.get_trace(trace_id) if trace and policy_from_context(trace.context).requires_task_protocol: state = ensure_task_protocol(trace.context) submitted_at = state.get("task_report_submitted_at_sequence") if isinstance(submitted_at, int) and submitted_at > cutoff: if state.get("task_report"): state["report_history"].append(state["task_report"]) state["task_report"] = None state["task_report_submitted_at_sequence"] = None state["task_report_progress_revision"] = None state["task_report_validation"] = None validation_cache = state.get("task_report_validation") if ( isinstance(validation_cache, dict) and int(validation_cache.get("validated_at_sequence", 0) or 0) > cutoff ): state["task_report_validation"] = None state["pending_reviews"] = { child_id: entry for child_id, entry in state["pending_reviews"].items() if entry.get("received_at_sequence", 0) <= cutoff } reverted_reviews = [ review for review in state["reviews"] if review.get("reviewed_at_sequence", 0) > cutoff ] for review in reverted_reviews: entry = review.get("pending_review") child_id = review.get("child_trace_id") if ( child_id and isinstance(entry, dict) and entry.get("received_at_sequence", 0) <= cutoff ): state["pending_reviews"][child_id] = entry state["reviews"] = [ review for review in state["reviews"] if review.get("reviewed_at_sequence", 0) <= cutoff ] state["next_actions"] = [ action for action in state["next_actions"] if action.get("created_at_sequence", 0) <= cutoff ] while ( state.get("task_brief") is not None and state.get("task_brief_effective_at_sequence", 0) > cutoff and state.get("task_brief_history") ): previous = state["task_brief_history"].pop() state["task_brief"] = previous["task_brief"] state["task_brief_version"] = previous["version"] state["task_brief_effective_at_sequence"] = previous.get( "effective_at_sequence", 0, ) if policy_from_context(trace.context).requires_task_progress: rewind_task_progress(state, cutoff) state["pending_replans"] = rebuild_pending_replans(state) root_history = [ item for item in state.get("root_validation_history", []) if int( item.get( "validated_at_sequence", item.get("evaluated_at_sequence", 0), ) or 0 ) <= cutoff ] state["root_validation_history"] = root_history state["root_validation_attempts"] = len(root_history) state["root_validation_passed"] = bool( root_history and root_history[-1].get("outcome") == "passed" ) state["protocol_correction_attempts"] = 0 prune_context_access(trace.context, cutoff) await self.trace_store.update_trace(trace_id, context=trace.context) projected_statuses: dict[str, str] = {} for entry in state["pending_reviews"].values(): if entry.get("goal_id"): projected_statuses[entry["goal_id"]] = "pending_review" for action in state["next_actions"]: goal_id = action.get("goal_id") if goal_id and goal_id not in projected_statuses: projected_statuses[goal_id] = "in_progress" if goal_tree and projected_statuses: for goal_id, status in projected_statuses.items(): goal = goal_tree.find(goal_id) if goal: goal.status = status await self.trace_store.update_goal_tree(trace_id, goal_tree) # 4. 更新 head_sequence 到 rewind 点 await self.trace_store.update_trace(trace_id, head_sequence=cutoff) # 5. 返回 next sequence（全局递增，不复用） max_seq = max((m.sequence for m in all_messages), default=0) return max_seq + 1 def _find_safe_cutoff(self, messages: List[Message], after_sequence: int) -> int: """ 找到安全的截断点。 如果 after_sequence 指向一条带 tool_calls 的 assistant message， 则自动扩展到其所有对应的 tool response 之后。 """ cutoff = after_sequence # 找到 after_sequence 对应的 message target_msg = None for msg in messages: if msg.sequence == after_sequence: target_msg = msg break if not target_msg: return cutoff # 如果是 assistant 且有 tool_calls，找到所有对应的 tool responses if target_msg.role == "assistant": content = target_msg.content if isinstance(content, dict) and content.get("tool_calls"): tool_call_ids = set() for tc in content["tool_calls"]: if isinstance(tc, dict) and tc.get("id"): tool_call_ids.add(tc["id"]) # 找到这些 tool_call 对应的 tool messages for msg in messages: if (msg.role == "tool" and msg.tool_call_id and msg.tool_call_id in tool_call_ids): cutoff = max(cutoff, msg.sequence) return cutoff async def _heal_orphaned_tool_calls( self, messages: List[Message], trace_id: str, goal_tree: Optional[GoalTree], sequence: int, ) -> tuple: """ 检测并修复消息历史中的 orphaned tool_calls。 当 agent 被 stop/crash 中断时，可能有 assistant 的 tool_calls 没有对应的 tool results（包括多 tool_call 部分完成的情况）。直接发给 LLM 会导致 400。 修复策略：为每个缺失的 tool_result 插入合成的"中断通知"消息，而非裁剪。 - 普通工具：简短中断提示 - agent/evaluate：包含 sub_trace_id、执行统计、continue_from 指引 合成消息持久化到 store，确保幂等（下次续跑不再触发）。 Returns: (healed_messages, next_sequence) """ if not messages: return messages, sequence # 收集所有 tool_call IDs → (assistant_msg, tool_call_dict) tc_map: Dict[str, tuple] = {} result_ids: set = set() for msg in messages: if msg.role == "assistant": content = msg.content if isinstance(content, dict) and content.get("tool_calls"): for tc in content["tool_calls"]: tc_id = tc.get("id") if tc_id: tc_map[tc_id] = (msg, tc) elif msg.role == "tool" and msg.tool_call_id: result_ids.add(msg.tool_call_id) orphaned_ids = [tc_id for tc_id in tc_map if tc_id not in result_ids] if not orphaned_ids: return messages, sequence self.log.info( "检测到 %d 个 orphaned tool_calls，生成合成中断通知", len(orphaned_ids), ) healed = list(messages) head_seq = messages[-1].sequence for tc_id in orphaned_ids: assistant_msg, tc = tc_map[tc_id] tool_name = tc.get("function", {}).get("name", "unknown") if tool_name in ("agent", "evaluate"): result_text = self._build_agent_interrupted_result( tc, goal_tree, assistant_msg, ) else: result_text = build_tool_interrupted_message(tool_name) synthetic_msg = Message.create( trace_id=trace_id, role="tool", sequence=sequence, goal_id=assistant_msg.goal_id, parent_sequence=head_seq, tool_call_id=tc_id, content={"tool_name": tool_name, "result": result_text}, ) if self.trace_store: await self.trace_store.add_message(synthetic_msg) healed.append(synthetic_msg) head_seq = sequence sequence += 1 # 更新 trace head/last sequence if self.trace_store: await self.trace_store.update_trace( trace_id, head_sequence=head_seq, last_sequence=max(head_seq, sequence - 1), ) return healed, sequence def _build_agent_interrupted_result( self, tc: Dict, goal_tree: Optional[GoalTree], assistant_msg: Message, ) -> str: """为中断的 agent/evaluate 工具调用构建合成结果（对齐正常返回值格式）""" args_str = tc.get("function", {}).get("arguments", "{}") try: args = json.loads(args_str) if isinstance(args_str, str) else args_str except json.JSONDecodeError: args = {} task = args.get("task", "未知任务") if isinstance(task, list): task = "; ".join(task) tool_name = tc.get("function", {}).get("name", "agent") mode = "evaluate" if tool_name == "evaluate" else "delegate" # 从 goal_tree 查找 sub_trace 信息 sub_trace_id = None stats = None if goal_tree and assistant_msg.goal_id: goal = goal_tree.find(assistant_msg.goal_id) if goal and goal.sub_trace_ids: first = goal.sub_trace_ids[0] if isinstance(first, dict): sub_trace_id = first.get("trace_id") elif isinstance(first, str): sub_trace_id = first if goal.cumulative_stats: s = goal.cumulative_stats if s.message_count > 0: stats = { "message_count": s.message_count, "total_tokens": s.total_tokens, "total_cost": round(s.total_cost, 4), } result: Dict[str, Any] = { "mode": mode, "status": "interrupted", "summary": AGENT_INTERRUPTED_SUMMARY, "task": task, } if sub_trace_id: result["sub_trace_id"] = sub_trace_id result["hint"] = build_agent_continue_hint(sub_trace_id) if stats: result["stats"] = stats return json.dumps(result, ensure_ascii=False, indent=2) # ===== 上下文注入 ===== # ===== Skill 指定注入 ===== def _check_skills_need_injection( self, trace: Trace, inject_skills: List[str], history: List[Dict], recency_threshold: int, ) -> List[str]: """ 检查哪些 skill 需要注入。 通过 trace.context["injected_skills"] 中记录的 message_id 检查是否仍在当前 history 的最近 recency_threshold 条消息中。 Returns: 需要注入的 skill 名称列表 """ injected = (trace.context or {}).get("injected_skills", {}) # 收集 history 中最近 recency_threshold 条消息的 message_id recent_msgs = history[-recency_threshold:] if recency_threshold > 0 else [] recent_ids = set() for msg in recent_msgs: mid = msg.get("message_id") or msg.get("_message_id") if mid: recent_ids.add(mid) needs_inject = [] for skill_name in inject_skills: record = injected.get(skill_name) if not record: needs_inject.append(skill_name) continue if record.get("message_id") not in recent_ids: needs_inject.append(skill_name) return needs_inject async def _update_skill_injection_record( self, trace_id: str, trace: Trace, skill_name: str, message_id: str, sequence: int, ): """更新 trace.context 中的 skill 注入记录""" if not trace.context: trace.context = {} if "injected_skills" not in trace.context: trace.context["injected_skills"] = {} trace.context["injected_skills"][skill_name] = { "message_id": message_id, "sequence": sequence, } if self.trace_store: await self.trace_store.update_trace(trace_id, context=trace.context) # ===== 上下文注入 ===== def _build_context_injection( self, trace: Trace, goal_tree: Optional[GoalTree], ) -> str: """构建周期性注入的上下文（GoalTree + Active Collaborators + Focus 提醒 + IM 消息通知）""" from datetime import datetime parts = [f"## Current Time\n\n{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"] if trace and policy_from_context(trace.context).requires_task_protocol: parts.append(render_recursive_context(trace.context)) # GoalTree if goal_tree and goal_tree.goals: parts.append(f"## Current Plan\n\n{goal_tree.to_prompt()}") if goal_tree.current_id: # 检测 focus 在有子节点的父目标上：提醒模型 focus 到具体子目标 children = goal_tree.get_children(goal_tree.current_id) pending_children = [c for c in children if c.status in ("pending", "in_progress")] if pending_children: child_ids = ", ".join( goal_tree._generate_display_id(c) for c in pending_children[:3] ) parts.append( f"**提醒**：当前焦点在父目标上，建议用 `goal(focus=\"...\")` " f"切换到具体子目标（如 {child_ids}）再执行。" ) else: # 无焦点：提醒模型 focus parts.append( "**提醒**：当前没有焦点目标。请用 `goal(focus=\"...\")` 选择一个目标开始执行。" ) # Active Collaborators collaborators = trace.context.get("collaborators", []) if collaborators: lines = ["## Active Collaborators"] for c in collaborators: status_str = c.get("status", "unknown") ctype = c.get("type", "agent") summary = c.get("summary", "") name = c.get("name", "unnamed") lines.append(f"- {name} [{ctype}, {status_str}]: {summary}") parts.append("\n".join(lines)) # IM 消息通知（Research Agent） im_config = trace.context.get("im_config") if im_config: contact_id = im_config.get("contact_id") chat_id = im_config.get("chat_id") if contact_id and chat_id: # 尝试导入 IM 模块并检查通知 try: from cyber_agent.tools.builtin.im import chat as im_chat notification = im_chat._notifications.get((contact_id, chat_id)) if notification: count = notification.get("count", 0) senders = notification.get("from", []) senders_str = ", ".join(senders) parts.append( f"## IM 消息通知\n\n" f"你有 {count} 条新消息，来自: {senders_str}\n" f"使用 `im_receive_messages(contact_id=\"{contact_id}\", chat_id=\"{chat_id}\")` 查看消息内容。" ) else: parts.append("## IM 消息通知\n\n暂无新消息") except (ImportError, AttributeError): # IM 模块未加载或不可用 pass # Knowledge Manager 队列状态 km_queue_size = trace.context.get("km_queue_size") if km_queue_size is not None: current_sender = trace.context.get("current_sender", "unknown") if km_queue_size > 0: parts.append( f"## 消息队列状态\n\n" f"当前处理: {current_sender} 的消息\n" f"队列中还有 {km_queue_size} 条待处理消息" ) else: parts.append( f"## 消息队列状态\n\n" f"当前处理: {current_sender} 的消息\n" f"队列为空，处理完本条消息后将进入休眠" ) return "\n\n".join(parts) # ===== 辅助方法 ===== async def _optimize_images( self, messages: List[Dict], model: str, *, trace_id: Optional[str] = None, ) -> List[Dict]: """ 分级优化已处理的图片，节省 token 策略（基于图片距离最后一条 assistant 的"轮次"）： 1. 最近 1-2 轮：保留原图 2. 3-5 轮：降低分辨率和压缩（节省 token 但保留视觉信息） 3. 5 轮以上：调用小模型生成文本描述 + 保留 URL 处理结果会缓存，避免重复的 PIL 解码/编码和 LLM 调用。 Args: messages: 原始消息列表 model: 当前使用的模型（用于选择描述生成模型） Returns: 优化后的消息列表（深拷贝） """ if not messages: return messages # 找到最后一条 assistant message 的位置 last_assistant_idx = -1 for i in range(len(messages) - 1, -1, -1): if messages[i].get("role") == "assistant": last_assistant_idx = i break # 如果没有 assistant message，说明还没开始对话，不优化 if last_assistant_idx == -1: return messages # 统计从每个位置到最后一条 assistant 之间的 assistant 数量（作为"轮次"） assistant_count_after = [0] * len(messages) count = 0 for i in range(len(messages) - 1, -1, -1): assistant_count_after[i] = count if messages[i].get("role") == "assistant": count += 1 # 深拷贝避免修改原始数据 import copy import hashlib import asyncio import base64 as b64mod import httpx import mimetypes messages = copy.deepcopy(messages) # 预处理：将所有 HTTP(S) URL 图片下载并转为 base64 data URL # Qwen API 无法访问外部签名 URL（如 BFL、火山引擎 TOS），必须在本地转换 url_download_jobs = [] # [(msg_idx, block_idx, url)] for i, msg in enumerate(messages): if msg.get("role") != "tool": continue content = msg.get("content") if not isinstance(content, list): continue for block_idx, block in enumerate(content): if isinstance(block, dict) and block.get("type") == "image_url": url = block.get("image_url", {}).get("url", "") if url.startswith(("http://", "https://")): url_download_jobs.append((i, block_idx, url)) if url_download_jobs: async def _download_image_to_data_url(url: str) -> str | None: try: async with httpx.AsyncClient(timeout=60, trust_env=False) as client: resp = await client.get(url) resp.raise_for_status() ct = resp.headers.get("content-type", "").split(";")[0].strip() if not ct.startswith("image/"): ct = mimetypes.guess_type(url.split("?")[0])[0] or "image/png" b64 = b64mod.b64encode(resp.content).decode() return f"data:{ct};base64,{b64}" except Exception: return None results = await asyncio.gather( *[_download_image_to_data_url(url) for _, _, url in url_download_jobs], return_exceptions=True ) converted = 0 for (msg_idx, block_idx, original_url), result in zip(url_download_jobs, results): if isinstance(result, str) and result.startswith("data:"): messages[msg_idx]["content"][block_idx]["image_url"]["url"] = result converted += 1 if converted: self.log.info(f"[Image Optimization] URL→base64 预转换: {converted}/{len(url_download_jobs)} 张") # 统计优化情况 stats = {"kept": 0, "downscaled": 0, "described": 0, "cache_hit": 0} # 收集需要降分辨率或尺寸补齐的图片（用于并发处理） process_jobs = [] # [(msg_idx, block_idx, image_url, cache_key, max_size, cache_field)] # 第一遍：扫描并收集需要处理的图片 for i in range(last_assistant_idx): msg = messages[i] if msg.get("role") != "tool": continue content = msg.get("content") if not isinstance(content, list): continue rounds_ago = assistant_count_after[i] for block_idx, block in enumerate(content): if isinstance(block, dict) and block.get("type") == "image_url": image_url_obj = block.get("image_url", {}) image_url = image_url_obj.get("url", "") if image_url.startswith("data:"): cache_key = hashlib.md5(image_url[:200].encode()).hexdigest() else: cache_key = hashlib.md5(image_url.encode()).hexdigest() # 1-5 轮都需要检查尺寸 if rounds_ago <= 5: cached = self._image_opt_cache.get(cache_key, {}) cache_field = "pad_only" if rounds_ago <= 2 else "downscaled" if cache_field not in cached and image_url.startswith("data:"): max_size = None if rounds_ago <= 2 else 512 process_jobs.append((i, block_idx, image_url, cache_key, max_size, cache_field)) # 并发处理所有尺寸任务 if process_jobs: process_results = await asyncio.gather( *[self._process_image_size(url, max_size=ms) for _, _, url, _, ms, _ in process_jobs], return_exceptions=True ) for (_, _, _, cache_key, _, cache_field), result in zip(process_jobs, process_results): if not isinstance(result, Exception) and result is not None: self._image_opt_cache.setdefault(cache_key, {})[cache_field] = result # 第二遍：应用处理结果 for i in range(last_assistant_idx): msg = messages[i] if msg.get("role") != "tool": continue content = msg.get("content") if not isinstance(content, list): continue # 计算这条消息距离最后一条 assistant 的"轮次" rounds_ago = assistant_count_after[i] # 处理每个 content block new_content = [] for block in content: if isinstance(block, dict) and block.get("type") == "image_url": image_url_obj = block.get("image_url", {}) image_url = image_url_obj.get("url", "") # 生成缓存 key（URL 图片用 URL 本身，base64 用前 64 字符 hash） if image_url.startswith("data:"): cache_key = hashlib.md5(image_url[:200].encode()).hexdigest() else: cache_key = hashlib.md5(image_url.encode()).hexdigest() # 根据距离决定处理策略 if rounds_ago <= 2: # 最近 1-2 轮：只补齐过小图片，保留原分辨率 cached = self._image_opt_cache.get(cache_key, {}) if "pad_only" in cached: new_content.append({ "type": "image_url", "image_url": {"url": cached["pad_only"]} }) stats["kept"] += 1 stats["cache_hit"] += 1 elif image_url.startswith("data:"): processed = await self._process_image_size(image_url, max_size=None) if processed: self._image_opt_cache.setdefault(cache_key, {})["pad_only"] = processed new_content.append({ "type": "image_url", "image_url": {"url": processed} }) else: new_content.append(block) stats["kept"] += 1 else: new_content.append(block) stats["kept"] += 1 elif rounds_ago <= 5: # 3-5 轮：降低分辨率（优先从缓存取） cached = self._image_opt_cache.get(cache_key, {}) if "downscaled" in cached: new_content.append({ "type": "image_url", "image_url": {"url": cached["downscaled"]} }) stats["downscaled"] += 1 stats["cache_hit"] += 1 elif image_url.startswith("data:"): processed = await self._process_image_size(image_url, max_size=512) if processed: # 缓存结果 self._image_opt_cache.setdefault(cache_key, {})["downscaled"] = processed new_content.append({ "type": "image_url", "image_url": {"url": processed} }) stats["downscaled"] += 1 else: new_content.append(block) stats["kept"] += 1 else: # URL 图片：无法直接处理，保留原图 new_content.append(block) stats["kept"] += 1 else: # 5 轮以上：生成文本描述（优先从缓存取） cached = self._image_opt_cache.get(cache_key, {}) if "description" in cached: new_content.append(cached["description"]) stats["described"] += 1 stats["cache_hit"] += 1 else: description = await self._generate_image_description( image_url, model, trace_id=trace_id, ) url_info = f" (URL: {image_url[:100]}...)" if not image_url.startswith("data:") else "" desc_block = { "type": "text", "text": f"[Image description: {description}]{url_info}" } # 缓存结果 self._image_opt_cache.setdefault(cache_key, {})["description"] = desc_block new_content.append(desc_block) stats["described"] += 1 else: new_content.append(block) msg["content"] = new_content # print(f"[Image Opt Check] 扫描到 {stats['kept'] + stats['downscaled'] + stats['described']} 张图片上下文") if stats["downscaled"] > 0 or stats["described"] > 0: self.log.info( f"[Image Optimization] 保留 {stats['kept']} 张，" f"降分辨率 {stats['downscaled']} 张，" f"文本描述 {stats['described']} 张，" f"缓存命中 {stats['cache_hit']} 次" ) return messages async def _process_image_size(self, base64_url: str, max_size: Optional[int] = 512, min_size: int = 11) -> Optional[str]: """ 处理 base64 图片的尺寸： - 若 max_size 不为 None 且大于该值，则等比例缩放 - 若任意一边小于 min_size，则补充白边 (Padding) """ try: from PIL import Image import io import base64 # 解析 base64 数据 if not base64_url.startswith("data:"): return None header, data = base64_url.split(",", 1) media_type = header.split(";")[0].split(":")[1] # image/png # 解码图片 img_data = base64.b64decode(data) img = Image.open(io.BytesIO(img_data)) width, height = img.size needs_downscale = max_size is not None and (width > max_size or height > max_size) needs_pad = width < min_size or height < min_size # 尺寸正常，无需处理 if not needs_downscale and not needs_pad: return base64_url new_width, new_height = width, height # 1. 降分辨率 if needs_downscale: if width > height: new_width = max_size new_height = int(height * max_size / width) else: new_height = max_size new_width = int(width * max_size / height) if (new_width, new_height) != (width, height): img_resized = img.resize((new_width, new_height), Image.Resampling.BILINEAR) else: img_resized = img # 2. 补齐白边 (Padding) pad_width = max(new_width, min_size) pad_height = max(new_height, min_size) if pad_width > new_width or pad_height > new_height: # 创建白色背景 padded_img = Image.new("RGBA" if img_resized.mode in ("RGBA", "P") else "RGB", (pad_width, pad_height), (255, 255, 255, 255)) offset_x = (pad_width - new_width) // 2 offset_y = (pad_height - new_height) // 2 padded_img.paste(img_resized, (offset_x, offset_y)) img_resized = padded_img # 转换为 RGB（JPEG不支持 RGBA, P 等具有透明度或索引的模式） if img_resized.mode != "RGB": if img_resized.mode == "RGBA" or img_resized.mode == "P": # Create a white background for transparent images background = Image.new("RGB", img_resized.size, (255, 255, 255)) if img_resized.mode == "P" and "transparency" in img_resized.info: img_resized = img_resized.convert("RGBA") if img_resized.mode == "RGBA": background.paste(img_resized, mask=img_resized.split()[3]) img_resized = background img_resized = img_resized.convert("RGB") # 重新编码为 JPEG（如果只是补齐没有缩放，可以稍微保留高点质量） buffer = io.BytesIO() quality = 60 if needs_downscale else 85 img_resized.save(buffer, format="JPEG", quality=quality, optimize=False) new_data = base64.b64encode(buffer.getvalue()).decode("utf-8") return f"data:image/jpeg;base64,{new_data}" except Exception as e: self.log.warning(f"[Image Process] 处理图片尺寸失败: {e}") return None async def _generate_image_description( self, image_url: str, current_model: str, *, trace_id: Optional[str] = None, ) -> str: """ 使用小模型生成图片的文本描述 Args: image_url: 图片 URL（base64 或 http(s)） current_model: 当前使用的模型 Returns: 图片描述文本 """ try: # 使用 qwen-vl-max（通义千问视觉模型）生成描述 # 注意：qwen-vl 系列专门支持视觉输入 description_model = "qwen-vl-max" # 构建描述请求 messages = [ { "role": "user", "content": [ { "type": "image_url", "image_url": {"url": image_url} }, { "type": "text", "text": "请用 1-2 句话简洁描述这张图片的主要内容。" } ] } ] # 调用 LLM call_kwargs = { "messages": messages, "model": description_model, "tools": None, "temperature": 0.3, } result = ( await self.call_recursive_llm( trace_id, purpose="ordinary", fail_on_post_response_exhaustion=True, **call_kwargs, ) if trace_id else await self.llm_call(**call_kwargs) ) description = result.get("content", "").strip() return description if description else "图片内容" except (ResourceBudgetExceeded, ResourceBudgetStateError): raise except Exception as e: self.log.warning(f"[Image Description] 生成描述失败: {e}") return "图片内容" def _add_cache_control( self, messages: List[Dict], model: str, enable: bool ) -> List[Dict]: """ 为支持的模型添加 Prompt Caching 标记 策略：固定位置 + 延迟缓存 1. 如果有未处理的图片（最后一条 assistant 之后的 tool messages 中有图片），跳过缓存 2. system message 添加缓存（如果足够长） 3. 固定位置缓存点（20, 40, 60, 80），确保每个缓存点间隔 >= 1024 tokens 4. 最多使用 4 个缓存点（含 system） Args: messages: 原始消息列表 model: 模型名称 enable: 是否启用缓存 Returns: 添加了 cache_control 的消息列表（深拷贝） """ if not enable: return messages # 只对 Claude 模型启用 if "claude" not in model.lower(): return messages # 延迟缓存：检查是否有未处理的图片 last_assistant_idx = -1 for i in range(len(messages) - 1, -1, -1): if messages[i].get("role") == "assistant": last_assistant_idx = i break # 检查最后一条 assistant 之后是否有包含图片的 tool messages has_unprocessed_images = False if last_assistant_idx >= 0: for i in range(last_assistant_idx + 1, len(messages)): msg = messages[i] if msg.get("role") == "tool": content = msg.get("content") if isinstance(content, list): has_unprocessed_images = any( isinstance(block, dict) and block.get("type") == "image_url" for block in content ) if has_unprocessed_images: break if has_unprocessed_images: self.log.debug("[Cache] 检测到未处理的图片，延迟缓存建立") return messages # 深拷贝避免修改原始数据 import copy messages = copy.deepcopy(messages) # 策略 1: 为 system message 添加缓存 system_cached = False for msg in messages: if msg.get("role") == "system": content = msg.get("content", "") if isinstance(content, str) and len(content) > 1000: msg["content"] = [{ "type": "text", "text": content, "cache_control": {"type": "ephemeral"} }] system_cached = True self.log.debug(f"[Cache] 为 system message 添加缓存标记 (len={len(content)})") break # 策略 2: 固定位置缓存点 CACHE_INTERVAL = 20 MAX_POINTS = 3 if system_cached else 4 MIN_TOKENS = 1024 AVG_TOKENS_PER_MSG = 70 total_msgs = len(messages) if total_msgs == 0: return messages cache_positions = [] last_cache_pos = 0 for i in range(1, MAX_POINTS + 1): target_pos = i * CACHE_INTERVAL - 1 # 19, 39, 59, 79 if target_pos >= total_msgs: break # 从目标位置开始查找合适的 user/assistant 消息 for j in range(target_pos, total_msgs): msg = messages[j] if msg.get("role") not in ("user", "assistant"): continue content = msg.get("content", "") if not content: continue # 检查 content 是否非空 is_valid = False if isinstance(content, str): is_valid = len(content) > 0 elif isinstance(content, list): is_valid = any( isinstance(block, dict) and block.get("type") == "text" and len(block.get("text", "")) > 0 for block in content ) if not is_valid: continue # 检查 token 距离 msg_count = j - last_cache_pos estimated_tokens = msg_count * AVG_TOKENS_PER_MSG if estimated_tokens >= MIN_TOKENS: cache_positions.append(j) last_cache_pos = j self.log.debug(f"[Cache] 在位置 {j} 添加缓存点 (估算 {estimated_tokens} tokens)") break # 应用缓存标记 for idx in cache_positions: msg = messages[idx] content = msg.get("content", "") if isinstance(content, str): msg["content"] = [{ "type": "text", "text": content, "cache_control": {"type": "ephemeral"} }] self.log.debug(f"[Cache] 为 message[{idx}] ({msg.get('role')}) 添加缓存标记") elif isinstance(content, list): # 在最后一个 text block 添加 cache_control for block in reversed(content): if isinstance(block, dict) and block.get("type") == "text": block["cache_control"] = {"type": "ephemeral"} self.log.debug(f"[Cache] 为 message[{idx}] ({msg.get('role')}) 添加缓存标记") break self.log.debug( f"[Cache] 总消息: {total_msgs}, " f"缓存点: {len(cache_positions)} at {cache_positions}" ) return messages def _get_configured_tool_names( self, tools: Optional[List[str]] = None, tool_groups: Optional[List[str]] = None, exclude_tools: Optional[List[str]] = None, ) -> set[str]: """解析 RunConfig 的基础工具能力集合。""" if tools is not None: # Explicit tools are an exact capability set, not an addition to # the default groups. tool_names = set(tools) elif tool_groups is not None: tool_names = set(self.tools.get_tool_names(groups=tool_groups)) else: tool_names = set(self.tools.get_tool_names()) if exclude_tools: tool_names -= set(exclude_tools) return tool_names def _get_tool_schemas( self, tools: Optional[List[str]] = None, tool_groups: Optional[List[str]] = None, exclude_tools: Optional[List[str]] = None, ) -> List[Dict]: """获取 RunConfig 基础能力对应的工具 Schema。""" tool_names = self._get_configured_tool_names( tools, tool_groups, exclude_tools, ) return self.tools.get_schemas(list(tool_names)) def _get_runtime_tool_names( self, config: RunConfig, trace: Trace, *, in_side_branch: bool = False, ) -> set[str]: """在 RunConfig 基础能力上应用 Recursive 协议状态门禁。 主循环每轮调用，使待审核、待执行与报告阶段只暴露当前允许的工具。 """ tool_names = self._get_configured_tool_names( config.tools, config.tool_groups, config.exclude_tools, ) policy = policy_from_context(trace.context) protocol_tools = { "submit_task_report", "review_task_result", "update_task_progress", "read_context_ref", } if not policy.requires_task_protocol or in_side_branch: available = tool_names - protocol_tools if policy.requires_task_protocol: available.discard("evaluate") return available if self.application_binding is not None and trace.context.get( "application_ref" ): role = self.application_binding.role( trace.context.get("application_role_id") ) limits = trace.context.get("effective_run_limits") or {} depth = int(trace.context.get("agent_depth", 0) or 0) if ( not role.role.allowed_child_roles or depth >= int(limits.get("max_depth", policy.max_depth)) ): tool_names.discard("agent") state = ensure_task_protocol(trace.context) tool_names.discard("evaluate") if state["pending_reviews"]: return tool_names & {"review_task_result", "read_context_ref"} if state["next_actions"]: return tool_names & {"agent", "read_context_ref"} tool_names.discard("review_task_result") if not policy.requires_task_progress or state.get("task_report") is not None: tool_names.discard("update_task_progress") if not trace.parent_trace_id or state.get("task_report") is not None: tool_names.discard("submit_task_report") return tool_names def _get_runtime_tool_schemas( self, config: RunConfig, trace: Trace, *, in_side_branch: bool = False, runtime_tool_names: Optional[set[str]] = None, ) -> List[Dict]: """按 Trace 持久化模式和当前协议状态生成工具 Schema。 主循环把结果交给 LLM；Recursive revision 2 只允许受治理的本地 ``task_brief`` 委托，Legacy 和 Recursive revision 1 继续使用 ``task``。 """ tool_names = ( runtime_tool_names if runtime_tool_names is not None else self._get_runtime_tool_names( config, trace, in_side_branch=in_side_branch, ) ) schemas = deepcopy(self.tools.get_schemas(list(tool_names))) policy = policy_from_context(trace.context) for schema in schemas: function = schema.get("function", {}) if function.get("name") != "agent": continue parameters = function.get("parameters", {}) properties = parameters.get("properties", {}) required = parameters.setdefault("required", []) if policy.requires_task_protocol: properties.pop("messages", None) properties.pop("task", None) if "task_brief" in properties: properties["task_brief"]["description"] = ( "Required structured contract for Recursive local delegation. " "remote_* agents are not supported in Recursive revision 3." ) if "task" in required: required.remove("task") if "task_brief" in properties and "task_brief" not in required: required.append("task_brief") if self.application_binding is not None and trace.context.get( "application_ref" ): role = self.application_binding.role( trace.context.get("application_role_id") ) properties.pop("skills", None) if "skills" in required: required.remove("skills") if "agent_type" in properties: properties["agent_type"]["enum"] = list( role.role.allowed_child_roles ) properties["agent_type"]["description"] = ( "Required application role for the direct child." ) if "agent_type" not in required: required.append("agent_type") else: properties.pop("task_brief", None) if "task" in properties and "task" not in required: required.append("task") return schemas def _build_tool_context( self, *, config: RunConfig, trace: Trace, trace_id: str, goal_id: Optional[str], goal_tree: Optional[GoalTree], sequence: int, side_branch_ctx: Optional[SideBranchContext], trigger_event: Optional[str], ) -> Dict[str, Any]: """构建 ToolRegistry 执行时注入的隐藏上下文。 主循环在 dispatch 前调用；Recursive 权限快照和调度配置最后覆盖，不可伪造。 """ framework_context = { "store": self.trace_store, "task_protocol_service": self.task_protocol_service, "trace_id": trace_id, "goal_id": goal_id, "runner": self, "goal_tree": goal_tree, "knowledge_config": config.knowledge, "memory_config": config.memory, "dream_scope": ( DreamScope( uid=trace.uid, agent_type=trace.agent_type or config.agent_type, memory_identity=compute_memory_identity(config.memory), ) if config.memory is not None else None ), "sequence": sequence, "side_branch": { "type": side_branch_ctx.type, "branch_id": side_branch_ctx.branch_id, "is_side_branch": True, "max_turns": side_branch_ctx.max_turns, "trigger_event": trigger_event, } if side_branch_ctx else None, } if policy_from_context(trace.context).mode is AgentMode.RECURSIVE: context = {**(config.context or {}), **framework_context} context.update({ RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY: sorted( self._get_configured_tool_names( config.tools, config.tool_groups, config.exclude_tools, ) ), RECURSIVE_CHILD_EXECUTION_MODE_CONTEXT_KEY: config.child_execution_mode, RECURSIVE_MAX_PARALLEL_CHILDREN_CONTEXT_KEY: config.max_parallel_children, }) return context # Framework-owned hidden values are authoritative in every mode. A # caller-provided custom context may add page/business metadata, but it # cannot replace the store, Trace identity or per-run MemoryConfig. return {**(config.context or {}), **framework_context} # 默认 system prompt 前缀（当 config.system_prompt 和前端都未提供 system message 时使用） # 注意：此常量已迁移到 cyber_agent.core.prompts，这里保留引用以保持向后兼容 async def _build_system_prompt(self, config: RunConfig, base_prompt: Optional[str] = None) -> Optional[str]: """构建 system prompt（注入 skills） 优先级： 1. base_prompt（来自消息） 2. config.system_prompt（显式指定） 3. preset.system_prompt（预设的完整 system prompt） 4. 默认模板 + skills Skills 注入优先级： 1. config.skills 显式指定 → 按名称过滤 2. config.skills 为 None → 查 preset 的默认 skills 列表 3. preset 也无 skills（None）→ 加载全部（向后兼容） Args: base_prompt: 已有 system 内容（来自消息）， None 时使用 config.system_prompt 或 preset.system_prompt """ from cyber_agent.core.presets import AGENT_PRESETS # 确定 system_prompt 来源 if base_prompt is not None: system_prompt = base_prompt elif config.system_prompt is not None: system_prompt = config.system_prompt else: # 尝试从 preset 获取 system_prompt preset = AGENT_PRESETS.get(config.agent_type) system_prompt = preset.system_prompt if preset and preset.system_prompt else None # 确定要加载哪些 skills skills_filter: Optional[List[str]] = config.skills if skills_filter is None: preset = AGENT_PRESETS.get(config.agent_type) if preset is not None: skills_filter = preset.skills # 可能仍为 None（加载全部） # 加载并过滤 all_skills = load_skills_from_dir(self.skills_dir) if skills_filter is not None: skills = [s for s in all_skills if s.name in skills_filter] else: skills = all_skills skills_text = self._format_skills(skills) if skills else "" if system_prompt: if skills_text: system_prompt += f"\n\n## Skills\n{skills_text}" else: system_prompt = DEFAULT_SYSTEM_PREFIX if skills_text: system_prompt += f"\n\n## Skills\n{skills_text}" if config.max_iterations and config.max_iterations > 0: system_prompt += f"\n\n## Execution Constraint\n这是一项有严格步数限制的任务。你最多可以用 {config.max_iterations} 轮交互来解决问题。\n请务必【边查边写、随时存档】！每当你收集或得出一个有价值的独立结果（如收集到一个独立 Case），请立刻调用工具写入或追加到结果文件中，绝对不要等到所有任务都做完再最后一次性输出。这样即使触达步数上限被强制打断，你已经收集的成果也能安全保留！" # Memory 注入（memory-bearing Agent）——在 system prompt 末尾追加 # 初版选择 system prompt 追加（见 cyber_agent/docs/memory.md 待定问题 1）。 # 好处：run 启动一次性注入、所有后续轮次都能看到、与 skills 注入方式一致。 # 代价：若记忆文件很大会持续占 prompt tokens —— 待观察后决定是否切换方案。 if config.memory: try: from cyber_agent.core.memory import load_memory_files, format_memory_injection files = load_memory_files(config.memory) memory_text = format_memory_injection(files) if memory_text: system_prompt += f"\n\n{memory_text}" except Exception as e: self.log.warning(f"[Memory] 加载记忆失败，跳过注入: {e}") return system_prompt @staticmethod def _task_text(messages: List[Dict]) -> str: """Extract the same plain task text used by Legacy title generation.""" text_parts = [] for msg in messages: content = msg.get("content", "") if isinstance(content, str): text_parts.append(content) elif isinstance(content, list): for part in content: if isinstance(part, dict) and part.get("type") == "text": text_parts.append(part.get("text", "")) return " ".join(text_parts).strip() @classmethod def _fallback_task_name(cls, messages: List[Dict]) -> str: """Build a deterministic title without spending an untracked LLM call.""" raw_text = cls._task_text(messages) if not raw_text: return TASK_NAME_FALLBACK return raw_text[:50] + ("..." if len(raw_text) > 50 else "") async def _generate_task_name(self, messages: List[Dict]) -> str: """生成任务名称：优先使用 utility_llm，fallback 到文本截取""" fallback = self._fallback_task_name(messages) raw_text = self._task_text(messages) # 尝试使用 utility_llm 生成标题 if self.utility_llm_call: try: result = await self.utility_llm_call( messages=[ {"role": "system", "content": TASK_NAME_GENERATION_SYSTEM_PROMPT}, {"role": "user", "content": raw_text[:2000]}, ], model="gpt-4o-mini", # 使用便宜模型 ) title = result.get("content", "").strip() if title and len(title) < 100: return title except Exception: pass # Fallback: 截取前 50 字符 return fallback def _format_skills(self, skills: List[Skill]) -> str: if not skills: return "" return "\n\n".join(s.to_prompt_text() for s in skills)