(self, texts: List[str])
| 800 | self.a0_model_conf = model_config |
| 801 | |
| 802 | def embed_documents(self, texts: List[str]) -> List[List[float]]: |
| 803 | configure_litellm() |
| 804 | # Apply rate limiting if configured |
| 805 | apply_rate_limiter_sync(self.a0_model_conf, " ".join(texts)) |
| 806 | |
| 807 | resp = embedding( |
| 808 | model=self.model_name, |
| 809 | input=texts, |
| 810 | **_merge_litellm_call_kwargs(self.kwargs), |
| 811 | ) |
| 812 | return [ |
| 813 | item.get("embedding") if isinstance(item, dict) else item.embedding # type: ignore |
| 814 | for item in resp.data # type: ignore |
| 815 | ] |
| 816 | |
| 817 | def embed_query(self, text: str) -> List[float]: |
| 818 | configure_litellm() |
nothing calls this directly
no test coverage detected