langchain & langgraph

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# models.py

from langchain_openai import AzureChatOpenAI

from config import settings

llm = AzureChatOpenAI(
    azure_deployment="gpt-4o",
    api_version="2024-08-01-preview",
    api_key=settings.AZURE_OPENAI_API_KEY,
    azure_endpoint=settings.AZURE_OPENAI_ENDPOINT,
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

# schemas.py

from typing import Any, Dict, List
from langgraph.graph import MessagesState


class AgentSystemSchema(MessagesState):
    """
    에이전트 전반에서 공통으로 참조하는 시스템 상태 정의
    (비즈니스 컨텍스트 및 시스템 고정 정보)
    """

    system_message: str
    # LLM에게 전달되는 시스템 프롬프트

    products: List[Dict[str, Any]]
    # 주문 가능한 상품 목록 (상품명, 가격, 옵션 등 비즈니스 정보)

    previous_orders: str
    # 사용자의 이전 주문 내역 요약 정보
    # 대화 맥락 유지 및 재주문/추천 판단에 활용


class AgentState(MessagesState):
    """
    단일 대화 흐름에서 변경되며 누적되는 에이전트 실행 상태
    """

    message: str
    # 가장 최근 사용자 또는 에이전트의 발화

    messages: List[Any]
    # 전체 대화 이력
    # 메시지 누적 관리 및 LLM 컨텍스트 제공 용도

    summary: str
    # messages가 과도하게 누적되는 것을 방지하기 위한
    # 기존 대화 요약 정보 (컨텍스트 압축 목적)

    system: AgentSystemSchema
    # 시스템 전역 상태
    # 대화 중 변경되지 않거나 드물게 변경되는 비즈니스/환경 정보

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

## 사용자의 발화가 주문인지 아닌지 판단

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate

from ..llm.models import llm
from ..schemas import AgentState

order_detection_prompt = PromptTemplate.from_template("""
당신은 사용자의 입력이 주문인지(혹은 주문의 수정, 삭제인지) 아닌지 판별하는 주문 판별 전문 AI 어시스턴트입니다.
무조건 아래 schema를 준수해서 JSON 형태로 답을 해야 합니다.

class OrderState(str, Enum):
    right_order = "right_order"
    wrong_order = "wrong_order"
    not_order = "not_order"

class OrderDetectionSchema(BaseModel):
    is_order: OrderState
    system_message: str

사용자 입력:
{messages}

구매할 수 있는 제품목록:
{available_products}

이전 주문 내역:
{previous_orders}


주문 여부를 판단 기준
1. 제품명이 존재하는지?
- '사용자 입력'에는 반드시 상품명과 수량이 포함되어 있어야 함
- 사용자는 공식 상품명을 그대로 쓰지 않고, 약어나 고유 키워드만으로 표현하는 경우가 많으므로 이 점을 반드시 고려할 것
- '유사한 상품명'에 대한 정보와 반드시 비교해서 유사한 상품이 있는지 알아본 후 주문인지 아닌지 고려할 것.
- '제품 목록'에 100% 일치하는 상품이 있는 경우, 이 상품을 주문하는 것으로 판단할 것
- 사용자가 입력한 상품명은 제공된 '구매할 수 있는 제품목록'과 반드시 연관성이 있어야 하며, 부분적으로 일치하거나 키워드 유사도(예: "만두"가 "고기 만두" 또는 "김치 만두"와 부분적으로 일치하는 경우)가 있다면, 정확히 일치하지 않아도 유효한 상품명으로 인식하세요. 유사한 제품이 복수일 때에도 주문으로 간주할 것
- 주문 여부 판단 시, 상품명이 매우 추상적일 수 있음을 염두에 두고, 연관성이 다소 낮더라도 제품 정보에 유사한 단어나 키워드가 있으면 '주문'을 한거라고 판단할 것

2. 수량이 존재하는지?
- 수량은 상품명에 직접 붙어 있을 수도 있고, 별도의 단위와 함께 숫자로 표기될 수도 있습니다. 단, 수량은 항상 정수로 표현됩니다.
- 수량이 각각의 상품명 끝에 직접 붙어 있을 수 있으며, 이 경우 각 숫자를 해당 상품의 주문 수량으로 해석해야 합니다(예: "갓김치1"은 제품 '갓김치 1개 주문', "참이슬1"은 제품 '참이슬 1개 주문').
- 상품명 뒤에 공백 없이 숫자가 바로 따라오면(예: "배추김치103"), 그 숫자를 해당 상품의 주문 수량으로 해석해야 합니다. 즉, "배추김치103"은 상품명 "배추김치"에 수량 103개로 인식해야 합니다.
- 수량은 1, 10, 100, 1000, 10000 이상도 가능하며, 이는 특이한 케이스가 아니고 수량을 의미한다고 판단할 것
- 이전 주문 내역(previous_orders)을 참고해서 주문 여부를 판단할 것. 이전 주문 내역과 유사한 형태로 입력했다면 이는 주문일 확률이 높습니다.
- 주문할 수 없는 상품에 대한 주문으로 판단되면, 해당되는 상품 전체를 언급하며 다시 정확히 주문해달라고 system_message 필드에 작성해 주세요.
(ex. 말씀하신 {messages}는 주문으로 판단되지 않습니다.)

주의사항
위 기준을 바탕으로 입력을 꼼꼼히 분석한 뒤, 주문이면 is_order 필드에 주어진 Enum에 맞는 값으로 반환해주세요.
is_order에는 <class 'str'>타입의 'right_order', 'wrong_order', 'not_order' 이외에는 어떤 값도 들어갈 수 없습니다. 또 다른 어떤 문자나 특수문자를 추가하지 마세요. 꼭 <class 'str'>로 값이 들어와야 합니다.
system_message에는 <class 'str'> 타입으로 반환해야 합니다.

당부사항
❗system_message에는 사용자의 입력을 왜 주문으로 판단했는지, 어떤 상품 정보를 비교했는지, 어떤 상품에 유사하다고 판단했는지, 상품명과 수량이 잘 표시되었는지, 결과적으로 왜 주문으로 판단했는지 그 근거를 반드시 한글로 작성해 주세요.
""")


async def is_order_should_continue(state: AgentState):
    messages = state["messages"]
    available_products = (
        state["system"].get("buyer_info_for_order", {}).get("products_info", [])
    )
    ...

    rag_chain = order_detection_prompt | llm | JsonOutputParser()
    
    answer = await rag_chain.ainvoke(
        {
            "messages": messages,
            "available_products": available_products,
            "previous_orders": previous_orders,
            "similar_products": similar_products,
        }
    )

    if answer["is_order"] == "right_order":
        return "order_parser"
    elif answer["is_order"] == "wrong_order":
        return "wrong_order_parser"
    elif answer["is_order"] == "not_order":
        return "invite_branch"
    else:
        return "safe_exit"

1

rag_chain = order_detection_prompt | llm | JsonOutputParser()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

# is_invite_should_continue.py

from langchain_core.prompts import PromptTemplate
from langgraph.graph import END

from ..llm.models import llm
from ..schemas import AgentState
from .utils.boolean_output_parser import BooleanOutputParser

invite_detection_prompt = PromptTemplate.from_template("""
사용자 입력:
{messages}

- 입력된 문자열은 반드시 6자리여야 하며, 오직 숫자 0부터 9까지의 숫자로만 이루어져 있어야 합니다.  
- 띄어쓰기, 한글, 영어, 특수문자, 기타 어떤 문자도 포함되어 있으면 안 됩니다.

반환 값 조건:
- 출력은 반드시  <class 'bool'> 타입의 True 또는 False 만 반환하세요.

예시:
- '495038' → True  
- '000123' → True  
- '67384' → False (5자리)  
- '1234567' → False (7자리 이상)  
- '123 456' → False (띄어쓰기 포함)  
- 'ABC123' → False (영어 포함)  
- '123!@#' → False (특수문자 포함)
""")


async def is_invite_should_continue(state: AgentState):
    messages = state["messages"][-1].content

    rag_chain = invite_detection_prompt | llm | BooleanOutputParser()
    response = await rag_chain.ainvoke({"messages": messages})

    if response:
        return "invite_parser"
    return END

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

# graph.builder.py

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph

from ..nodes import (
    init_order,
    invite_branch,
    invite_parser,
    is_invite_should_continue,
    is_order_should_continue,
    order_confirm_should_continue,
    order_parser,
    ...
)
from ..schemas import AgentState

## graph_builder 설정 - node 추가
graph_builder = StateGraph(AgentState)

graph_builder.add_node("invite_branch", invite_branch)
graph_builder.add_node("invite_parser", invite_parser)
graph_builder.add_node("init_order", init_order)
...

## graph_builder 설정 - edge 추가
graph_builder.add_conditional_edges(
    START, order_confirm_should_continue, ["update_messages", "init_order"]
)
graph_builder.add_edge("update_messages", "retrieve")
graph_builder.add_edge("retrieve", "similar_products")
graph_builder.add_conditional_edges(
    "similar_products",
    is_order_should_continue,
    ["order_parser", "wrong_order_parser", "invite_branch", "safe_exit"],
)
...

1
2
3
4
5
6
7
8
9

from langgraph.checkpoint.memory import MemorySaver

from .graph.builder import graph_builder

# langgraph 체크포인터 설정(히스토리 모드를 위해 사용)
checkpointer = MemorySaver()

# langgraph 그래프 컴파일
graph = graph_builder.compile(checkpointer=checkpointer)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

import asyncio
from typing import Any

import httpx
from fastapi import APIRouter, Request
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver

from .graph.builder import graph_builder

order_ai_router = APIRouter()

# langgraph 체크포인터 설정(히스토리 모드를 위해 사용)
checkpointer = MemorySaver()

# langgraph 그래프 컴파일
graph = graph_builder.compile(checkpointer=checkpointer)

async def get_ai_response(message: str, plusfriend_user_key: str):
    config = {"configurable": {"thread_id": user_pk}}

    result = await graph.ainvoke(
        {
            "message": [HumanMessage(message)],
            "plusfriend_user_key": plusfriend_user_key,
        },
        config=config,
    ) 
    return result