셀트리온제약의 의약품 공급망 관리 강화를 위한 멀티 에이전트 시스템 구축 여정

# 1. 사용자 요청 & AI 응답 Node State에 추가
async def drug_searcher_node(state: State) -> Command[Literal["__end__"]]:
        ... 
    history = state.get("history", [])
    history.append({"agent": "drug_searcher", "message": final_response})
    ...

    return Command(
        update={
                        **contents,
            "history": history,
        },
        goto="__end__",
    )
    
# 2. history 내역 Agent Instruction 에 주입
def apply_prompt_template(prompt_name: str, state: AgentState, prompt_cache=False, cache_type="default") -> list:
    system_prompts = get_prompt_template(prompt_name)
    context = {"CURRENT_TIME": datetime.now().strftime("%a %b %d %Y %H:%M:%S %z"), "PREVIOUS_CONTEXT": format_previous_context(state.get("history", []))}
    try:
        system_prompts = system_prompts.format(**context)
    except BaseException as e:
        raise("System Prompts 변환 오류 발생 : ", e)
        
    system_prompts = bedrock_utils.get_system_prompt(system_prompts=system_prompts, prompt_cache=prompt_cache, cache_type=cache_type)         
    return system_prompts, state["messages"]
    
# 3. drug_searcher.md (Agent Instruction 파일)
...
# Previous Context Instructions 
{PREVIOUS_CONTEXT}
...

# SCMS System's Database Description

<scms_system_information>
    <database_system_information>
        It is postgreSQL, version "PostgreSQL 14.19 (Ubuntu 14.19-0ubuntu0.22.04.1) on x86_64-pc-linux-gnu, compiled by gcc (Ubuntu 11.4.0-1ubuntu1~22.04.2) 11.4.0, 64-bit".
    </database_system_information>
    <system_description>
        The SCMS is acronym that is 'S'upply 'C'hain 'M'anagement 'S'ystem. It is for S&OP committee...
      </system_description>
      <words_description>
          DB에서 사용되는 업무와 관련된 단어 설명
      </words_description>
      <databse_authorization>
        AI 에이전트가 접속할 DB 계정의 권한 설명
      </databse_authorization>      
      <database_table_name>
          DB Table 이름
      </database_table_name>
      <database_table_description>
      DB Table 설명  
    </database_table_description>
    <database_table>
        <columns>
            <columns1> Column1 설명.. </columns1>
            <columns2> Column2 설명.. </columns2>            
            ...
        </columns>        
    </database_table>        
</scms_system_information>

async def get_result_scms_db(query: Annotated[str, ""]) -> str:
    """
    Use this tool to find out internal information from database for SCMS.

    Args:
        query : The SQL for SCMS DB.

    Returns:
        The information which is managed internally. (e.g. 제품코드, 제품명, 사업단가, 배치사이즈, 판매단위(팩) etc.)
    """

    def validate_sql(sql: str) -> bool:
        """
        The function prevent to execute invalid SQL except 'SELECT'.
        """
        if not sql.strip().upper().startswith('SELECT'):
            raise Exception("SELECT 문만 실행 가능합니다.")

        dangerous = ['DROP', 'DELETE', 'UPDATE', 'INSERT', 'CREATE', 'ALTER', 'TRUNCATE']
        for keyword in dangerous:
            if keyword in sql.upper():
                raise Exception(f"보안상 {keyword} 문은 실행할 수 없습니다.")
        return True

    if not query:
        return f"It is not existed SQL."

    try:
        conn = await asyncpg.connect(
            host=os.getenv('DB_HOST'),
            port=os.getenv('DB_PORT'),
            database=os.getenv('DB_NAME'),
            user=os.getenv('DB_USER'),
            password=os.getenv('DB_PASSWORD')
        )

        validate_sql(query)
        rows = await conn.fetch(query)
        results = "\n".join([str(row) for row in rows])

        await conn.close()
        return results
    except BaseException as e:
        return f"Error fetching data from SCMS DB: {str(e)}"

async def korean_drug_search(drug_name: Annotated[str, "The drug name"]) -> str:
    """
    Use this tool to find out drug information from MFDS web-site.

    Args:
        drug_name : Text from user request for drug

    Returns:
        The drug information (e.g. 품목기준코드, 제품명, 업체명, 허가일, 취소/취하, 취소/취하일자, 주성분/주원료)
    """

    async def get_mfds_drug_details(product_code: Annotated[str, "The drug code"]) -> str:
        """
        Use this tool to find out URL for detail information on MFDS web-site by '품목기준코드'.

        Args:
            product_code : The key to find out drug information.

        Returns:
            URL for detail drugs information.
        """
        url = f"https://nedrug.mfds.go.kr/pbp/CCBBB01/getItemDetailCache?cacheSeq={product_code}"

        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(url) as response:
                    response.raise_for_status()
                    html_content = await response.text()
                    soup = BeautifulSoup(html_content, 'html.parser')

            consign_manufact_row = soup.find('th', string='위탁제조업체').find_parent('tr')
            consign_manufact = consign_manufact_row.find('td').text.strip().split(', ') if consign_manufact_row else []
            standard_code_row = soup.find('th', string='표준코드').find_parent('tr')
            standard_codes = standard_code_row.find('td').text.strip().split(', ') if standard_code_row else []
            usage_period_row = soup.find('th', string='사용기간').find_parent('tr')
            usage_period = usage_period_row.find('td').text.strip() if usage_period_row else "Not found"
            usage_pirce_row = soup.find('th', string='보험약가').find_parent('tr')
            usage_pirce = usage_pirce_row.find('td').text.strip() if usage_pirce_row else "Not found"


            effect_div = soup.find('div', id='_ee_doc')
            dosage = soup.find('div', id='_ud_doc')

            packing_information_row = soup.find('th', string='포장정보').find_parent('tr')
            packing_information = packing_information_row.find(
                'td').text.strip() if packing_information_row else "Not found"

            recipe = soup.find('div', id='scroll_02')

            result_text = f"품목기준코드: {product_code}\n"
            result_text += f"표준코드 리스트: {', '.join(standard_codes)}\n"
            result_text += f"사용기간: {usage_period}"
            result_text += f"보험약가: {usage_pirce}"
            result_text += f"위탁제조업체: {consign_manufact}"
            result_text += f"효능효과: {effect_div}"
            result_text += f"포장정보: {packing_information}"
            result_text += f"원료약품 및 분량/주성분/부형제/첨가제: {recipe}"

            return result_text
        except aiohttp.ClientError as e:
            return f"Error fetching data from MFDS website: {str(e)}"

    url = "https://nedrug.mfds.go.kr/searchDrug"
    params = {
        "itemName": drug_name
    }
    
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
    }
    
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url, params=params, headers=headers) as response:
                response.raise_for_status()
                html_content = await response.text()
                soup = BeautifulSoup(html_content, 'html.parser')

        table = soup.find('table', class_=['dr_table2', 'dr_table_type2'])
        if not table:
            return f"No results found for {drug_name}."

        rows = table.find_all('tr')
        item_code = ""
        result = []
        for row in rows[1:]:
            columns = row.find_all('td')
            item_name = ""; company = ""; approval_date = ""
            cancel_status = ""; cancel_date = ""; drug_api =""
            if len(columns) >= 10:
                item_name = columns[1].find('a').text.strip() if columns[1].find('a') else "Not found"
                company = columns[3].text.strip().replace("업체명", "")
                item_code = columns[5].text.strip().replace("품목기준코드", "")
                approval_date = columns[7].text.strip()
                cancel_status = columns[9].text.strip()
                cancel_date = columns[10].text.strip()
                drug_api = columns[11].text.strip()

            result_item = {
                "품목기준코드": item_code,
                "제품명": item_name,
                "업체명": company,
                "허가일": approval_date,
                "취소/취하": cancel_status,
                "취소/취하일자": cancel_date,
                "주성분/주원료": drug_api,
                "상세정보": ""
            }

            result_item["상세정보"] = await get_mfds_drug_details(item_code)
            result.append(result_item)
        return result
    
    except aiohttp.ClientError as e:
        return f"Error occurred while fetching data: {str(e)}"

async def get_result_knowledge_base(query: Annotated[str, "Must be found information from Bedrock's Knowledge Base."]) -> str:
    """
    Use this tool to find out internal information from AWS Bedrock's Knowledge Base.

    Args:
        query : Text from user request

    Returns:
        The internal information( e.g., 사업계획 매출액, 공헌이익 etc.)
    """
    try:
        bedrock_agent = await get_bedrock_agent_client_async()
    
        retrieve_response = await bedrock_agent.retrieve(
            knowledgeBaseId=os.getenv('KNOWLEDGE_BASE_ID'),
            retrievalQuery={'text': query},
            retrievalConfiguration={'vectorSearchConfiguration': {'numberOfResults': 5}}
        )
    
        result = [result['content']['text'] for result in retrieve_response.get('retrievalResults', []) if
                         'content' in result and 'text' in result['content']]
        if not result:
            return "AWS Knowledge Base에서 관련 정보를 찾을 수 없습니다."
    except Exception as e:
        raise f"Bedrock-KB 호출 에러 발생 : {e}"
    finally:
        import aioboto3
        if isinstance(bedrock_agent, aioboto3.Session):
            await bedrock_agent.close()
    return result

...
# Tools
tavily_tool : You can look up specific website.
[CRITICAL] It should be used when you can't find out answer using korean_drug_search.
[CRITICAL] You can use this tool only website below :
     1) Access URL 1 : (약학정보원) https://www.health.kr/
     2) Access URL 2 : (특허청) https://www.kipris.or.kr/
Examples
[Information X] About drug information on official two websites.
...

# 1. Amazon Bedrock-Knowledge Base 에서 AI 답변과 관련된 이미지가 저장된 S3경로 Return 함수
async def get_result_business_report(query: Annotated[str, "Must be final AI's response."]) -> str:
    report_agent = await get_bedrock_agent_client_async()

    retrieve_response = await report_agent.retrieve(
        knowledgeBaseId="Amazon Bedrock-Knowledge Base's ID"
        retrievalQuery={'text': query},
        retrievalConfiguration={'vectorSearchConfiguration': {'numberOfResults': 5}}
    )

    result = [result['content']['text'] for result in retrieve_response.get('retrievalResults', []) if
                     'content' in result and 'text' in result['content']]
    if not result:
        return None

    s3_paths = []
    for text in result:
        try:
            if '"s3_path"' in text:
                s3_path_match = re.search(r'"s3_path":\s*"([^"]+)"', text)
                if s3_path_match:
                    s3_paths.append(s3_path_match.group(1))
        except Exception as e:
            print(f"s3_path 추출 중 오류: {e}")
            continue
    
    await report_agent.close()
    
    if s3_paths:
        unique_s3_paths = list(set(s3_paths))
        return "\n".join(unique_s3_paths)
    else:
        return None
  
# 2. drug_searcher Node에서 S3경로 리턴받는 부분      
async def drug_searcher_node(state: State) -> Command[Literal["__end__"]]:
        ... 
      s3_paths = ""
    try:
        logger.info(f"{Colors.BLUE}===== Business Report 검색 시작 ====={Colors.END}")
        business_report_result = await get_result_business_report(query=user_query)
        
        # ③ 에이전트 최종 답변으로 해당 페이지의 설명 검색, S3경로를 리턴 받음
        if business_report_result:
            logger.info(f"{Colors.BLUE}Business Report에서 관련 문서를 찾았습니다.{Colors.END}")
            s3_paths = business_report_result.split("\n")
        else:
            logger.info(f"{Colors.BLUE}Business Report에서 관련 문서를 찾지 못했습니다.{Colors.END}")
        final_response = result["content"][-1]["text"]
    except Exception as e:
        logger.error(f"Business Report 검색 중 오류 발생: {e}")
        final_response = result["content"][-1]["text"]

    # ④ LangGraph State의 s3_path 속성 업데이트, 최종 답변과 함께 리턴
    return Command(
        update={
            **contents,
            "s3_paths": s3_paths,
        },
        goto="__end__",
    )

name: CustomAgent
nodes:
  - name: Coordinator  
  - name: drug_searcher
  - name: tool list
edges:
  - from: __start__
    to: Coordinator
  - from: tool list
    to: drug_searcher
  - from: Coordinator
    condition: 사내정보 여부
    paths: [drug_searcher, __end__]
  - from: drug_searcher
    condition: 의약품 정보
    paths: [__end__, tool list]

from typing_extensions import TypedDict
from stub import CustomAgent

class SomeState(TypedDict):
    foo: str

def Coordinator(state: SomeState) -> dict:
    print("In node: Coordinator")
    return {
        # Add your state update logic here
    }

def drug_searcher(state: SomeState) -> dict:
    print("In node: drug_searcher")
    return {
        # Add your state update logic here
    }

def tool_list(state: SomeState) -> dict:
    print("In node: tool list")
    return {
        # Add your state update logic here
    }

def 사내정보_여부(state: SomeState) -> str:
    print("In condition: 사내정보 여부")
    raise NotImplementedError("Implement me.")

def 의약품_정보(state: SomeState) -> str:
    print("In condition: 의약품 정보")
    raise NotImplementedError("Implement me.")

agent = CustomAgent(
    state_schema=SomeState,
    impl=[
        ("Coordinator", Coordinator),
        ("drug_searcher", drug_searcher),
        ("tool_list", tool_list),
        ("사내정보 여부", 사내정보_여부),
        ("의약품 정보", 의약품_정보),
    ],
)

compiled_agent = agent.compile()