[Django] drf AsyncOpenAI 적용기

어제 만든거에 문제가 생겼다.

서브 모듈이 private 이고 또 서브 모듈을 사용하면 ec2 에 추가적으로 무슨 작업을 해야하는 듯 했다.

그냥 private 라 문제가 생기는 것 같기도하다.

 

그래서 ecs 를 바꿨는데.. 바꿀때마다 pr 날리고 확인하는게 귀찮아서 이런걸 찾아버렸다.

https://github.com/nektos/act

GitHub - nektos/act: Run your GitHub Actions locally 🚀

 

ecs 를 로컬에서 확인할 수 있게 해준다는 그런 거던데 일단 사용해본다.

일단 다운로드를 받아보자.

brew install act

 

다음과 같이 입력했다.

act -W 파일경로 --container-architecture linux/amd64 (m2라서..)

 

음 그런데 계속 에러가 떠서... 

에러 내용은 다음과 같았다.

저거 .secrets 파일에 넣어줘야하는 모양이었다. 

근데 저거 다시 얻을 방법을 딱히 모르겠어서..

그냥 pr 날리고 테스트하려고한다.

 

완료했다...

 

토큰이 제일 맘 편했다.. ssh 는 뭔가 많았다. 뭔가 뭔가..

그 다음 여러가지 자잘자잘한 것들 수정하였다.

완료!

---

저번에 async 관련으로 문제가 발생하는 이유가 여기에서 발생한게 아닐까 하는 합리적 의심이 들고 있다. 해본다!

 

https://brownbears.tistory.com/540

여기를 스윽 보니 애초부터 async 안쓰고 있었다... create_task 로 해야했던 것 같다.

일단 실험을 해본다.

으음...

코드를 스윽 다시 읽어보니 코드 자체가 잘못된 것 같다.

asyncio.run 이 함수 내에있다. 다른 코드들을 보니 Main 쯤에 하나씩 불러주는 느낌인 것 같은데..

 

 

 

일단 drf 에서 async 를 어떻게 사용할지를 찾아보니

drf 는 지원을 안하지만  django 는 지원을 하니,,

1. drf 로 뷰를 만들지 말고, django 로 뺀다.

2. 다른 사람들이 drf 에서 async 사용하는 방법을 확인해본다.

3. async 문제인지 확인한다.

 

일단 3번부터 해본다.

나는 지금까지 asyncopenAI 자체에 문제가 있다고 생각했는데, 이슈에 딱히 그런 말도 없고 약 2년 전에 해결되었다는 의견은 봤다.

그래서 django 가 아니라 python 에서 작동을 해보고 잘 되는지 확인해보는 걸로 해보자.

이런식으로 코드를 작성해주고 진행했는데 잘 나왔다.

그러면 이제 async 관련 문제가 아니라 django 에서의 async 관련으로 뭔가 충돌이 나는 거라고 예상이된다.

진짜 잘나온다...ㅎㅎㅎ

그러면 django에서 async 관련 문제인 듯 하니, 저부분만 drf 가 아닌 django 자체를 사용하는 방식으로 사용해보려고한다.

2번 drf 의 경우에는 열띤 토론과 토의가 진행중인데.. 근데 문제는 결론은 딱히 지원을 해주지 않는다라는 것 같았다.

https://forum.djangoproject.com/t/options-for-async-requests-in-drf-adrf-normal-django-views/34730

Options for async requests in DRF? (ADRF, normal django views)

 

celery 쓸까 고민중인데,,

그 전에 drf -> django 로 변환이 되는지 한번 확인하면 참 좋을 것 같다.

그래서 일단 그렇게 진행해보려고한다.

class RecommendSubTodo(View):
    @swagger_auto_schema(
        tags=["RecommendSubTodo"],
        manual_parameters=[
            openapi.Parameter(
                "todo_id",
                openapi.IN_QUERY,
                type=openapi.TYPE_INTEGER,
                description="todo_id",
                required=True,
            ),
        ],
        operation_summary="Recommend subtodo",
        responses={200: SubTodoSerializer},
    )
    def get(self, request):
        """
        - 이 함수는 sub todo를 추천하는 함수입니다.
        """
        set_sentry_user(request.user)
        # user = request.user
        # if not user.is_authenticated:
        #     return JsonResponse(
        #         {"error": "Authentication required"}, status=403
        #     )

        # user_id = user.id
        try:
            # flag, message = UserLastUsage.check_rate_limit(
            #     user_id=user_id, RATE_LIMIT_SECONDS=RATE_LIMIT_SECONDS
            # )

            # if flag is False:
            #     return JsonResponse(
            #         {"error": message},
            #         status=429,
            #     )
            todo_id = request.GET.get("todo_id")
            if todo_id is None:
                sentry_sdk.capture_message(
                    "Todo_id not provided", level="error"
                )
                return JsonResponse(
                    {"error": "todo_id must be provided"},
                    status=400,
                )
            # 비동기적으로 OpenAI API 호출 처리
            todo = Todo.objects.get_with_id(id=todo_id)
            todo_data = {
                "id": todo.id,
                "content": todo.content,
                "date": todo.date,
                "due_time": todo.due_time,
                "category_id": todo.category_id,
            }
            completion = async_to_sync(self.get_openai_completion)(todo_data)
            response_data = json.loads(completion.choices[0].message.content)

            return JsonResponse(response_data, status=200)
        except Todo.DoesNotExist as e:
            sentry_sdk.capture_exception(e)
            return JsonResponse(
                {"error": "Todo not found"},
                status=404,
            )
        except Exception as e:
            sentry_sdk.capture_exception(e)
            return JsonResponse({"error": str(e)}, status=400)

    # 비동기적으로 OpenAI API를 호출하는 함수
    async def get_openai_completion(self, todo):
        return await openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": """너는 퍼스널 매니저야.
                        너가 하는 일은 이 사람이 할 이야기를 듣고 작업을 나눠줘야 해.
                        아래는 너가 나눠줄 작업 형식이야.
                        { id : 1, content: "운영체제 중간고사 준비", date="2024-09-01", due_time=None}
                        이런  형식으로 작성된 작업을 받았을 때 너는 이 작업을 어떻게 나눠줄 것인지를 알려주면 돼.
                        Output a JSON object structured like:
                        {id, content, date, due_time, category_id, children : [
                        {content, todo(parent todo id)}, ...}]}
                        [조건]
                        - 작업은 한 서브투두를 해결하는데 1시간 정도로 이루어지도록 제시할 것
                        - 언어는 주어진 todo content의 언어에 따를 것
                        """,  # noqa: E501
                },
                {
                    "role": "user",
                    "content": f"id: {todo['id']}, "
                    f"content: {todo['content']}, "
                    f"date: {todo['date']}, "
                    f"due_time: {todo['due_time']}, "
                    f"category_id: {todo['category_id']}",
                },
            ],
            response_format={"type": "json_object"},
        )

 

이런식으로 ... django 관련으로 코드를 작성한 후에 진행해도 똑같은 결과가 발생했다.

 

셀러리를 도입해야 할 것 같다....

 

도입해본다

https://backstreet-programmer.tistory.com/198

[Django + Redis + Celery] Python 분산 비동기 작업 큐 (Distributed Task/Job Queue) 튜토리얼

여기 참고했다.

음

셀러리 비트 사용하면 되지 않을까..

그리고 그 후에 email 보내는 것도 로직에 있어서 이거 celery 로 빼면 될듯

 

 

 

---

 

https://github.com/cpprhtn/pypi-Korean-Translations?tab=readme-ov-file

GitHub - cpprhtn/pypi-Korean-Translations

몇가지 번역을 해보았다!