_acall() — langchain Function Reference

Architecture documentation for the _acall() function in llms.py from the langchain codebase.

Function python LangChainCore LanguageModelBase calls 1

Entity Profile

LangChainCore→ LanguageModelBase→ _acall() — langchain Function Reference

Dependency Diagram

graph TD
  1b16bd3c_3c6a_1003_7ff4_901808dc90d7["_acall()"]
  4dfba76d_967f_bdbe_e65a_3ddf00dc8892["Fireworks"]
  1b16bd3c_3c6a_1003_7ff4_901808dc90d7 -->|defined in| 4dfba76d_967f_bdbe_e65a_3ddf00dc8892
  8d4f4229_b5de_0a63_174b_10061594c160["_format_output()"]
  1b16bd3c_3c6a_1003_7ff4_901808dc90d7 -->|calls| 8d4f4229_b5de_0a63_174b_10061594c160
  style 1b16bd3c_3c6a_1003_7ff4_901808dc90d7 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/partners/fireworks/langchain_fireworks/llms.py lines 177–233

    async def _acall(
        self,
        prompt: str,
        stop: list[str] | None = None,
        run_manager: AsyncCallbackManagerForLLMRun | None = None,
        **kwargs: Any,
    ) -> str:
        """Call Fireworks model to get predictions based on the prompt.

        Args:
            prompt: The prompt to pass into the model.
            stop: Optional list of strings to stop generation when encountered.
            run_manager: (Not used) Optional callback manager for async runs.
            kwargs: Additional parameters to pass to the model.

        Returns:
            The string generated by the model.

        """
        headers = {
            "Authorization": f"Bearer {self.fireworks_api_key.get_secret_value()}",
            "Content-Type": "application/json",
        }
        stop_to_use = stop[0] if stop and len(stop) == 1 else stop
        payload: dict[str, Any] = {
            **self.default_params,
            "prompt": prompt,
            "stop": stop_to_use,
            **kwargs,
        }

        # filter None values to not pass them to the http payload
        payload = {k: v for k, v in payload.items() if v is not None}
        async with (
            ClientSession() as session,
            session.post(
                self.base_url,
                json=payload,
                headers=headers,
                timeout=ClientTimeout(total=self.timeout),
            ) as response,
        ):
            if response.status >= 500:
                msg = f"Fireworks Server: Error {response.status}"
                raise Exception(msg)
            if response.status >= 400:
                msg = f"Fireworks received an invalid payload: {response.text}"
                raise ValueError(msg)
            if response.status != 200:
                msg = (
                    f"Fireworks returned an unexpected response with status "
                    f"{response.status}: {response.text}"
                )
                raise Exception(msg)

            response_json = await response.json()
            return self._format_output(response_json)