Prepare prompts, optionally populate anchors, and return mode hints.
(
self,
raw_inputs:Dict[str,str],
spatial_info:Dict[str,Dict],
temporal_info:Dict[str,Dict],
mode: str = "allow_kv_reuse",
**kwargs,
)
| 37 | self.constraint = self.prompt_set.get_analyze_constraint(self.role) |
| 38 | |
| 39 | async def _process_inputs( |
| 40 | self, |
| 41 | raw_inputs:Dict[str,str], |
| 42 | spatial_info:Dict[str,Dict], |
| 43 | temporal_info:Dict[str,Dict], |
| 44 | mode: str = "allow_kv_reuse", |
| 45 | **kwargs, |
| 46 | ) -> Dict[str, Any]: |
| 47 | """Prepare prompts, optionally populate anchors, and return mode hints.""" |
| 48 | if mode == "allow_kv_reuse": |
| 49 | request_uid = raw_inputs.get("_request_uid") or kwargs.get("request_uid") |
| 50 | if request_uid is None: |
| 51 | raise ValueError("request_uid is required for request-scoped anchor updates.") |
| 52 | |
| 53 | preferred_mode = "kv_reuse" |
| 54 | agent_memory = self.llm._ensure_agent_memory(self.id) |
| 55 | |
| 56 | if self.llm.has_prefix_initialized(self.id) and "placeholder_info" in agent_memory: |
| 57 | |
| 58 | prefix_text = kwargs.get('prefix', "The task is:") |
| 59 | user_content = prefix_text + raw_inputs['task'] |
| 60 | preferred_mode = self.llm.update_input_anchor( |
| 61 | request_uid=request_uid, |
| 62 | agent_id=self.id, |
| 63 | message=raw_inputs['task'], |
| 64 | user_content=user_content, |
| 65 | prefix_text=prefix_text, |
| 66 | test_time=True, |
| 67 | ) |
| 68 | logger.opt(colors=True).info( |
| 69 | "<green>[MODE]</green> Task: {} Agent {} ({}) mode: {}", |
| 70 | raw_inputs["task"], |
| 71 | self.id, |
| 72 | self.role, |
| 73 | preferred_mode, |
| 74 | ) |
| 75 | return {"preferred_mode": preferred_mode, "early_response": None} |
| 76 | |
| 77 | system_prompt = f"{self.constraint}" |
| 78 | user_input = "{user_question}" |
| 79 | user_prompt = f"The task is: {user_input}\n" |
| 80 | spatial_str = "" |
| 81 | temporal_str = "" |
| 82 | for agent_id, info in spatial_info.items(): |
| 83 | agent_role = info['role'] |
| 84 | agent_output = info['output'] if len(info['output']) > 0 else "{agent_" + agent_id + "_current}" |
| 85 | spatial_str += f"Agent {agent_id}, role is {agent_role}, output is:\n\n {agent_output}\n\n" |
| 86 | for agent_id, info in temporal_info.items(): |
| 87 | agent_role = info['role'] |
| 88 | agent_output = info['output'] if len(info['output']) > 0 else "{agent_" + agent_id + "_history}" |
| 89 | temporal_str += f"Agent {agent_id}, role is {agent_role}, output is:\n\n {agent_output}\n\n" |
| 90 | |
| 91 | if spatial_str: |
| 92 | user_prompt += ( |
| 93 | "At the same time, the outputs of other agents are as follows:\n\n" |
| 94 | f"{spatial_str} \n\n" |
| 95 | ) |
| 96 | if temporal_str: |
no test coverage detected