리팩토링

mafm/__init__.py

@@ -0,0 +1,4 @@
+"""MAFM (Multi-Agent File Manager) 패키지.
+
+LangGraph 기반 멀티 에이전트 파일 관리자입니다.
+"""

mafm/agent/__init__.py

@@ -0,0 +1,4 @@
+"""에이전트 패키지.
+
+LangGraph 기반 에이전트 모듈을 포함합니다.
+"""

mafm/agent/agents/__init__.py

@@ -1,5 +1,5 @@
-from .supervisor import supervisor_agent
-from .member import agent_node
-from .analyst import analyst_agent
+from mafm.agent.agents.analyst import analyst_agent
+from mafm.agent.agents.member import agent_node
+from mafm.agent.agents.supervisor import supervisor_agent
 
 __all__ = ["supervisor_agent", "agent_node", "analyst_agent"]

mafm/agent/agents/analyst.py

@@ -1,31 +1,64 @@
+"""분석가 에이전트 모듈.
+
+파일 경로 결과를 정리하는 분석가 에이전트를 정의합니다.
+"""
+
+from typing import Any
+
 from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
-from .llm_model import api_key
 from langchain_openai import ChatOpenAI
 from pydantic import BaseModel
-from typing import List, Literal
 
+from mafm.agent.agents.llm_model import api_key
 
-def analyst_agent(state, input_prompt: str, output_list: List[str]):
-    llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini")
 
-    class listResponse(BaseModel):
-        messages: List[str]
+class ListResponse(BaseModel):
+    """분석 결과 응답 모델.
+
+    Attributes:
+        messages: 필터링된 파일 경로 목록.
+    """
+
+    messages: list[str]
+
+
+def analyst_agent(
+    state: dict[str, Any],
+    input_prompt: str,
+    output_list: list[str],
+) -> dict[str, list[str]]:
+    """분석가 에이전트.
+
+    구성원들이 검색한 파일 경로들을 정리하고 필터링합니다.
+
+    Args:
+        state: 현재 에이전트 상태 (메시지 포함).
+        input_prompt: 사용자의 원본 요청.
+        output_list: 구성원들이 검색한 파일 경로 목록.
+
+    Returns:
+        필터링된 파일 경로 목록을 포함한 딕셔너리.
+    """
+    llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini")
 
     system_prompt = (
         "당신은 구성원들이 답변한 파일의 경로들을 받고 정리하는 감독자입니다."
     )
+
     prompt = ChatPromptTemplate.from_messages(
         [
             ("system", system_prompt),
             MessagesPlaceholder(variable_name="messages"),
             (
                 "system",
-                "주어진 파일 경로들 안에서 사용자 요청에 맞는 파일 경로만 뽑아주세요. 주어지지 않은 파일 경로는 뽑으면 안됩니다."
+                "주어진 파일 경로들 안에서 사용자 요청에 맞는 파일 경로만 뽑아주세요. "
+                "주어지지 않은 파일 경로는 뽑으면 안됩니다."
                 "사용자 요청: {input_prompt}"
                 "파일 경로: {output_list}",
             ),
         ]
     ).partial(input_prompt=input_prompt, output_list=", ".join(output_list))
+
     print(output_list)
-    analyst_chain = prompt | llm.with_structured_output(listResponse)
+    analyst_chain = prompt | llm.with_structured_output(ListResponse)
     return analyst_chain.invoke(state)

mafm/agent/agents/llm_model.py

@@ -1,6 +1,12 @@
-from dotenv import load_dotenv
+"""LLM 모델 설정 모듈.
+
+OpenAI API 키 로드 및 LLM 설정을 담당합니다.
+"""
+
 import os
-from langchain_openai import ChatOpenAI
+
+from dotenv import load_dotenv
 
 load_dotenv()
-api_key = os.getenv("OPENAI_API_KEY")
+
+api_key: str | None = os.getenv("OPENAI_API_KEY")

mafm/agent/agents/member.py

@@ -1,43 +1,62 @@
+"""멤버 에이전트 모듈.
+
+디렉토리 내 파일 검색을 담당하는 에이전트를 정의합니다.
+"""
+
+import os
+from typing import Any
+
 from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
+from langchain_openai import ChatOpenAI
 from pydantic import BaseModel, Field
-from typing import Literal, List
-from .llm_model import api_key
 
-# from .tools import get_file_list
-from langchain_openai import ChatOpenAI
-from langgraph.store.base import BaseStore
-from langchain.output_parsers import PydanticOutputParser
-from langchain_core.utils.function_calling import convert_to_openai_function
-from langchain_core.messages import HumanMessage
-import os
+from mafm.agent.agents.llm_model import api_key
+from mafm.rag.vector_db import search
 
-from rag.vectorDb import search
 
-global current_directory_name
+class QueryResponse(BaseModel):
+    """사용자 입력에서 추출한 검색 쿼리 응답 모델.
 
+    Attributes:
+        query: 검색에 사용할 쿼리 문장.
+    """
 
-class queryResponse(BaseModel):
     query: str = Field(description="query sentence")
 
 
-def get_file_list(query: queryResponse) -> List[str]:
-    """
-    get file list from user input
+def _get_file_list(query: QueryResponse, directory_name: str) -> list[str]:
+    """사용자 입력에서 파일 목록을 검색합니다.
+
+    Args:
+        query: 검색 쿼리 응답 객체.
+        directory_name: 검색할 디렉토리 이름.
+
+    Returns:
+        검색된 파일 경로 목록.
     """
-    global current_directory_name
+    print(f"current_directory_name: {directory_name}")
+    print(f"query: {query}")
+    db_path = f"{directory_name}/{os.path.basename(directory_name)}.db"
+    return search(db_path, [query.query])
 
-    print("current_directory_name: ", current_directory_name)
-    print("query: ", query)
-    return search(
-        current_directory_name + "/" + os.path.basename(current_directory_name) + ".db",
-        [query.query],
-    )
 
+def agent_node(
+    state: dict[str, Any],
+    directory_name: str,
+    output_list: list[str],
+) -> dict[str, list[str]]:
+    """파일 검색 에이전트 노드.
 
-def agent_node(state, directory_name: str, output_list: List[str]):
-    global current_directory_name
-    current_directory_name = directory_name
+    주어진 디렉토리에서 사용자 요청에 맞는 파일을 검색합니다.
 
+    Args:
+        state: 현재 에이전트 상태 (메시지 포함).
+        directory_name: 검색할 디렉토리 경로.
+        output_list: 검색 결과를 추가할 출력 리스트.
+
+    Returns:
+        검색 결과 메시지를 포함한 딕셔너리.
+    """
     llm = ChatOpenAI(
         api_key=api_key,
         model="gpt-4o-mini",
@@ -49,17 +68,18 @@ def agent_node(state, directory_name: str, output_list: List[str]):
             (
                 "system",
                 "current directory name: {directory_name} "
-                "사용자에 요청에 따라서 디렉토리에서 파일을 검색하려고 합니다 쿼리를 문장으로 정리해주세요",
+                "사용자에 요청에 따라서 디렉토리에서 파일을 검색하려고 합니다 "
+                "쿼리를 문장으로 정리해주세요",
             ),
         ]
-    ).partial(
-        directory_name=directory_name,
-    )
-    query_chain = prompt | llm.with_structured_output(queryResponse)
-    chain = query_chain | get_file_list
-    res = chain.invoke(state)
-    if res:
-        output_list.extend(res)
-        return {"messages": res}
-    else:
-        return {"messages": []}
+    ).partial(directory_name=directory_name)
+
+    query_chain = prompt | llm.with_structured_output(QueryResponse)
+
+    query_result = query_chain.invoke(state)
+    file_list = _get_file_list(query_result, directory_name)
+
+    if file_list:
+        output_list.extend(file_list)
+        return {"messages": file_list}
+    return {"messages": []}

mafm/agent/agents/supervisor.py

@@ -1,16 +1,43 @@
+"""감독자 에이전트 모듈.
+
+디렉토리 선택을 담당하는 감독자 에이전트를 정의합니다.
+"""
+
+from typing import Any, Literal
+
 from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
-from .llm_model import api_key
 from langchain_openai import ChatOpenAI
 from pydantic import BaseModel
-from typing import List, Literal
 
+from mafm.agent.agents.llm_model import api_key
+
+
+def supervisor_agent(
+    state: dict[str, Any],
+    member_list: list[str],
+) -> dict[str, str]:
+    """감독자 에이전트.
 
-def supervisor_agent(state, member_list: List[str]):
+    사용자 요청에 따라 다음에 실행할 디렉토리를 선택합니다.
+
+    Args:
+        state: 현재 에이전트 상태 (메시지 포함).
+        member_list: 선택 가능한 디렉토리 목록.
+
+    Returns:
+        다음 노드 이름을 포함한 딕셔너리.
+    """
     llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini")
 
     next_options = member_list + ["analyst"]
 
-    class routeResponse(BaseModel):
+    class RouteResponse(BaseModel):
+        """라우팅 응답 모델.
+
+        Attributes:
+            next: 다음으로 실행할 노드 이름.
+        """
+
         next: Literal[*(next_options)]
 
     system_prompt = (
@@ -25,10 +52,11 @@ class routeResponse(BaseModel):
             (
                 "system",
                 "선택할 수 있는 디렉토리는 다음과 같습니다: {members}. "
-                "디렉토리를 선택해주세요. 절대로 같은 디렉토리를 두 번 선택하지 마세요.",
+                "디렉토리를 선택해주세요. "
+                "절대로 같은 디렉토리를 두 번 선택하지 마세요.",
             ),
         ]
     ).partial(members=", ".join(member_list))
 
-    supervisor_chain = prompt | llm.with_structured_output(routeResponse)
+    supervisor_chain = prompt | llm.with_structured_output(RouteResponse)
     return supervisor_chain.invoke(state)

mafm/agent/agents/tools.py

@@ -1,20 +1,9 @@
-# # Legacy
-# from typing import Annotated
-# from pydantic import BaseModel, Field
-# from langchain_core.tools import tool
-# from rag.vectorDb import search
+"""레거시 도구 모듈.
 
+이 모듈은 더 이상 사용되지 않습니다. 향후 삭제 예정입니다.
 
-# @tool("get_file_list")
-# def get_file_list(
-#     query: Annotated[str, "query"], directory_name: Annotated[str, "directory name"]
-# ) -> Annotated[list, "file_list"]:
-#     """
-#     get file list from user input
-#     """
-#     # return search(member + ".db", query)
-#     return [
-#         f"file1_{directory_name}.txt",
-#         f"file2_{directory_name}.txt",
-#         f"file3_{directory_name}.txt",
-#     ]
+Deprecated:
+    이 모듈의 기능은 member.py로 이전되었습니다.
+"""
+
+# Legacy code - 향후 삭제 예정