repositories
loading repo index
repositories
loading repo index
repository
loading code, commits, and activity
public Clawd ADK gateway launch mirror
stars
latest
clone command
git clone gitlawb://did:key:z6Mkq5mY...iFZ5/my-project-publ...git clone gitlawb://did:key:z6Mkq5mY.../my-project-publ...2fa351d6docs: add automaton and perps launch sources16d ago| #1 | import os |
| #2 | from typing import Dict, List, Optional |
| #3 | |
| #4 | from azure.identity import DefaultAzureCredential, get_bearer_token_provider |
| #5 | from openai import AzureOpenAI |
| #6 | |
| #7 | from mem0.configs.llms.base import BaseLlmConfig |
| #8 | from mem0.llms.base import LLMBase |
| #9 | |
| #10 | SCOPE = "https://cognitiveservices.azure.com/.default" |
| #11 | |
| #12 | |
| #13 | class AzureOpenAIStructuredLLM(LLMBase): |
| #14 | def __init__(self, config: Optional[BaseLlmConfig] = None): |
| #15 | super().__init__(config) |
| #16 | |
| #17 | # Model name should match the custom deployment name chosen for it. |
| #18 | if not self.config.model: |
| #19 | self.config.model = "gpt-4.1-nano-2025-04-14" |
| #20 | |
| #21 | api_key = self.config.azure_kwargs.api_key or os.getenv("LLM_AZURE_OPENAI_API_KEY") |
| #22 | azure_deployment = self.config.azure_kwargs.azure_deployment or os.getenv("LLM_AZURE_DEPLOYMENT") |
| #23 | azure_endpoint = self.config.azure_kwargs.azure_endpoint or os.getenv("LLM_AZURE_ENDPOINT") |
| #24 | api_version = self.config.azure_kwargs.api_version or os.getenv("LLM_AZURE_API_VERSION") |
| #25 | default_headers = self.config.azure_kwargs.default_headers |
| #26 | |
| #27 | # If the API key is not provided or is a placeholder, use DefaultAzureCredential. |
| #28 | if api_key is None or api_key == "" or api_key == "your-api-key": |
| #29 | self.credential = DefaultAzureCredential() |
| #30 | azure_ad_token_provider = get_bearer_token_provider( |
| #31 | self.credential, |
| #32 | SCOPE, |
| #33 | ) |
| #34 | api_key = None |
| #35 | else: |
| #36 | azure_ad_token_provider = None |
| #37 | |
| #38 | # Can display a warning if API version is of model and api-version |
| #39 | self.client = AzureOpenAI( |
| #40 | azure_deployment=azure_deployment, |
| #41 | azure_endpoint=azure_endpoint, |
| #42 | azure_ad_token_provider=azure_ad_token_provider, |
| #43 | api_version=api_version, |
| #44 | api_key=api_key, |
| #45 | http_client=self.config.http_client, |
| #46 | default_headers=default_headers, |
| #47 | ) |
| #48 | |
| #49 | def generate_response( |
| #50 | self, |
| #51 | messages: List[Dict[str, str]], |
| #52 | response_format: Optional[str] = None, |
| #53 | tools: Optional[List[Dict]] = None, |
| #54 | tool_choice: str = "auto", |
| #55 | ) -> str: |
| #56 | """ |
| #57 | Generate a response based on the given messages using Azure OpenAI. |
| #58 | |
| #59 | Args: |
| #60 | messages (List[Dict[str, str]]): A list of dictionaries, each containing a 'role' and 'content' key. |
| #61 | response_format (Optional[str]): The desired format of the response. Defaults to None. |
| #62 | |
| #63 | Returns: |
| #64 | str: The generated response. |
| #65 | """ |
| #66 | |
| #67 | user_prompt = messages[-1]["content"] |
| #68 | |
| #69 | user_prompt = user_prompt.replace("assistant", "ai") |
| #70 | |
| #71 | messages[-1]["content"] = user_prompt |
| #72 | |
| #73 | params = { |
| #74 | "model": self.config.model, |
| #75 | "messages": messages, |
| #76 | "temperature": self.config.temperature, |
| #77 | "max_tokens": self.config.max_tokens, |
| #78 | "top_p": self.config.top_p, |
| #79 | } |
| #80 | if response_format: |
| #81 | params["response_format"] = response_format |
| #82 | if tools: |
| #83 | params["tools"] = tools |
| #84 | params["tool_choice"] = tool_choice |
| #85 | |
| #86 | if tools: |
| #87 | params["tools"] = tools |
| #88 | params["tool_choice"] = tool_choice |
| #89 | |
| #90 | response = self.client.chat.completions.create(**params) |
| #91 | return self._parse_response(response, tools) |
| #92 |