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 base64 |
| #2 | import hashlib |
| #3 | import os |
| #4 | from pathlib import Path |
| #5 | |
| #6 | from openai import OpenAI |
| #7 | |
| #8 | from embedchain.helpers.json_serializable import register_deserializable |
| #9 | from embedchain.loaders.base_loader import BaseLoader |
| #10 | |
| #11 | DESCRIBE_IMAGE_PROMPT = "Describe the image:" |
| #12 | |
| #13 | |
| #14 | @register_deserializable |
| #15 | class ImageLoader(BaseLoader): |
| #16 | def __init__(self, max_tokens: int = 500, api_key: str = None, prompt: str = None): |
| #17 | super().__init__() |
| #18 | self.custom_prompt = prompt or DESCRIBE_IMAGE_PROMPT |
| #19 | self.max_tokens = max_tokens |
| #20 | self.api_key = api_key or os.environ["OPENAI_API_KEY"] |
| #21 | self.client = OpenAI(api_key=self.api_key) |
| #22 | |
| #23 | @staticmethod |
| #24 | def _encode_image(image_path: str): |
| #25 | with open(image_path, "rb") as image_file: |
| #26 | return base64.b64encode(image_file.read()).decode("utf-8") |
| #27 | |
| #28 | def _create_completion_request(self, content: str): |
| #29 | return self.client.chat.completions.create( |
| #30 | model="gpt-4o", messages=[{"role": "user", "content": content}], max_tokens=self.max_tokens |
| #31 | ) |
| #32 | |
| #33 | def _process_url(self, url: str): |
| #34 | if url.startswith("http"): |
| #35 | return [{"type": "text", "text": self.custom_prompt}, {"type": "image_url", "image_url": {"url": url}}] |
| #36 | elif Path(url).is_file(): |
| #37 | extension = Path(url).suffix.lstrip(".") |
| #38 | encoded_image = self._encode_image(url) |
| #39 | image_data = f"data:image/{extension};base64,{encoded_image}" |
| #40 | return [{"type": "text", "text": self.custom_prompt}, {"type": "image", "image_url": {"url": image_data}}] |
| #41 | else: |
| #42 | raise ValueError(f"Invalid URL or file path: {url}") |
| #43 | |
| #44 | def load_data(self, url: str): |
| #45 | content = self._process_url(url) |
| #46 | response = self._create_completion_request(content) |
| #47 | content = response.choices[0].message.content |
| #48 | |
| #49 | doc_id = hashlib.sha256((content + url).encode()).hexdigest() |
| #50 | return {"doc_id": doc_id, "data": [{"content": content, "meta_data": {"url": url, "type": "image"}}]} |
| #51 |