(
self,
history: Sequence[str],
*,
states: Sequence[str] | None = None,
prompt_bundle: PromptBundle | None = None,
)
| 58 | self._llm = llm |
| 59 | |
| 60 | def optimize( |
| 61 | self, |
| 62 | history: Sequence[str], |
| 63 | *, |
| 64 | states: Sequence[str] | None = None, |
| 65 | prompt_bundle: PromptBundle | None = None, |
| 66 | ) -> PromptBundle: |
| 67 | target_states = [str(state) for state in (states or self.config.optimization.optimize_states)] |
| 68 | if not target_states: |
| 69 | return prompt_bundle or DEFAULT_PROMPTS.copy() |
| 70 | |
| 71 | bundle = prompt_bundle or DEFAULT_PROMPTS.copy() |
| 72 | grouped_history = collect_stage_history(history, target_states) |
| 73 | llm = self._llm or self._build_optimizer_model() |
| 74 | updates: dict[str, str] = {} |
| 75 | |
| 76 | for state in target_states: |
| 77 | if not hasattr(bundle, state): |
| 78 | continue |
| 79 | source_prompt = getattr(bundle, state) |
| 80 | special_block = _extract_special_block(source_prompt) |
| 81 | stage_history = grouped_history.get(state, []) |
| 82 | if not special_block or not stage_history: |
| 83 | continue |
| 84 | |
| 85 | message = bundle.optimize.format(source=special_block, history=stage_history) |
| 86 | response = llm.invoke(self._build_messages(message)) |
| 87 | content = str(getattr(response, "content", response)) |
| 88 | updates[state] = _replace_special_block(source_prompt, content) |
| 89 | |
| 90 | if not updates: |
| 91 | return bundle |
| 92 | return bundle.copy(**updates) |
| 93 | |
| 94 | @staticmethod |
| 95 | def _build_messages(message: str) -> list[Any]: |
no test coverage detected