Math agent that aggregates peer signals and computes final answers.
| 10 | |
| 11 | @AgentRegistry.register('MathSolver') |
| 12 | class MathSolver(Node): |
| 13 | """Math agent that aggregates peer signals and computes final answers.""" |
| 14 | def __init__( |
| 15 | self, |
| 16 | id: str | None = None, |
| 17 | role: str = None, |
| 18 | domain: str = "", |
| 19 | llm_name: str = "", |
| 20 | llm_config: KVCommConfig | None = None, |
| 21 | ): |
| 22 | super().__init__(id, "MathSolver" ,domain, llm_name) |
| 23 | prefix = "A: " |
| 24 | self.llm = LLMRegistry.get(llm_name, prefix=prefix) |
| 25 | self.prompt_set = PromptSetRegistry.get(domain) |
| 26 | self.role = self.prompt_set.get_role() if role is None else role |
| 27 | self.llm.set_id(self.id, self.role) |
| 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, |
nothing calls this directly
no outgoing calls
no test coverage detected