commit c04391138b5cf40e25887e65f3d9c73146ff33f8
Author: Damian Myrda <damian@prime8.dev>
Date: Mon, 10 Jun 2024 13:54:43 -0500
batman
Diffstat:
9 files changed, 246 insertions(+), 0 deletions(-)
diff --git a/.gitignore b/.gitignore
@@ -0,0 +1,2 @@
+/openai.env
+/tests/
diff --git a/Dockerfile b/Dockerfile
@@ -0,0 +1,7 @@
+FROM python:3
+WORKDIR /usr/src/app
+
+RUN pip install fastapi aiokafka
+COPY . .
+EXPOSE 8000
+CMD [ "fastapi", "run" ]
diff --git a/compose.yml b/compose.yml
@@ -0,0 +1,70 @@
+services:
+ zookeeper:
+ image: confluentinc/cp-zookeeper
+ hostname: zookeeper
+ container_name: zookeeper
+ ports:
+ - "2181:2181"
+ environment:
+ ZOOKEEPER_CLIENT_PORT: 2181
+ ZOOKEEPER_SERVER_ID: 1
+ ZOOKEEPER_SERVERS: zookeeper:2888:3888
+ attach: false
+
+ kafka:
+ image: confluentinc/cp-kafka
+ hostname: kafka
+ container_name: kafka
+ ports:
+ - "9092:9092"
+ environment:
+ KAFKA_ADVERTISED_LISTENERS: "INTERNAL://kafka:29092,EXTERNAL://localhost:9092"
+ KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
+ KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
+ KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
+ KAFKA_BROKER_ID: 1
+ KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"
+ KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+ KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+ KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+ KAFKA_JMX_PORT: 9999
+ KAFKA_JMX_HOSTNAME: ${DOCKER_HOST_IP:-127.0.0.1}
+ KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
+ KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "true"
+ KAFKA_MESSAGE_MAX_BYTES: 26214400
+ attach: false
+ depends_on:
+ - zookeeper
+
+ voice_to_text:
+ build: ./transcribe
+ container_name: transcriber
+ env_file:
+ - "kafka.env"
+ - "openai.env"
+ depends_on:
+ - kafka
+ restart: always
+
+ text_to_summary:
+ build: ./summarize
+ container_name: summarizer
+ env_file:
+ - "kafka.env"
+ - "openai.env"
+ depends_on:
+ - kafka
+ restart: always
+
+ frontend:
+ build: .
+ container_name: frontend
+ env_file:
+ - "kafka.env"
+ ports:
+ - "8000:8000"
+ depends_on:
+ - kafka
+ - voice_to_text
+ - text_to_summary
+ restart: always
diff --git a/kafka.env b/kafka.env
@@ -0,0 +1 @@
+KAFKA_ADDRESS="kafka:29092"
diff --git a/main.py b/main.py
@@ -0,0 +1,69 @@
+from fastapi import FastAPI, UploadFile
+from fastapi.responses import HTMLResponse
+import os
+from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
+import pickle
+
+app = FastAPI()
+address = os.environ.get("KAFKA_ADDRESS", "localhost:9092")
+
+
+@app.get("/", response_class=HTMLResponse)
+async def index():
+ return HTMLResponse(
+ content="""
+ <html>
+ <form enctype="multipart/form-data" method="post">
+ <input type="file" name="file" />
+ <button type="submit">Submit</button>
+ </form>
+ </html>
+ """
+ )
+
+
+@app.post("/")
+async def summarize(file: UploadFile):
+ print("transribing file")
+ producer = AIOKafkaProducer(
+ max_request_size=26214400, bootstrap_servers=address
+ )
+ await producer.start()
+ await producer.send(
+ "transcription.input",
+ pickle.dumps(
+ {
+ "name": file.filename,
+ "data": await file.read(),
+ }
+ ),
+ )
+ await producer.stop()
+ consumer = AIOKafkaConsumer(
+ "transcription.output", bootstrap_servers=address
+ )
+ await consumer.start()
+ message = await consumer.getone()
+ await consumer.stop()
+ text = message.value.decode()
+
+ print("summarizing text")
+ producer = AIOKafkaProducer(
+ max_request_size=26214400, bootstrap_servers=address
+ )
+ await producer.start()
+ await producer.send("summarization.input", str.encode(text))
+ await producer.stop()
+ consumer = AIOKafkaConsumer(
+ "summarization.output", bootstrap_servers=address
+ )
+ await consumer.start()
+ message = await consumer.getone()
+ await consumer.stop()
+ summary = message.value.decode()
+
+ print("request completed")
+ return {
+ "text": text,
+ "summary": summary,
+ }
diff --git a/summarize/Dockerfile b/summarize/Dockerfile
@@ -0,0 +1,6 @@
+FROM python:3
+WORKDIR /usr/src/app
+
+RUN pip install aiokafka openai
+COPY . .
+CMD [ "python", "main.py" ]
diff --git a/summarize/main.py b/summarize/main.py
@@ -0,0 +1,40 @@
+import os
+import asyncio
+from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
+from openai import AsyncOpenAI
+
+client = AsyncOpenAI()
+address = os.environ.get("KAFKA_ADDRESS", "localhost:9092")
+
+
+async def summarizer():
+ consumer = AIOKafkaConsumer(
+ "summarization.input", bootstrap_servers=address
+ )
+ await consumer.start()
+
+ while True:
+ message = await consumer.getone()
+ text = message.value.decode()
+
+ summarization = await client.chat.completions.create(
+ messages=[
+ {
+ "role": "user",
+ "content": "summarize the following:\n" + text,
+ }
+ ],
+ model="gpt-4-turbo",
+ )
+ summary = summarization.choices[0].message.content
+
+ producer = AIOKafkaProducer(bootstrap_servers=address)
+ await producer.start()
+ await producer.send("summarization.output", str.encode(summary))
+ await producer.stop()
+
+ await consumer.stop()
+
+
+if __name__ == "__main__":
+ asyncio.run(summarizer())
diff --git a/transcribe/Dockerfile b/transcribe/Dockerfile
@@ -0,0 +1,6 @@
+FROM python:3
+WORKDIR /usr/src/app
+
+RUN pip install aiokafka openai
+COPY . .
+CMD [ "python", "main.py" ]
diff --git a/transcribe/main.py b/transcribe/main.py
@@ -0,0 +1,45 @@
+import os
+import asyncio
+from openai import AsyncOpenAI
+from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
+import pickle
+import io
+
+client = AsyncOpenAI()
+address = os.environ.get("KAFKA_ADDRESS", "localhost:9092")
+
+
+async def transcribe():
+ consumer = AIOKafkaConsumer(
+ "transcription.input", bootstrap_servers=address
+ )
+ await consumer.start()
+
+ while True:
+ message = await consumer.getone()
+ file = pickle.loads(message.value)
+
+ text = ""
+
+ CHUNK_SIZE = 26214400 - 1024
+ for i in range(0, len(file["data"]), CHUNK_SIZE):
+ audio_file = io.BytesIO(file["data"][i : i + CHUNK_SIZE])
+ audio_file.name = file["name"]
+
+ transcription = await client.audio.transcriptions.create(
+ model="whisper-1", file=audio_file
+ )
+ text += transcription.text
+
+ producer = AIOKafkaProducer(
+ max_request_size=26214400, bootstrap_servers=address
+ )
+ await producer.start()
+ await producer.send("transcription.output", str.encode(text))
+ await producer.stop()
+
+ await consumer.stop()
+
+
+if __name__ == "__main__":
+ asyncio.run(transcribe())
