Capturing and Storing Real-World Optics With MongoDB Atlas, OpenAI GPT-4o, and PyMongo

1
{
2
  "mappings": {
3
    "dynamic": true,
4
    "fields": {
5
      "api_key": {
6
        "type": "string"
7
      },
8
      "ocr": {
9
        "dynamic": true,
10
        "type": "document"
11
      }
12
    }
13
  }
14
}

1
{
2
  "fields": [
3
    {
4
      "numDimensions": 1536,
5
      "path": "embedding",
6
      "similarity": "cosine",
7
      "type": "vector"
8
    },
9
    {
10
      "path": "api_key",
11
      "type": "filter"
12
    }
13
  ]
14
}

1
git clone https://github.com/Pash10g/allcr-ai.git
2
cd allcr-app
3
pip install -r requirements.txt

1
OPENAI_API_KEY=your_openai_api_key
2
MONGODB_ATLAS_URI=your_mongodb_atlas_uri

1
streamlit run app.py

1
# MongoDB connection
2
client = MongoClient(os.environ.get("MONGODB_ATLAS_URI"))
3
db = client['ocr_db']
4
collection = db['ocr_documents']

1
def auth_form():
2
    
3
4
    st.write("Please enter the API code to access the application.")
5
    api_code = st.text_input("API Code", type="password")
6
    if st.button("Submit"):
7
        st.toast("Authenticating...", icon="⚠️")
8
        db_api_key=auth_collection.find_one({"api_key":api_code})
9
        if db_api_key:
10
            st.session_state.authenticated = True
11
            st.session_state.api_code = api_code
12
            st.success("Authentication successful.")
13
            st.rerun()  # Re-run the script to remove the auth form
14
        else:
15
            st.error("Authentication failed. Please try again.")

1
# Function to transform image to text using OpenAI
2
def transform_image_to_text(image, format):
3
    img_byte_arr = io.BytesIO()
4
    image.save(img_byte_arr, format=format)
5
    img_byte_arr = img_byte_arr.getvalue()
6
    encoded_image = base64.b64encode(img_byte_arr).decode('utf-8')
7
8
    response = openai.chat.completions.create(
9
        model="gpt-4o",
10
        messages=[{
11
        "role": "system",
12
        "content": "You are an ocr to json expert looking to transcribe an image. If the type is 'other' please specify the type of object and clasiffy as you see fit."
13
        },
14
    {
15
      "role": "user",
16
      "content": [
17
        {
18
          "type": "text",
19
          "text": f"Please trunscribe this {transcribed_object} into a json only output for MongoDB store, calture all data as a single document. Always have a 'name', 'summary' (for embedding ) and 'type' top field (type is a subdocument with user and 'ai_classified') as well as other fields as you see fit."
20
        },
21
        {
22
          "type": "image_url",
23
          "image_url": {
24
            "url": f"data:image/jpeg;base64,{encoded_image}"
25
          }
26
        }
27
      ]
28
    }
29
  ]
30
    )
31
    extracted_text = response.choices[0].message.content
32
    return extracted_text

1
def clean_document(document):
2
    cleaned_document = document.strip().strip("```json").strip("```").strip()
3
    return json.loads(cleaned_document)
4
5
# Function to save image and text to MongoDB
6
def save_image_to_mongodb(image, description):
7
    img_byte_arr = io.BytesIO()
8
    image.save(img_byte_arr, format=image.format)
9
    img_byte_arr = img_byte_arr.getvalue()
10
    encoded_image = base64.b64encode(img_byte_arr).decode('utf-8')
11
    
12
    # Remove the ```json and ``` parts
13
    
14
15
    # Parse the cleaned JSON string into a Python dictionary
16
    document = clean_document(description)
17
18
    response = openai.embeddings.create(
19
    input=json.dumps({
20
        'name' : document['name'],
21
        'summary' : document['summary']
22
    }),
23
    model="text-embedding-3-small"
24
)
25
26
    gen_embeddings=response.data[0].embedding
27
28
    collection.insert_one({
29
        'image': encoded_image,
30
        'api_key': st.session_state.api_code,
31
        'embedding' : gen_embeddings,
32
        'ocr': document,
33
        'ai_tasks': []
34
    })

1
def search_aggregation(search_query):
2
    docs = list(collection.aggregate([
3
        {
4
            '$search': {
5
                'index': 'search', 
6
                'compound': {
7
                    'should': [
8
                        {
9
                            'text': {
10
                                'query': search_query, 
11
                                'path': {
12
                                    'wildcard': '*'
13
                                }
14
                            }
15
                        }
16
                    ], 
17
                    'filter': [
18
                        {
19
                            'queryString': {
20
                                'defaultPath': 'api_key', 
21
                                'query': st.session_state.api_code
22
                            }
23
                        }
24
                    ]
25
                }
26
            }
27
        }
28
    ]))
29
    return docs   
30
31
def vector_search_aggregation(search_query, limit):
32
    query_resp = openai.embeddings.create(
33
        input=search_query,
34
        model="text-embedding-3-small"
35
    )
36
    query_vec = query_resp.data[0].embedding
37
    docs = list(collection.aggregate([
38
        {
39
            '$vectorSearch': {
40
                'index': 'vector_index', 
41
                'queryVector': query_vec, 
42
                'path': 'embedding', 
43
                'numCandidates' : 20,
44
                'limit' : limit,
45
                'filter': {
46
                    'api_key': st.session_state.api_code
47
                }
48
            }},
49
            { '$project' : {'embedding'  : 0} }
50
    ]))
51
    return docs

1
def get_ai_task(ocr,prompt):
2
    ## Use existing document as context and perform another GPT task
3
    ocr_text = json.dumps(ocr)
4
    response = openai.chat.completions.create(
5
        model="gpt-4o",
6
        messages=[{
7
        "role": "system",
8
        "content": "You are  a task assistant looking to create a task for the AI to perform on the JSON object. Please return plain output which is only copy paste with no explanation."
9
        },
10
        {
11
        "role": "user",
12
        "content": f"Please perform the following task {prompt}  on the following JSON object {ocr_text}. Make sure that the output is stright forward to copy paste."
13
        }
14
        ]
15
        )
16
    
17
    return response.choices[0].message.content
18
19
def save_ai_task(task_id, task_result, prompt):
20
21
    collection.update_one(
22
        {"_id": ObjectId(task_id)},
23
        {"$push" : {"ai_tasks" : {'prompt' : prompt, 'result' : task_result}}}
24
    )
25
26
    return "Task saved successfully."