Prepare prompts, possibly set anchors, and return mode hints.
(
self,
raw_inputs:Dict[str,str],
spatial_info:Dict[str,Dict],
temporal_info:Dict[str,Dict],
mode: str = "default",
**kwargs,
)
| 28 | self.constraint = self.prompt_set.get_constraint(self.role) |
| 29 | |
| 30 | async def _process_inputs( |
| 31 | self, |
| 32 | raw_inputs:Dict[str,str], |
| 33 | spatial_info:Dict[str,Dict], |
| 34 | temporal_info:Dict[str,Dict], |
| 35 | mode: str = "default", |
| 36 | **kwargs, |
| 37 | )->Dict[str, Any]: |
| 38 | """Prepare prompts, possibly set anchors, and return mode hints.""" |
| 39 | if mode == "allow_kv_reuse": |
| 40 | request_uid = raw_inputs.get("_request_uid") or kwargs.get("request_uid") |
| 41 | if request_uid is None: |
| 42 | raise ValueError("request_uid is required for request-scoped anchor updates.") |
| 43 | |
| 44 | preferred_mode = "kv_reuse" |
| 45 | agent_memory = self.llm._ensure_agent_memory(self.id) |
| 46 | |
| 47 | if self.llm.has_prefix_initialized(self.id) and "placeholder_info" in agent_memory: |
| 48 | for agent_id, info in spatial_info.items(): |
| 49 | if info["role"] != "Programming Expert": |
| 50 | continue |
| 51 | answer = execute_code_get_return( |
| 52 | info["output"].lstrip("```python\n").rstrip("\n```") |
| 53 | ) |
| 54 | self.llm.update_condition_anchor( |
| 55 | request_uid=request_uid, |
| 56 | owner_agent_id=agent_id, |
| 57 | message=raw_inputs["task"], |
| 58 | content=f"The answer is:\n{answer}", |
| 59 | prefix_text="The answer is:\n", |
| 60 | ) |
| 61 | |
| 62 | prefix_text = kwargs.get("prefix", "Q:") |
| 63 | user_content = prefix_text + raw_inputs["task"] |
| 64 | preferred_mode = self.llm.update_input_anchor( |
| 65 | request_uid=request_uid, |
| 66 | agent_id=self.id, |
| 67 | message=raw_inputs["task"], |
| 68 | user_content=user_content, |
| 69 | prefix_text=prefix_text, |
| 70 | ) |
| 71 | logger.opt(colors=True).info( |
| 72 | "<green>[MODE]</green> Task: {} Agent {} ({}) mode: {}", |
| 73 | raw_inputs["task"], |
| 74 | self.id, |
| 75 | self.role, |
| 76 | preferred_mode, |
| 77 | ) |
| 78 | return {"preferred_mode": preferred_mode, "early_response": None} |
| 79 | |
| 80 | system_prompt = f"{self.constraint}" |
| 81 | user_input = "{user_question}" |
| 82 | user_prompt = self.prompt_set.get_answer_prompt(question=user_input, role=self.role) |
| 83 | spatial_str = "" |
| 84 | temporal_str = "" |
| 85 | for id, info in spatial_info.items(): |
| 86 | agent_output = info["output"] if len(info["output"]) > 0 else "{agent_" + id + "_current}" |
| 87 | if info["role"] == "Programming Expert": |
no test coverage detected