Настройка распознавания изображений в Dialogflow с RAG и OCR

import json

# Пример структуры для хранения изображений и описаний
knowledge_base = {
"images": [
{
"id": "img_001",
"description": "Диаграмма архитектуры микросервисов",
"tags": ["архитектура", "микросервисы", "diagram"],
"ocr_text": "Микросервисная архитектура..."
}
]
}

with open('knowledge_base.json', 'w') as f:
json.dump(knowledge_base, f)

from flask import Flask, request, jsonify
import google.cloud.dialogflow_v2 as dialogflow
import base64

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
req = request.get_json(silent=True, force=True)

# Извлечение информации о изображении из запроса
image_data = req.get('queryResult', {}).get('parameters', {}).get('image')

if image_data:
# Обработка изображения через RAG
response = process_image_with_rag(image_data)
else:
# Обработка текстового запроса
response = {"fulfillmentText": "Пожалуйста, предоставьте изображение для анализа"}

return jsonify(response)

def process_image_with_rag(image_data):
# Декодирование изображения
image_bytes = base64.b64decode(image_data)

# Извлечение текста с помощью OCR
ocr_result = extract_text_with_ocr(image_bytes)

# Анализ структуры документа
layout_result = analyze_layout(image_bytes)

# Поиск релевантных данных в базе знаний
relevant_info = search_knowledge_base(ocr_result, layout_result)

# Генерация ответа
response = generate_response(relevant_info)

return response

from google.cloud import vision
import io

def extract_text_with_google_vision(image_bytes):
 client = vision.ImageAnnotatorClient()
 image = vision.Image(content=image_bytes)
 
 # Распознавание текста
 response = client.text_detection(image=image)
 texts = response.text_annotations
 
 if response.error.message:
 raise Exception(f"{response.error.message}")
 
 # Извлечение распознанного текста
 extracted_text = []
 for text in texts:
 extracted_text.append(text.description)
 
 return extracted_text

import pytesseract
from PIL import Image

def extract_text_with_tesseract(image_bytes):
image = Image.open(io.BytesIO(image_bytes))
text = pytesseract.image_to_string(image)
return text

from azure.cognitiveservices.vision.computervision import ComputerVisionClient
from azure.cognitiveservices.vision.computervision.models import TextOperationStatusCodes
from azure.cognitiveservices.vision.computervision.models import TextRecognitionMode
from msrest.authentication import CognitiveServicesCredentials

def extract_text_with_azure(image_bytes):
credentials = CognitiveServicesCredentials('your-api-key')
client = ComputerVisionClient('endpoint', credentials)

result = client.recognize_printed_text_in_stream(io.BytesIO(image_bytes))

extracted_text = []
for region in result.regions:
for line in region.lines:
for word in line.words:
extracted_text.append(word.text)

return extracted_text

import layoutparser as lp
import cv2

# Загрузка модели для обнаружения текста
model = lp.Detectron2LayoutModel(
 'lp://PubLayNet/faster_rcnn_R_50_FPN_3x/config',
 extra_config=["MODEL.ROI_HEADS.SCORE_THRESH_TEST", 0.8],
 label_map={0: "Text", 1: "Title", 2: "List", 3: "Table", 4: "Figure"}
)

# Загрузка изображения
image = cv2.imread("document.png")
image = image[..., ::-1] # Конвертация BGR в RGB

# Анализ макета
layout = model.detect(image)

# Визуализация результатов
lp.draw_box(image, layout[0:3], box_width=3, show_element_type=True)

from layoutparser.io import load_image, load_layout
from layoutparser.ocr import TesseractAgent, GCVAgent

# Инициализация OCR движка
ocr_agent = TesseractAgent()

# Или использование Google Cloud Vision
# ocr_agent = GCVAgent(api_key="your-api-key")

# Загрузка изображения
image = load_image("document.png")

# Загрузка макета документа
layout = model.detect(image)

# Распознавание текста для каждого блока
for block in layout:
 if block.type == "Text":
 text = ocr_agent.detect(block)
 block.set(text=text, inplace=True)

# Вывод результатов
for block in layout:
 if block.type in ["Text", "Title"]:
 print(f"{block.type}: {block.text}")

# Создание иерархической структуры
text_blocks = [b for b in layout if b.type == "Text"]

# Группировка текстовых блоков по близости
from layoutparser.elements import Rectangle
from layoutparser.ocr.group_text_blocks import group_text_blocks

# Параметры группировки
grouped_blocks = group_text_blocks(
 text_blocks,
 x_tolerance=50, # Горизонтальный допуск в пикселях
 y_tolerance=10, # Вертикальный допуск в пикселях
)

# Вывод сгруппированных блоков
for i, group in enumerate(grouped_blocks):
 text = " ".join([block.text for block in group])
 print(f"Группа {i+1}: {text}")

# Пример некорректной структуры
{"image": "base64_encoded_image", "description": "Описание изображения"}

# Правильная структура для Dialogflow
{
 "queryResult": {
 "parameters": {
 "image": "base64_encoded_image",
 "context": "дополнительный контекст"
 }
 }
}

import base64

# Корректное кодирование изображения
with open("image.jpg", "rb") as image_file:
 encoded_string = base64.b64encode(image_file.read()).decode('utf-8')

# Некорректная обработка
try:
 encoded_string = base64.b64encode("image.jpg") # Ошибка - передается путь, а не бинарные данные
except Exception as e:
 print(f"Ошибка: {e}")

# Пример проблемы с обработкой контекста
def handle_jsonl_request(jsonl_data):
 try:
 # Разбор JSONL
 data = json.loads(jsonl_data)
 
 # Проверка наличия необходимых полей
 if 'image' not in data:
 raise ValueError("Отсутствует поле 'image'")
 
 # Обработка изображения
 image_result = process_image(data['image'])
 
 # Добавление контекста
 if 'context' in data:
 context = data['context']
 # Проблема: некорректная передача контекста в Dialogflow
 # Вместо этого нужно использовать механизм контекста Dialogflow
 result = add_context_to_response(image_result, context)
 else:
 result = image_result
 
 return result
 
 except json.JSONDecodeError as e:
 return {"error": f"Некорректный JSON: {str(e)}"}
 except Exception as e:
 return {"error": f"Ошибка обработки: {str(e)}"}

# Проверка размера данных
def validate_jsonl_size(jsonl_data, max_size_mb=10):
 size_bytes = len(jsonl_data.encode('utf-8'))
 max_size_bytes = max_size_mb * 1024 * 1024
 
 if size_bytes > max_size_bytes:
 raise ValueError(f"Размер данных ({size_bytes/1024/1024:.2f} MB) превышает лимит ({max_size_mb} MB)")
 
 return True

# Пример конфликта форматов
mixed_data = {
 "text": "Текст запроса",
 "image": "base64_encoded_image",
 "context": {"key": "value"},
 "metadata": {"timestamp": "2023-01-01T00:00:00Z"}
}

# Проблема: Dialogflow ожидает определенную структуру
# Решение: нормализация данных перед отправкой
normalized_data = normalize_for_dialogflow(mixed_data)

# Многоэтапная обработка изображений
class MultiStageImageProcessor:
 def __init__(self):
 self.ocr_engine = GoogleVisionOCR()
 self.layout_analyzer = LayoutParser()
 self.rag_system = RAGSystem()
 
 def process_image(self, image_data):
 # Этап 1: OCR распознавание
 ocr_result = self.ocr_engine.extract_text(image_data)
 
 # Этап 2: Анализ структуры
 layout_result = self.layout_analyzer.analyze(image_data)
 
 # Этап 3: Поиск в RAG системе
 rag_result = self.rag_system.search(ocr_result, layout_result)
 
 return {
 "ocr_text": ocr_result,
 "layout": layout_result,
 "rag_context": rag_result
 }

# Использование в Dialogflow
processor = MultiStageImageProcessor()

def webhook_handler(request):
 image_data = request.get_image_data()
 result = processor.process_image(image_data)
 
 # Отправка результатов в Dialogflow
 return create_dialogflow_response(result)

# Формат для мультимодальных данных
multimodal_data = {
 "image": {
 "format": "base64",
 "data": "base64_encoded_image",
 "metadata": {
 "width": 1920,
 "height": 1080,
 "format": "jpeg"
 }
 },
 "context": {
 "user_id": "user123",
 "session_id": "session456",
 "timestamp": "2023-01-01T00:00:00Z"
 },
 "parameters": {
 "analysis_type": "ocr",
 "language": "ru"
 }
}

# Сериализация в формат, совместимый с Dialogflow
def serialize_for_dialogflow(data):
 return {
 "queryResult": {
 "parameters": {
 "image": data["image"]["data"],
 "context": data["context"],
 "parameters": data["parameters"]
 }
 }
 }

# Промежуточный сервис обработки изображений
@app.route('/process-image', methods=['POST'])
def process_image_service():
 try:
 # Получение данных от клиента
 data = request.json
 
 # Извлечение изображения
 image_data = data.get('image')
 if not image_data:
 return jsonify({"error": "Изображение не предоставлено"}), 400
 
 # Обработка изображения
 processor = ImageProcessor()
 result = processor.process(image_data)
 
 # Сохранение результатов в базу данных
 db.save_result(result)
 
 # Возврат обработанных данных
 return jsonify({
 "status": "success",
 "result_id": result.id,
 "preview": result.preview
 })
 
 except Exception as e:
 return jsonify({"error": str(e)}), 500

# Вебхук Dialogflow
@app.route('/dialogflow-webhook', methods=['POST'])
def dialogflow_webhook():
 # Получение ID результата из запроса
 result_id = request.json.get('queryResult', {}).get('parameters', {}).get('result_id')
 
 if result_id:
 # Получение сохраненного результата
 result = db.get_result(result_id)
 return create_dialogflow_response(result)
 else:
 # Обработка нового изображения
 image_data = request.json.get('queryResult', {}).get('parameters', {}).get('image')
 if image_data:
 return jsonify({
 "fulfillmentText": "Изображение обрабатывается. Пожалуйста, подождите...",
 "outputContexts": [{
 "name": "image-processing",
 "parameters": {
 "image": image_data,
 "status": "processing"
 }
 }]
 })

# Потоковая обработка изображений
class StreamingImageProcessor:
 def __init__(self):
 self.chunk_size = 1024 * 1024 # 1MB chunks
 
 def process_stream(self, image_stream):
 # Инициализация процессоров
 ocr_processor = StreamingOCR()
 layout_processor = StreamingLayoutParser()
 
 # Потоковая обработка
 for chunk in image_stream:
 # OCR обработка
 ocr_result = ocr_processor.process_chunk(chunk)
 
 # Анализ структуры
 layout_result = layout_processor.process_chunk(chunk)
 
 # Комбинирование результатов
 yield combine_results(ocr_result, layout_result)
 
 def create_dialogflow_response(self, results):
 # Агрегация результатов для Dialogflow
 aggregated = {
 "text": combine_text_results(results),
 "layout": combine_layout_results(results),
 "confidence": calculate_confidence(results)
 }
 
 return {
 "fulfillmentText": format_response(aggregated),
 "outputContexts": [{
 "name": "image-processing-complete",
 "parameters": aggregated
 }]
 }

import asyncio

# Асинхронный обработчик изображений
class AsyncImageProcessor:
 async def process_image_async(self, image_data):
 # Параллельная обработка
 ocr_task = asyncio.create_task(self.ocr_process(image_data))
 layout_task = asyncio.create_task(self.layout_process(image_data))
 rag_task = asyncio.create_task(self.rag_process(image_data))
 
 # Ожидание результатов
 ocr_result, layout_result, rag_result = await asyncio.gather(
 ocr_task, layout_task, rag_task
 )
 
 return {
 "ocr": ocr_result,
 "layout": layout_result,
 "rag": rag_result
 }
 
 async def ocr_process(self, image_data):
 # Асинхронная OCR обработка
 loop = asyncio.get_event_loop()
 return await loop.run_in_executor(None, self.ocr_engine.extract_text, image_data)
 
 async def layout_process(self, image_data):
 # Асинхронный анализ структуры
 loop = asyncio.get_event_loop()
 return await loop.run_in_executor(None, self.layout_analyzer.analyze, image_data)
 
 async def rag_process(self, image_data):
 # Асинхронный поиск в RAG
 ocr_text = await self.ocr_process(image_data)
 return await self.rag_system.search_async(ocr_text)

# Вебхук с асинхронной обработкой
@app.route('/async-webhook', methods=['POST'])
async def async_webhook():
 image_data = request.json.get('queryResult', {}).get('parameters', {}).get('image')
 
 if image_data:
 processor = AsyncImageProcessor()
 result = await processor.process_image_async(image_data)
 
 return {
 "fulfillmentText": "Обработка изображения завершена",
 "outputContexts": [{
 "name": "async-processing-complete",
 "parameters": result
 }]
 }
 else:
 return jsonify({"fulfillmentText": "Изображение не предоставлено"})

import cv2
import numpy as np

def optimize_image_for_ocr(image_bytes):
 # Загрузка изображения
 image = cv2.imdecode(np.frombuffer(image_bytes, np.uint8), cv2.IMREAD_COLOR)
 
 # Улучшение качества
 # Увеличение разрешения
 height, width = image.shape[:2]
 if width < 1000:
 scale = 1000 / width
 image = cv2.resize(image, None, fx=scale, fy=scale, interpolation=cv2.INTER_CUBIC)
 
 # Удаление шума
 image = cv2.fastNlMeansDenoisingColored(image, None, 10, 10, 7, 21)
 
 # Улучшение контрастности
 lab = cv2.cvtColor(image, cv2.COLOR_BGR2LAB)
 l, a, b = cv2.split(lab)
 clahe = cv2.createCLAHE(clipLimit=3.0, tileGridSize=(8, 8))
 cl = clahe.apply(l)
 limg = cv2.merge((cl, a, b))
 final = cv2.cvtColor(limg, cv2.COLOR_LAB2BGR)
 
 # Конвертация обратно в байты
 _, buffer = cv2.imencode('.jpg', final)
 return buffer.tobytes()

import hashlib
from functools import lru_cache

class ImageRecognitionCache:
 def __init__(self, max_size=1000):
 self.cache = {}
 self.max_size = max_size
 
 def get_cache_key(self, image_data):
 # Создание уникального ключа на основе изображения
 return hashlib.md5(image_data).hexdigest()
 
 def get(self, image_data):
 key = self.get_cache_key(image_data)
 return self.cache.get(key)
 
 def set(self, image_data, result):
 key = self.get_cache_key(image_data)
 if len(self.cache) >= self.max_size:
 # Удаление самого старого элемента
 oldest_key = next(iter(self.cache))
 del self.cache[oldest_key]
 
 self.cache[key] = result
 
 def clear(self):
 self.cache.clear()

# Использование кэша
image_cache = ImageRecognitionCache()

def process_image_with_cache(image_data):
 # Проверка кэша
 cached_result = image_cache.get(image_data)
 if cached_result:
 return cached_result
 
 # Обработка изображения
 result = process_image(image_data)
 
 # Сохранение в кэш
 image_cache.set(image_data, result)
 
 return result

import logging
from datetime import datetime

# Настройка логирования
logging.basicConfig(
 level=logging.INFO,
 format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
 handlers=[
 logging.FileHandler('image_recognition.log'),
 logging.StreamHandler()
 ]
)

logger = logging.getLogger('image_recognition')

class ImageRecognitionMonitor:
 def __init__(self):
 self.stats = {
 'total_requests': 0,
 'successful_recognitions': 0,
 'failed_recognitions': 0,
 'average_processing_time': 0,
 'start_time': datetime.now()
 }
 
 def log_request(self, image_size, processing_time, success):
 self.stats['total_requests'] += 1
 
 if success:
 self.stats['successful_recognitions'] += 1
 else:
 self.stats['failed_recognitions'] += 1
 
 # Обновление среднего времени обработки
 current_avg = self.stats['average_processing_time']
 new_avg = (current_avg * (self.stats['total_requests'] - 1) + processing_time) / self.stats['total_requests']
 self.stats['average_processing_time'] = new_avg
 
 # Логирование
 logger.info(f"Обработка изображения размером {image_size} заняла {processing_time:.2f}с. "
 f"Успешно: {success}. "
 f"Статистика: {self.get_stats_summary()}")
 
 def get_stats_summary(self):
 uptime = datetime.now() - self.stats['start_time']
 success_rate = (self.stats['successful_recognitions'] / self.stats['total_requests'] * 100) if self.stats['total_requests'] > 0 else 0
 
 return {
 'uptime': str(uptime),
 'total_requests': self.stats['total_requests'],
 'success_rate': f"{success_rate:.1f}%",
 'avg_processing_time': f"{self.stats['average_processing_time']:.2f}s"
 }

# Использование монитора
monitor = ImageRecognitionMonitor()

def monitored_image_processing(image_data):
 start_time = datetime.now()
 image_size = len(image_data)
 
 try:
 result = process_image(image_data)
 processing_time = (datetime.now() - start_time).total_seconds()
 monitor.log_request(image_size, processing_time, True)
 return result
 except Exception as e:
 processing_time = (datetime.now() - start_time).total_seconds()
 monitor.log_request(image_size, processing_time, False)
 logger.error(f"Ошибка обработки изображения: {str(e)}")
 raise

class RobustImageProcessor:
 def __init__(self):
 self.processors = [
 self.process_with_google_vision,
 self.process_with_tesseract,
 self.process_with_azure
 ]
 self.fallback_processors = [
 self.basic_ocr_processing,
 self.manual_text_extraction
 ]
 
 def process_image(self, image_data):
 last_error = None
 
 # Попытка основными процессорами
 for processor in self.processors:
 try:
 result = processor(image_data)
 if result and result.get('text'):
 return result
 except Exception as e:
 last_error = e
 logger.warning(f"Процессор {processor.__name__} не сработал: {str(e)}")
 
 # Попытка альтернативными процессорами
 for processor in self.fallback_processors:
 try:
 result = processor(image_data)
 if result:
 return result
 except Exception as e:
 last_error = e
 logger.warning(f"Резервный процессор {processor.__name__} не сработал: {str(e)}")
 
 # Если все попытки неудачны
 raise Exception(f"Все процессоры не смогли обработать изображение. Последняя ошибка: {str(last_error)}")
 
 def process_with_google_vision(self, image_data):
 # Реализация обработки через Google Vision
 pass
 
 def process_with_tesseract(self, image_data):
 # Реализация обработки через Tesseract
 pass
 
 def process_with_azure(self, image_data):
 # Реализация обработки через Azure
 pass
 
 def basic_ocr_processing(self, image_data):
 # Базовая OCR обработка
 pass
 
 def manual_text_extraction(self, image_data):
 # Ручное извлечение текста
 pass

import unittest
from unittest.mock import patch, MagicMock

class TestImageRecognition(unittest.TestCase):
 def setUp(self):
 self.processor = ImageProcessor()
 self.test_image = self.load_test_image()
 
 def load_test_image(self):
 # Загрузка тестового изображения
 with open('test_image.jpg', 'rb') as f:
 return f.read()
 
 @patch('google.cloud.vision.ImageAnnotatorClient')
 def test_google_vision_integration(self, mock_client):
 # Мокирование ответа Google Vision
 mock_response = MagicMock()
 mock_response.text_annotations = [
 MagicMock(description="Тестовый текст из изображения")
 ]
 mock_client.return_value.text_detection.return_value = mock_response
 
 # Тест обработки изображения
 result = self.processor.process_with_google_vision(self.test_image)
 
 # Проверка результатов
 self.assertIn('text', result)
 self.assertEqual(result['text'], "Тестовый текст из изображения")
 
 def test_image_optimization(self):
 # Тест оптимизации изображения
 optimized = self.processor.optimize_image(self.test_image)
 
 # Проверка, что оптимизированное изображение не пустое
 self.assertTrue(len(optimized) > 0)
 
 def test_error_handling(self):
 # Тест обработки ошибок
 with self.assertRaises(Exception):
 self.processor.process_image(b'invalid_image_data')
 
 def test_caching(self):
 # Тест кэширования
 result1 = self.processor.process_image_with_cache(self.test_image)
 result2 = self.processor.process_image_with_cache(self.test_image)
 
 # Результаты должны быть одинаковыми
 self.assertEqual(result1, result2)

if __name__ == '__main__':
 unittest.main()