Prepare prompts, run quick internal checks, and return mode hints.
(
self,
raw_inputs:Dict[str,str],
spatial_info:Dict[str,Dict],
temporal_info:Dict[str,Dict],
mode: str = "default",
**kwargs,
)
| 29 | self.llm.set_id(self.id, self.role) |
| 30 | |
| 31 | async def _process_inputs( |
| 32 | self, |
| 33 | raw_inputs:Dict[str,str], |
| 34 | spatial_info:Dict[str,Dict], |
| 35 | temporal_info:Dict[str,Dict], |
| 36 | mode: str = "default", |
| 37 | **kwargs, |
| 38 | )->Dict[str, Any]: |
| 39 | """Prepare prompts, run quick internal checks, and return mode hints.""" |
| 40 | if mode == "allow_kv_reuse": |
| 41 | request_uid = raw_inputs.get("_request_uid") or kwargs.get("request_uid") |
| 42 | if request_uid is None: |
| 43 | raise ValueError("request_uid is required for request-scoped anchor updates.") |
| 44 | |
| 45 | preferred_mode = "kv_reuse" |
| 46 | shared_memory = self.llm._ensure_agent_memory(self.id) |
| 47 | |
| 48 | if self.llm.has_prefix_initialized(self.id) and "placeholder_info" in shared_memory: |
| 49 | for agent_id, info in spatial_info.items(): |
| 50 | if ( |
| 51 | self.role != 'Normal Programmer' |
| 52 | and self.role != 'Stupid Programmer' |
| 53 | and info['role'] != 'Algorithm Designer' |
| 54 | ): |
| 55 | agent_mem = self.llm._ensure_agent_memory(agent_id) |
| 56 | if raw_inputs['task'] in agent_mem.get('condition', {}): |
| 57 | continue |
| 58 | code = info['output'].split("```python\n")[-1].split("\n```")[0] |
| 59 | is_solved, feedback, _ = PyExecutor().execute(code, self.internal_tests, timeout=1) |
| 60 | condition_text = ( |
| 61 | "Whether it passes internal testing?\n" |
| 62 | f"{is_solved}.\n\nThe feedback is:\n\n {feedback}." |
| 63 | ) |
| 64 | self.llm.update_condition_anchor( |
| 65 | request_uid=request_uid, |
| 66 | owner_agent_id=agent_id, |
| 67 | message=raw_inputs['task'], |
| 68 | content=condition_text, |
| 69 | prefix_text="Whether it passes internal testing?\n", |
| 70 | ) |
| 71 | |
| 72 | prefix_text = kwargs.get('prefix', "The task is:\n\n") |
| 73 | user_content = prefix_text + raw_inputs['task'] |
| 74 | preferred_mode = self.llm.update_input_anchor( |
| 75 | request_uid=request_uid, |
| 76 | agent_id=self.id, |
| 77 | message=raw_inputs['task'], |
| 78 | user_content=user_content, |
| 79 | prefix_text=prefix_text, |
| 80 | ) |
| 81 | logger.opt(colors=True).info( |
| 82 | "<green>[MODE]</green> Task: {} Agent {} ({}) mode: {}", |
| 83 | raw_inputs["task"], |
| 84 | self.id, |
| 85 | self.role, |
| 86 | preferred_mode, |
| 87 | ) |
| 88 | return {"preferred_mode": preferred_mode, "early_response": None} |
no test coverage detected