[PYTHON/LANGCHAIN] CompiledStateGraph 클래스 : astream 메소드를 사용해 반복적 정제를 통한 텍스트 요약하기

■ CompiledStateGraph 클래스의 astream 메소드를 사용해 반복적 정제를 통한 텍스트를 요약하는 방법을 보여준다.

※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다.

▶ main.py


import asyncio

from dotenv                        import load_dotenv
from langchain_core.documents      import Document
from langchain_core.prompts        import ChatPromptTemplate
from langchain_openai              import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from typing                        import TypedDict
from typing                        import List
from langchain_core.runnables      import RunnableConfig
from typing                        import Literal
from langgraph.graph               import END
from langgraph.graph               import StateGraph
from langgraph.graph               import START

load_dotenv()

documentList = [
    Document(page_content = "Apples are red"      , metadata = {"title" : "apple_book"    }),
    Document(page_content = "Blueberries are blue", metadata = {"title" : "blueberry_book"}),
    Document(page_content = "Bananas are yelow"   , metadata = {"title" : "banana_book"   })
]

summarizeChatPromptTemplate = ChatPromptTemplate([("human", "Write a concise summary of the following : {context}")])

chatOpenAI = ChatOpenAI(model = "gpt-4o-mini")

summarizeRunnableSequence = summarizeChatPromptTemplate | chatOpenAI | StrOutputParser()

refineChatPromptTemplateString = """
Produce a final summary.

Existing summary up to this point:
{existing_answer}

New context:
------------
{context}
------------

Given the new context, refine the original summary.
"""

refineChatPromptTemplate = ChatPromptTemplate([("human", refineChatPromptTemplateString)])

refineRunnableSequence = refineChatPromptTemplate | chatOpenAI | StrOutputParser()

class State(TypedDict):
    contentList : List[str]
    index       : int
    summary     : str

async def generateInitialSummary(state : State, config : RunnableConfig):
    responseString = await summarizeRunnableSequence.ainvoke(state["contentList"][0], config)
    return {"summary" : responseString, "index" : 1}

async def refineSummary(state : State, config : RunnableConfig):
    content = state["contentList"][state["index"]]
    responseString = await refineRunnableSequence.ainvoke({"existing_answer" : state["summary"], "context" : content}, config)
    return {"summary" : responseString, "index" : state["index"] + 1}

def shouldRefine(state : State) -> Literal["refine_summary", END]:
    if state["index"] >= len(state["contentList"]):
        return END
    else:
        return "refine_summary"

stateGraph = StateGraph(State)

stateGraph.add_node("generate_initial_summary", generateInitialSummary)
stateGraph.add_node("refine_summary"          , refineSummary         )

stateGraph.add_edge(START, "generate_initial_summary")
stateGraph.add_conditional_edges("generate_initial_summary", shouldRefine)
stateGraph.add_conditional_edges("refine_summary"          , shouldRefine)

compiledStateGraph = stateGraph.compile()

async def main():
    async for addableUpdatesDict  in compiledStateGraph.astream({"contentList" : [documentList.page_content for documentList in documentList]}, stream_mode = "values"):
        if summary := addableUpdatesDict .get("summary"):
            print(summary)

asyncio.run(main())

"""
Apples are typically red in color.
Apples are typically red in color, while blueberries are blue.
Apples are typically red in color, blueberries are blue, and bananas are yellow.
"""

import asyncio

from dotenv import load_dotenv

from langchain_core.documents import Document

from langchain_core.prompts import ChatPromptTemplate

from langchain_openai import ChatOpenAI

from langchain_core.output_parsers import StrOutputParser

from typing import TypedDict

from typing import List

from langchain_core.runnables import RunnableConfig

from typing import Literal

from langgraph.graph import END

from langgraph.graph import StateGraph

from langgraph.graph import START

load_dotenv()

documentList = [

Document(page_content = "Apples are red" , metadata = {"title" : "apple_book" }),

Document(page_content = "Blueberries are blue", metadata = {"title" : "blueberry_book"}),

Document(page_content = "Bananas are yelow" , metadata = {"title" : "banana_book" })

]

summarizeChatPromptTemplate = ChatPromptTemplate([("human", "Write a concise summary of the following : {context}")])

chatOpenAI = ChatOpenAI(model = "gpt-4o-mini")

summarizeRunnableSequence = summarizeChatPromptTemplate | chatOpenAI | StrOutputParser()

refineChatPromptTemplateString = """

Produce a final summary.

Existing summary up to this point:

{existing_answer}

New context:

------------

{context}

------------

Given the new context, refine the original summary.

"""

refineChatPromptTemplate = ChatPromptTemplate([("human", refineChatPromptTemplateString)])

refineRunnableSequence = refineChatPromptTemplate | chatOpenAI | StrOutputParser()

class State(TypedDict):

contentList : List[str]

index : int

summary : str

async def generateInitialSummary(state : State, config : RunnableConfig):

responseString = await summarizeRunnableSequence.ainvoke(state["contentList"][0], config)

return {"summary" : responseString, "index" : 1}

async def refineSummary(state : State, config : RunnableConfig):

content = state["contentList"][state["index"]]

responseString = await refineRunnableSequence.ainvoke({"existing_answer" : state["summary"], "context" : content}, config)

return {"summary" : responseString, "index" : state["index"] + 1}

def shouldRefine(state : State) -> Literal["refine_summary", END]:

if state["index"] >= len(state["contentList"]):

return END

else:

return "refine_summary"

stateGraph = StateGraph(State)

stateGraph.add_node("generate_initial_summary", generateInitialSummary)

stateGraph.add_node("refine_summary" , refineSummary )

stateGraph.add_edge(START, "generate_initial_summary")

stateGraph.add_conditional_edges("generate_initial_summary", shouldRefine)

stateGraph.add_conditional_edges("refine_summary" , shouldRefine)

compiledStateGraph = stateGraph.compile()

async def main():

async for addableUpdatesDict in compiledStateGraph.astream({"contentList" : [documentList.page_content for documentList in documentList]}, stream_mode = "values"):

if summary := addableUpdatesDict .get("summary"):

print(summary)

asyncio.run(main())

"""

Apples are typically red in color.

Apples are typically red in color, while blueberries are blue.

Apples are typically red in color, blueberries are blue, and bananas are yellow.

"""

▶ requirements.txt


aiohappyeyeballs==2.4.4
aiohttp==3.11.11
aiosignal==1.3.2
annotated-types==0.7.0
anyio==4.7.0
attrs==24.3.0
beautifulsoup4==4.12.3
bs4==0.0.2
certifi==2024.12.14
charset-normalizer==3.4.0
colorama==0.4.6
dataclasses-json==0.6.7
distro==1.9.0
frozenlist==1.5.0
greenlet==3.1.1
h11==0.14.0
httpcore==1.0.7
httpx==0.28.1
httpx-sse==0.4.0
idna==3.10
jiter==0.8.2
jsonpatch==1.33
jsonpointer==3.0.0
langchain==0.3.13
langchain-community==0.3.13
langchain-core==0.3.28
langchain-openai==0.2.14
langchain-text-splitters==0.3.4
langgraph==0.2.60
langgraph-checkpoint==2.0.9
langgraph-sdk==0.1.48
langsmith==0.2.4
marshmallow==3.23.2
msgpack==1.1.0
multidict==6.1.0
mypy-extensions==1.0.0
numpy==2.2.1
openai==1.58.1
orjson==3.10.12
packaging==24.2
propcache==0.2.1
pydantic==2.10.4
pydantic-settings==2.7.0
pydantic_core==2.27.2
python-dotenv==1.0.1
PyYAML==6.0.2
regex==2024.11.6
requests==2.32.3
requests-toolbelt==1.0.0
sniffio==1.3.1
soupsieve==2.6
SQLAlchemy==2.0.36
tenacity==9.0.0
tiktoken==0.8.0
tqdm==4.67.1
typing-inspect==0.9.0
typing_extensions==4.12.2
urllib3==2.3.0
yarl==1.18.3

aiohappyeyeballs==2.4.4

aiohttp==3.11.11

aiosignal==1.3.2

annotated-types==0.7.0

anyio==4.7.0

attrs==24.3.0

beautifulsoup4==4.12.3

bs4==0.0.2

certifi==2024.12.14

charset-normalizer==3.4.0

colorama==0.4.6

dataclasses-json==0.6.7

distro==1.9.0

frozenlist==1.5.0

greenlet==3.1.1

h11==0.14.0

httpcore==1.0.7

httpx==0.28.1

httpx-sse==0.4.0

idna==3.10

jiter==0.8.2

jsonpatch==1.33

jsonpointer==3.0.0

langchain==0.3.13

langchain-community==0.3.13

langchain-core==0.3.28

langchain-openai==0.2.14

langchain-text-splitters==0.3.4

langgraph==0.2.60

langgraph-checkpoint==2.0.9

langgraph-sdk==0.1.48

langsmith==0.2.4

marshmallow==3.23.2

msgpack==1.1.0

multidict==6.1.0

mypy-extensions==1.0.0

numpy==2.2.1

openai==1.58.1

orjson==3.10.12

packaging==24.2

propcache==0.2.1

pydantic==2.10.4

pydantic-settings==2.7.0

pydantic_core==2.27.2

python-dotenv==1.0.1

PyYAML==6.0.2

regex==2024.11.6

requests==2.32.3

requests-toolbelt==1.0.0

sniffio==1.3.1

soupsieve==2.6

SQLAlchemy==2.0.36

tenacity==9.0.0

tiktoken==0.8.0

tqdm==4.67.1

typing-inspect==0.9.0

typing_extensions==4.12.2

urllib3==2.3.0

yarl==1.18.3

※ pip install python-dotenv langchain langchain-community langchain-openai langgraph bs4 명령을 실행했다.