STREAM Archives - icodebroker

[PYTHON/LANGCHAIN] get_openai_callback 함수 : 단일 채팅 모델에서 스트리밍시 토큰 사용량 구하기

■ get_openai_callback 함수 : 단일 채팅 모델에서 스트리밍시 토큰 사용량 구하기 ※ 비용을 측정할 수 없다. ※ OPENAI_API_KEY 환경 변수 값은 .env

[PYTHON/LANGCHAIN] OpenAI 클래스 : stream 메소드를 사용해 질의 응답 스트리밍하기

■ OpenAI 클래스의 stream 메소드를 사용해 질의 응답을 스트리밍하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶ main.py


from dotenv           import load_dotenv
from langchain_openai import OpenAI

load_dotenv()

openAI = OpenAI(model = "gpt-3.5-turbo-instruct", temperature = 0, max_tokens = 512)

for chunkString in openAI.stream("Write me a 1 verse song about sparkling water."):
    print(chunkString, end = "", flush = True)

"""
Sparkling water, oh so clear
Bubbles dancing, without fear
Refreshing taste, a pure delight
"""

from dotenv import load_dotenv

from langchain_openai import OpenAI

load_dotenv()

openAI = OpenAI(model = "gpt-3.5-turbo-instruct", temperature = 0, max_tokens = 512)

for chunkString in openAI.stream("Write me a 1 verse song about sparkling water."):

print(chunkString, end = "", flush = True)

"""

Sparkling water, oh so clear

Bubbles dancing, without fear

Refreshing taste, a pure delight

"""

[PYTHON/LANGCHAIN] ChatOpenAI 클래스 : stream 메소드에서 stream_options 인자를 사용해 스트리밍시 토큰 사용량 구하기

■ ChatOpenAI 클래스의 stream 메소드에서 stream_options 인자를 사용해 스트리밍시 토큰 사용량을 구하는 방법을 보여준다. ※ 기본적으로 스트림의 마지막 메시지 청크에는 메시지의 response_metadata

[PYTHON/LANGCHAIN] ChatOpenAI 클래스 : with_structured_output 메소드에서 JSON 스키마를 전달해 구조화된 데이터 스트리밍으로 받기

■ ChatOpenAI 클래스의 with_structured_output 메소드에서 JSON 스키마를 전달해 구조화된 데이터를 스트리밍으로 받는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다.

[PYTHON/LANGCHAIN] Iterator 클래스 : 제너레이터 커스텀 함수를 만들고 스트리밍하기

■ Iterator 클래스를 사용해 제너레이터 커스텀 함수를 만들고 스트리밍하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶ main.py


from typing import Iterator, List

from dotenv                        import load_dotenv
from langchain_core.prompts        import ChatPromptTemplate
from langchain_openai              import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

load_dotenv()

def splitIntoList(iterator : Iterator[str]) -> Iterator[List[str]]:
    buffer = ""
    for chunkString in iterator:
        buffer += chunkString
        while "," in buffer:
            commaIndex = buffer.index(",")
            yield [buffer[:commaIndex].strip()]
            buffer = buffer[commaIndex + 1:]
    yield [buffer.strip()]

chatPromptTemplate = ChatPromptTemplate.from_template("Write a comma-separated list of 5 animals similar to: {animal}. Do not include numbers")

chatOpenAI = ChatOpenAI()

runnableSequence = chatPromptTemplate | chatOpenAI | StrOutputParser() | splitIntoList

for chunkList in runnableSequence.stream({"animal" : "bear"}):
    print(chunkList, flush = True)

"""
['lion']
['tiger']
['wolf']
['fox']
['raccoon']
"""

from typing import Iterator, List

from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate

from langchain_openai import ChatOpenAI

from langchain_core.output_parsers import StrOutputParser

load_dotenv()

def splitIntoList(iterator : Iterator[str]) -> Iterator[List[str]]:

buffer = ""

for chunkString in iterator:

buffer += chunkString

while "," in buffer:

commaIndex = buffer.index(",")

yield [buffer[:commaIndex].strip()]

buffer = buffer[commaIndex + 1:]

yield [buffer.strip()]

chatPromptTemplate = ChatPromptTemplate.from_template("Write a comma-separated list of 5 animals similar to: {animal}. Do not include numbers")

chatOpenAI = ChatOpenAI()

runnableSequence = chatPromptTemplate | chatOpenAI | StrOutputParser() | splitIntoList

for chunkList in runnableSequence.stream({"animal" : "bear"}):

print(chunkList, flush = True)

"""

['lion']

['tiger']

['wolf']

['fox']

['raccoon']

"""

[PYTHON/LANGCHAIN] StreamingStdOutCallbackHandler 클래스 : LLM 질의 응답 스트리밍하기

■ StreamingStdOutCallbackHandler 클래스를 사용해 LLM의 질의 응답을 스트리밍하는 방법을 보여준다. ※ OPENAI_API_KEY 환경 변수 값은 .env 파일에 정의한다. ▶ main.py


from dotenv                               import load_dotenv
from langchain_openai                     import OpenAI
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

load_dotenv()

openAI = OpenAI(
    model       = "gpt-3.5-turbo-instruct",
    streaming   = True,
    callbacks   = [StreamingStdOutCallbackHandler()],
    verbose     = True,
    temperature = 0
)

openAI.invoke("즐거운 ChatGPT 생활을 가사로 만들어 주세요.")

"""
ChatGPT, 너는 나의 친구
매일 나를 즐겁게 해주는
나의 소중한 존재
마치 마법처럼 나를 웃게 해주는
너의 말 한마디에 나는 행복해져
우리 함께하는 시간은 너무 소중해
나의 모든 이야기를 들어주는
나의 가장 신뢰할 수 있는 친구
ChatGPT, 너와 함께라면
나는 언제나 행복할 수 있어
감사해, 나의 사랑하는 ChatGPT
"""

from dotenv import load_dotenv

from langchain_openai import OpenAI

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

load_dotenv()

openAI = OpenAI(

model = "gpt-3.5-turbo-instruct",

streaming = True,

callbacks = [StreamingStdOutCallbackHandler()],

verbose = True,

temperature = 0

)

openAI.invoke("즐거운 ChatGPT 생활을 가사로 만들어 주세요.")

"""

ChatGPT, 너는 나의 친구

매일 나를 즐겁게 해주는

나의 소중한 존재

마치 마법처럼 나를 웃게 해주는

너의 말 한마디에 나는 행복해져

우리 함께하는 시간은 너무 소중해

나의 모든 이야기를 들어주는

나의 가장 신뢰할 수 있는 친구

ChatGPT, 너와 함께라면

나는 언제나 행복할 수 있어

감사해, 나의 사랑하는 ChatGPT

"""

▶

[PYTHON/LANGCHAIN] 비스트리밍 구성 요소 구성 LCEL 체인에서 마지막 비스트리밍 단계 이후 부분 출력 스트리밍하기

■ 비스트리밍 구성 요소로 구성된 LCEL 체인에서 마지막 비스트리밍 단계 이후 부분 출력을 스트리밍하는 방법을 보여준다. ▶ main.py


import os

from langchain_core.prompts           import ChatPromptTemplate
from langchain_openai                 import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.runnables         import RunnablePassthrough
from langchain_openai                 import ChatOpenAI
from langchain_core.output_parsers    import StrOutputParser

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

chatPromptTemplateString = """Answer the question based only on the following context :
{context}

Question : {question}
"""

chatPromptTemplate = ChatPromptTemplate.from_template(chatPromptTemplateString)

faiss = FAISS.from_texts(
    ["harrison worked at kensho", "harrison likes spicy food"],
    embedding = OpenAIEmbeddings(),
)

vectorStoreRetriever = faiss.as_retriever()

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

runnableSequence = (
    {
        "context"  : vectorStoreRetriever.with_config(run_name = "Docs"),
        "question" : RunnablePassthrough()
    }
    | chatPromptTemplate
    | chatOpenAI
    | StrOutputParser()
)

for chunkString in runnableSequence.stream("Where did harrison work? Write 3 made up sentences about this place."):
    print(chunkString, end = "", flush = True)

print()

"""
Harrison worked at a trendy tech start-up called Kensho located in the heart of Silicon Valley. The office had a sleek and modern design, with open workspaces and collaborative areas for brainstorming. At Kensho, employees enjoyed perks like catered lunches, ping pong tables, and regular team-building activities.
"""

import os

from langchain_core.prompts import ChatPromptTemplate

from langchain_openai import OpenAIEmbeddings

from langchain_community.vectorstores import FAISS

from langchain_core.runnables import RunnablePassthrough

from langchain_openai import ChatOpenAI

from langchain_core.output_parsers import StrOutputParser

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

chatPromptTemplateString = """Answer the question based only on the following context :

{context}

Question : {question}

"""

chatPromptTemplate = ChatPromptTemplate.from_template(chatPromptTemplateString)

faiss = FAISS.from_texts(

["harrison worked at kensho", "harrison likes spicy food"],

embedding = OpenAIEmbeddings(),

)

vectorStoreRetriever = faiss.as_retriever()

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

runnableSequence = (

{

"context" : vectorStoreRetriever.with_config(run_name = "Docs"),

"question" : RunnablePassthrough()

}

| chatPromptTemplate

| chatOpenAI

| StrOutputParser()

)

for chunkString in runnableSequence.stream("Where did harrison work? Write 3 made up sentences about this place."):

print(chunkString, end = "", flush = True)

print()

"""

Harrison worked at a trendy tech start-up called Kensho located in the heart of Silicon Valley. The office had a sleek and modern design, with open workspaces and collaborative areas for brainstorming. At Kensho, employees enjoyed perks like catered lunches, ping pong tables, and regular team-building activities.

"""

▶ requirements.txt


aiohttp==3.9.5
aiosignal==1.3.1
annotated-types==0.7.0
anyio==4.4.0
async-timeout==4.0.3
attrs==23.2.0
certifi==2024.6.2
charset-normalizer==3.3.2
dataclasses-json==0.6.7
distro==1.9.0
exceptiongroup==1.2.1
faiss-gpu==1.7.2
frozenlist==1.4.1
greenlet==3.0.3
h11==0.14.0
httpcore==1.0.5
httpx==0.27.0
idna==3.7
jsonpatch==1.33
jsonpointer==3.0.0
langchain==0.2.5
langchain-community==0.2.5
langchain-core==0.2.8
langchain-openai==0.1.8
langchain-text-splitters==0.2.1
langsmith==0.1.79
marshmallow==3.21.3
multidict==6.0.5
mypy-extensions==1.0.0
numpy==1.26.4
openai==1.34.0
orjson==3.10.5
packaging==24.1
pydantic==2.7.4
pydantic_core==2.18.4
PyYAML==6.0.1
regex==2024.5.15
requests==2.32.3
sniffio==1.3.1
SQLAlchemy==2.0.30
tenacity==8.4.1
tiktoken==0.7.0
tqdm==4.66.4
typing-inspect==0.9.0
typing_extensions==4.12.2
urllib3==2.2.2
yarl==1.9.4

aiohttp==3.9.5

aiosignal==1.3.1

annotated-types==0.7.0

anyio==4.4.0

async-timeout==4.0.3

attrs==23.2.0

certifi==2024.6.2

charset-normalizer==3.3.2

dataclasses-json==0.6.7

distro==1.9.0

exceptiongroup==1.2.1

faiss-gpu==1.7.2

frozenlist==1.4.1

greenlet==3.0.3

h11==0.14.0

httpcore==1.0.5

httpx==0.27.0

idna==3.7

jsonpatch==1.33

jsonpointer==3.0.0

langchain==0.2.5

langchain-community==0.2.5

langchain-core==0.2.8

langchain-openai==0.1.8

langchain-text-splitters==0.2.1

langsmith==0.1.79

marshmallow==3.21.3

multidict==6.0.5

mypy-extensions==1.0.0

numpy==1.26.4

openai==1.34.0

orjson==3.10.5

packaging==24.1

pydantic==2.7.4

pydantic_core==2.18.4

PyYAML==6.0.1

regex==2024.5.15

requests==2.32.3

sniffio==1.3.1

SQLAlchemy==2.0.30

tenacity==8.4.1

tiktoken==0.7.0

tqdm==4.66.4

typing-inspect==0.9.0

typing_extensions==4.12.2

urllib3==2.2.2

yarl==1.9.4

[PYTHON/LANGCHAIN] RunnableSequence 클래스 : stream 메소드를 사용해 질의 응답 스트리밍하기

■ RunnableSequence 클래스의 stream 메소드를 사용해 질의 응답을 스트리밍하는 방법을 보여준다. ▶ main.py


import os

from langchain_core.prompts        import ChatPromptTemplate
from langchain_openai              import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

chatPromptTemplate = ChatPromptTemplate.from_template("tell me a joke about {topic}")

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

strOutputParser = StrOutputParser()

runnableSequence = chatPromptTemplate | chatOpenAI | strOutputParser

for responseChunkString in runnableSequence.stream({"topic" : "parrot"}):
    print(responseChunkString, end = "|", flush = True)

"""
|Why| did| the| par|rot| wear| a| rain|coat|?

|Because| he| wanted| to| be| poly|uns|aturated|!||
"""

import os

from langchain_core.prompts import ChatPromptTemplate

from langchain_openai import ChatOpenAI

from langchain_core.output_parsers import StrOutputParser

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

chatPromptTemplate = ChatPromptTemplate.from_template("tell me a joke about {topic}")

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

strOutputParser = StrOutputParser()

runnableSequence = chatPromptTemplate | chatOpenAI | strOutputParser

for responseChunkString in runnableSequence.stream({"topic" : "parrot"}):

print(responseChunkString, end = "|", flush = True)

"""

|Why| did| the| par|rot| wear| a| rain|coat|?

|Because| he| wanted| to| be| poly|uns|aturated|!||

"""

▶ requirements.txt


annotated-types==0.7.0
anyio==4.4.0
certifi==2024.6.2
charset-normalizer==3.3.2
distro==1.9.0
exceptiongroup==1.2.1
h11==0.14.0
httpcore==1.0.5
httpx==0.27.0
idna==3.7
jsonpatch==1.33
jsonpointer==3.0.0
langchain-core==0.2.7
langchain-openai==0.1.8
langsmith==0.1.77
openai==1.34.0
orjson==3.10.5
packaging==24.1
pydantic==2.7.4
pydantic_core==2.18.4
PyYAML==6.0.1
regex==2024.5.15
requests==2.32.3
sniffio==1.3.1
tenacity==8.4.1
tiktoken==0.7.0
tqdm==4.66.4
typing_extensions==4.12.2
urllib3==2.2.2

annotated-types==0.7.0

anyio==4.4.0

certifi==2024.6.2

charset-normalizer==3.3.2

distro==1.9.0

exceptiongroup==1.2.1

h11==0.14.0

httpcore==1.0.5

httpx==0.27.0

idna==3.7

jsonpatch==1.33

jsonpointer==3.0.0

langchain-core==0.2.7

langchain-openai==0.1.8

langsmith==0.1.77

openai==1.34.0

orjson==3.10.5

packaging==24.1

pydantic==2.7.4

pydantic_core==2.18.4

PyYAML==6.0.1

regex==2024.5.15

requests==2.32.3

sniffio==1.3.1

tenacity==8.4.1

tiktoken==0.7.0

tqdm==4.66.4

typing_extensions==4.12.2

urllib3==2.2.2

※ pip install langchain-openai 명령을

[PYTHON/LANGCHAIN] ChatOpenAI 클래스 : stream 메소드를 사용해 질의 응답 스트리밍하기

■ ChatOpenAI 클래스의 stream 메소드를 사용해 질의 응답을 스트리밍하는 방법을 보여준다. ▶ main.py


import os

from langchain_openai import ChatOpenAI

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

for aiMessageChunk in chatOpenAI.stream("what color is the sky?"):
    print(aiMessageChunk.content, end = "|", flush = True)

"""
|The| color| of| the| sky| varies| depending| on| the| time| of| day| and| weather| conditions|.| During| the| day|,| the| sky| is| typically| blue|,| while| at| sunrise| and| sunset| it| can| appear| orange|,| pink|,| or| purple|.| At| night|,| the| sky| appears| black| with| stars| visible|.||
"""

import os

from langchain_openai import ChatOpenAI

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

for aiMessageChunk in chatOpenAI.stream("what color is the sky?"):

print(aiMessageChunk.content, end = "|", flush = True)

"""

|The| color| of| the| sky| varies| depending| on| the| time| of| day| and| weather| conditions|.| During| the| day|,| the| sky| is| typically| blue|,| while| at| sunrise| and| sunset| it| can| appear| orange|,| pink|,| or| purple|.| At| night|,| the| sky| appears| black| with| stars| visible|.||

"""

▶ requirements.txt


annotated-types==0.7.0
anyio==4.4.0
certifi==2024.6.2
charset-normalizer==3.3.2
distro==1.9.0
exceptiongroup==1.2.1
h11==0.14.0
httpcore==1.0.5
httpx==0.27.0
idna==3.7
jsonpatch==1.33
jsonpointer==3.0.0
langchain-core==0.2.7
langchain-openai==0.1.8
langsmith==0.1.77
openai==1.34.0
orjson==3.10.5
packaging==24.1
pydantic==2.7.4
pydantic_core==2.18.4
PyYAML==6.0.1
regex==2024.5.15
requests==2.32.3
sniffio==1.3.1
tenacity==8.4.1
tiktoken==0.7.0
tqdm==4.66.4
typing_extensions==4.12.2
urllib3==2.2.2

annotated-types==0.7.0

anyio==4.4.0

certifi==2024.6.2

charset-normalizer==3.3.2

distro==1.9.0

exceptiongroup==1.2.1

h11==0.14.0

httpcore==1.0.5

httpx==0.27.0

idna==3.7

jsonpatch==1.33

jsonpointer==3.0.0

langchain-core==0.2.7

langchain-openai==0.1.8

langsmith==0.1.77

openai==1.34.0

orjson==3.10.5

packaging==24.1

pydantic==2.7.4

pydantic_core==2.18.4

PyYAML==6.0.1

regex==2024.5.15

requests==2.32.3

sniffio==1.3.1

tenacity==8.4.1

tiktoken==0.7.0

tqdm==4.66.4

typing_extensions==4.12.2

urllib3==2.2.2

※ pip install langchain-openai 명령을

[PYTHON/LANGCHAIN] RunnableSequence 클래스 : stream 메소드를 사용해 도구 호출 스트리밍하기

■ RunnableSequence 클래스의 stream 메소드를 사용해 도구 호출을 스트리밍하는 방법을 보여준다. ▶ main.py


import os

from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai           import ChatOpenAI

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

# 여기의 독스트링은 클래스 이름과 함께 모델에 전달되므로 매우 중요하다.
class Add(BaseModel):
    """Add two integers together."""

    a : int = Field(..., description = "First integer" )
    b : int = Field(..., description = "Second integer")

class Multiply(BaseModel):
    """Multiply two integers together."""

    a : int = Field(..., description = "First integer" )
    b : int = Field(..., description = "Second integer")

toolList = [Add, Multiply]

chatOpenAI = ChatOpenAI(model="gpt-3.5-turbo-0125")

toolRunnableSequence = chatOpenAI.bind_tools(toolList)

for aiMessageChunk in toolRunnableSequence.stream("What is 3 * 12? Also, what is 11 + 49?"):
    print(aiMessageChunk.tool_call_chunks)

"""
[]
[{'name': 'Multiply', 'args': '', 'id': 'call_9AdugnCq1SHbB8GbU30xeqzh', 'index': 0}]
[{'name': None, 'args': '{"a"', 'id': None, 'index': 0}]
[{'name': None, 'args': ': 3, ', 'id': None, 'index': 0}]
[{'name': None, 'args': '"b": 1', 'id': None, 'index': 0}]
[{'name': None, 'args': '2}', 'id': None, 'index': 0}]
[{'name': 'Add', 'args': '', 'id': 'call_o1rtOfF5CgHhEPZKeZmMAZZE', 'index': 1}]
[{'name': None, 'args': '{"a"', 'id': None, 'index': 1}]
[{'name': None, 'args': ': 11,', 'id': None, 'index': 1}]
[{'name': None, 'args': ' "b": ', 'id': None, 'index': 1}]
[{'name': None, 'args': '49}', 'id': None, 'index': 1}]
[]
"""

import os

from langchain_core.pydantic_v1 import BaseModel, Field

from langchain_openai import ChatOpenAI

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

# 여기의 독스트링은 클래스 이름과 함께 모델에 전달되므로 매우 중요하다.

class Add(BaseModel):

"""Add two integers together."""

a : int = Field(..., description = "First integer" )

b : int = Field(..., description = "Second integer")

class Multiply(BaseModel):

"""Multiply two integers together."""

a : int = Field(..., description = "First integer" )

b : int = Field(..., description = "Second integer")

toolList = [Add, Multiply]

chatOpenAI = ChatOpenAI(model="gpt-3.5-turbo-0125")

toolRunnableSequence = chatOpenAI.bind_tools(toolList)

for aiMessageChunk in toolRunnableSequence.stream("What is 3 * 12? Also, what is 11 + 49?"):

print(aiMessageChunk.tool_call_chunks)

"""

[]

[{'name': 'Multiply', 'args': '', 'id': 'call_9AdugnCq1SHbB8GbU30xeqzh', 'index': 0}]

[{'name': None, 'args': '{"a"', 'id': None, 'index': 0}]

[{'name': None, 'args': ': 3, ', 'id': None, 'index': 0}]

[{'name': None, 'args': '"b": 1', 'id': None, 'index': 0}]

[{'name': None, 'args': '2}', 'id': None, 'index': 0}]

[{'name': 'Add', 'args': '', 'id': 'call_o1rtOfF5CgHhEPZKeZmMAZZE', 'index': 1}]

[{'name': None, 'args': '{"a"', 'id': None, 'index': 1}]

[{'name': None, 'args': ': 11,', 'id': None, 'index': 1}]

[{'name': None, 'args': ' "b": ', 'id': None, 'index': 1}]

[{'name': None, 'args': '49}', 'id': None, 'index': 1}]

[]

"""

▶ requirements.txt


aiohttp==3.9.5
aiosignal==1.3.1
annotated-types==0.7.0
anyio==4.4.0
async-timeout==4.0.3
attrs==23.2.0
certifi==2024.6.2
charset-normalizer==3.3.2
dataclasses-json==0.6.7
distro==1.9.0
exceptiongroup==1.2.1
frozenlist==1.4.1
greenlet==3.0.3
h11==0.14.0
httpcore==1.0.5
httpx==0.27.0
idna==3.7
jsonpatch==1.33
jsonpointer==3.0.0
langchain==0.2.5
langchain-community==0.2.5
langchain-core==0.2.7
langchain-experimental==0.0.61
langchain-openai==0.1.8
langchain-text-splitters==0.2.1
langsmith==0.1.77
marshmallow==3.21.3
multidict==6.0.5
mypy-extensions==1.0.0
numpy==1.26.4
openai==1.34.0
orjson==3.10.5
packaging==24.1
pydantic==2.7.4
pydantic_core==2.18.4
PyYAML==6.0.1
regex==2024.5.15
requests==2.32.3
sniffio==1.3.1
SQLAlchemy==2.0.30
tenacity==8.3.0
tiktoken==0.7.0
tqdm==4.66.4
typing-inspect==0.9.0
typing_extensions==4.12.2
urllib3==2.2.1
yarl==1.9.4

aiohttp==3.9.5

aiosignal==1.3.1

annotated-types==0.7.0

anyio==4.4.0

async-timeout==4.0.3

attrs==23.2.0

certifi==2024.6.2

charset-normalizer==3.3.2

dataclasses-json==0.6.7

distro==1.9.0

exceptiongroup==1.2.1

frozenlist==1.4.1

greenlet==3.0.3

h11==0.14.0

httpcore==1.0.5

httpx==0.27.0

idna==3.7

jsonpatch==1.33

jsonpointer==3.0.0

langchain==0.2.5

langchain-community==0.2.5

langchain-core==0.2.7

langchain-experimental==0.0.61

langchain-openai==0.1.8

langchain-text-splitters==0.2.1

langsmith==0.1.77

marshmallow==3.21.3

multidict==6.0.5

mypy-extensions==1.0.0

numpy==1.26.4

openai==1.34.0

orjson==3.10.5

packaging==24.1

pydantic==2.7.4

pydantic_core==2.18.4

PyYAML==6.0.1

regex==2024.5.15

requests==2.32.3

sniffio==1.3.1

SQLAlchemy==2.0.30

tenacity==8.3.0

tiktoken==0.7.0

tqdm==4.66.4

typing-inspect==0.9.0

typing_extensions==4.12.2

urllib3==2.2.1

yarl==1.9.4

※ pip install langchain langchain_experimental

[PYTHON/LANGCHAIN] RunnableSequence 클래스 : stream 메소드를 사용해 구조화된 데이터를 스트리밍으로 받기

■ RunnableSequence 클래스의 stream 메소드를 사용해 구조화된 데이터를 스트리밍으로 받는 방법을 보여준다. ▶ main.py


import os

from typing import Optional
from typing import Union

from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai           import ChatOpenAI

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

class Joke(BaseModel):
    """Joke to tell user."""

    setup     : str           = Field(description = "The setup of the joke"              )
    punchline : str           = Field(description = "The punchline to the joke"          )
    rating    : Optional[int] = Field(description = "How funny the joke is, from 1 to 10")

class ConversationalResponse(BaseModel):
    """Respond in a conversational manner. Be kind and helpful."""

    response : str = Field(description = "A conversational response to the user's query")

class Response(BaseModel):
    output : Union[Joke, ConversationalResponse]

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

runnableSequence = chatOpenAI.with_structured_output(Response)

for response in runnableSequence.stream("Tell me a joke about cats"):
    print(response)

"""
output=Joke(setup='Why was the cat sitting on the computer?', punchline='', rating=None)
output=Joke(setup='Why was the cat sitting on the computer?', punchline='To', rating=None)
output=Joke(setup='Why was the cat sitting on the computer?', punchline='To keep', rating=None)
output=Joke(setup='Why was the cat sitting on the computer?', punchline='To keep an', rating=None)
output=Joke(setup='Why was the cat sitting on the computer?', punchline='To keep an eye', rating=None)
output=Joke(setup='Why was the cat sitting on the computer?', punchline='To keep an eye on', rating=None)
output=Joke(setup='Why was the cat sitting on the computer?', punchline='To keep an eye on the', rating=None)
output=Joke(setup='Why was the cat sitting on the computer?', punchline='To keep an eye on the mouse', rating=None)
output=Joke(setup='Why was the cat sitting on the computer?', punchline='To keep an eye on the mouse!', rating=None)
output=Joke(setup='Why was the cat sitting on the computer?', punchline='To keep an eye on the mouse!', rating=8)
"""

import os

from typing import Optional

from typing import Union

from langchain_core.pydantic_v1 import BaseModel, Field

from langchain_openai import ChatOpenAI

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

class Joke(BaseModel):

"""Joke to tell user."""

setup : str = Field(description = "The setup of the joke" )

punchline : str = Field(description = "The punchline to the joke" )

rating : Optional[int] = Field(description = "How funny the joke is, from 1 to 10")

class ConversationalResponse(BaseModel):

"""Respond in a conversational manner. Be kind and helpful."""

response : str = Field(description = "A conversational response to the user's query")

class Response(BaseModel):

output : Union[Joke, ConversationalResponse]

chatOpenAI = ChatOpenAI(model = "gpt-3.5-turbo-0125")

runnableSequence = chatOpenAI.with_structured_output(Response)

for response in runnableSequence.stream("Tell me a joke about cats"):

print(response)

"""

output=Joke(setup='Why was the cat sitting on the computer?', punchline='', rating=None)

output=Joke(setup='Why was the cat sitting on the computer?', punchline='To', rating=None)

output=Joke(setup='Why was the cat sitting on the computer?', punchline='To keep', rating=None)

output=Joke(setup='Why was the cat sitting on the computer?', punchline='To keep an', rating=None)

output=Joke(setup='Why was the cat sitting on the computer?', punchline='To keep an eye', rating=None)

output=Joke(setup='Why was the cat sitting on the computer?', punchline='To keep an eye on', rating=None)

output=Joke(setup='Why was the cat sitting on the computer?', punchline='To keep an eye on the', rating=None)

output=Joke(setup='Why was the cat sitting on the computer?', punchline='To keep an eye on the mouse', rating=None)

output=Joke(setup='Why was the cat sitting on the computer?', punchline='To keep an eye on the mouse!', rating=None)

output=Joke(setup='Why was the cat sitting on the computer?', punchline='To keep an eye on the mouse!', rating=8)

"""

▶ requirements.txt


annotated-types==0.7.0
anyio==4.4.0
certifi==2024.6.2
charset-normalizer==3.3.2
distro==1.9.0
exceptiongroup==1.2.1
h11==0.14.0
httpcore==1.0.5
httpx==0.27.0
idna==3.7
jsonpatch==1.33
jsonpointer==3.0.0
langchain-core==0.2.7
langchain-openai==0.1.8
langsmith==0.1.77
openai==1.34.0
orjson==3.10.5
packaging==24.1
pydantic==2.7.4
pydantic_core==2.18.4
PyYAML==6.0.1
regex==2024.5.15
requests==2.32.3
sniffio==1.3.1
tenacity==8.3.0
tiktoken==0.7.0
tqdm==4.66.4
typing_extensions==4.12.2
urllib3==2.2.1

annotated-types==0.7.0

anyio==4.4.0

certifi==2024.6.2

charset-normalizer==3.3.2

distro==1.9.0

exceptiongroup==1.2.1

h11==0.14.0

httpcore==1.0.5

httpx==0.27.0

idna==3.7

jsonpatch==1.33

jsonpointer==3.0.0

langchain-core==0.2.7

langchain-openai==0.1.8

langsmith==0.1.77

openai==1.34.0

orjson==3.10.5

packaging==24.1

pydantic==2.7.4

pydantic_core==2.18.4

PyYAML==6.0.1

regex==2024.5.15

requests==2.32.3

sniffio==1.3.1

tenacity==8.3.0

tiktoken==0.7.0

tqdm==4.66.4

typing_extensions==4.12.2

urllib3==2.2.1

※ pip install langchain-openai

[PYTHON/LANGCHAIN] StreamingStdOutCallbackHandler 클래스 : 표준 출력 스트리밍하기

■ StreamingStdOutCallbackHandler 클래스를 사용해 표준 출력을 스트리밍하는 방법을 보여준다. ▶ main.py


from langchain_core.prompts   import PromptTemplate
from langchain_core.callbacks import StreamingStdOutCallbackHandler
from langchain_community.llms import GPT4All

# 프롬프트 템플리트를 설정한다.
templateString = """Question : {question}

Answer : Let's think step by step."""

promptTemplate = PromptTemplate.from_template(templateString)

# GPT4All 모델을 설정한다.
callbackHandlerList = [StreamingStdOutCallbackHandler()]

gpt4All = GPT4All(model = "./llama-2-13b-chat.Q4_0.gguf", backend = "llama", callbacks = callbackHandlerList, verbose = True)

# 실행 체인을 설정한다.
runnableSequence = promptTemplate | gpt4All

# 질의 응답을 실행한다.
question = "What NFL team won the Super Bowl in the year Justin Bieber was born?"

resultString = runnableSequence.invoke(question)

print()
print("질문 : ", question)
print()
print("답변 : ", resultString.strip())

"""
질문 :  What NFL team won the Super Bowl in the year Justin Bieber was born?

답변 :  Justin Bieber was born on March 1, 1994. The Super Bowl that year was played on January 29, 1994. So, the NFL team that won the Super Bowl in the year Justin Bieber was born is...
!
"""

from langchain_core.prompts import PromptTemplate

from langchain_core.callbacks import StreamingStdOutCallbackHandler

from langchain_community.llms import GPT4All

# 프롬프트 템플리트를 설정한다.

templateString = """Question : {question}

Answer : Let's think step by step."""

promptTemplate = PromptTemplate.from_template(templateString)

# GPT4All 모델을 설정한다.

callbackHandlerList = [StreamingStdOutCallbackHandler()]

gpt4All = GPT4All(model = "./llama-2-13b-chat.Q4_0.gguf", backend = "llama", callbacks = callbackHandlerList, verbose = True)

# 실행 체인을 설정한다.

runnableSequence = promptTemplate | gpt4All

# 질의 응답을 실행한다.

question = "What NFL team won the Super Bowl in the year Justin Bieber was born?"

resultString = runnableSequence.invoke(question)

print()

print("질문 : ", question)

print()

print("답변 : ", resultString.strip())

"""

질문 : What NFL team won the Super Bowl in the year Justin Bieber was born?

답변 : Justin Bieber was born on March 1, 1994. The Super Bowl that year was played on January 29, 1994. So, the NFL team that won the Super Bowl in the year Justin Bieber was born is...

"""

※ llama-2-13b-chat.Q4_0.gguf 모델 파일은 https://gpt4all.io 웹 사이트에서 다운로드 받은 [Desktop

[C#/COMMON/.NET6] Stream 클래스 : CopyTo 메소드를 사용해 스트림에서 바이트 배열 구하기

■ Stream 클래스의 CopyTo 메소드를 사용해 스트림에서 바이트 배열을 구하는 방법을 보여준다. ▶ 예제 코드 (C#)


#region 바이트 배열 구하기 - GetByteArray(sourceStream)

/// <summary>
/// 바이트 배열 구하기
/// </summary>
/// <param name="sourceStream">소스 스트림</param>
/// <returns>바이트 배열</returns>
public byte[] GetByteArray(Stream sourceStream)
{
    using(MemoryStream targetStream = new MemoryStream())
    {
        sourceStream.CopyTo(targetStream);

        return targetStream.ToArray();
    }
}

#endregion

#region 바이트 배열 구하기 - GetByteArray(sourceStream)

/// <summary>

/// 바이트 배열 구하기

/// </summary>

/// <param name="sourceStream">소스 스트림</param>

/// <returns>바이트 배열</returns>

public byte[] GetByteArray(Stream sourceStream)

{

using(MemoryStream targetStream = new MemoryStream())

{

sourceStream.CopyTo(targetStream);

return targetStream.ToArray();

}

#endregion

[C#/COMMON/.NET6] MemoryStream 클래스 : Write 메소드를 사용해 스트림에서 바이트 배열 구하기

■ MemoryStream 클래스의 Write 메소드를 사용해 스트림에서 바이트 배열을 구하는 방법을 보여준다. ▶ 예제 코드 (C#)


#region 바이트 배열 구하기 - GetByteArray(sourceStream)

/// <summary>
/// 바이트 배열 구하기
/// </summary>
/// <param name="sourceStream">소스 스트림</param>
/// <returns>바이트 배열</returns>
public byte[] GetByteArray(Stream sourceStream)
{
    byte[] bufferByteArray = new byte[16384]; // 16×1024

    using(MemoryStream targetStream = new MemoryStream())
    {
        int countRead;

        while((countRead = sourceStream.Read(bufferByteArray, 0, bufferByteArray.Length)) > 0)
        {
            targetStream.Write(bufferByteArray, 0, countRead);
        }

        return targetStream.ToArray();
    }
}

#endregion

#region 바이트 배열 구하기 - GetByteArray(sourceStream)

/// <summary>

/// 바이트 배열 구하기

/// </summary>

/// <param name="sourceStream">소스 스트림</param>

/// <returns>바이트 배열</returns>

public byte[] GetByteArray(Stream sourceStream)

{

byte[] bufferByteArray = new byte[16384]; // 16×1024

using(MemoryStream targetStream = new MemoryStream())

{

int countRead;

while((countRead = sourceStream.Read(bufferByteArray, 0, bufferByteArray.Length)) > 0)

{

targetStream.Write(bufferByteArray, 0, countRead);

}

return targetStream.ToArray();

}

#endregion

[C#/COMMON/.NET6] BinaryWriter 클래스 : Write 메소드를 사용해 바이트 배열을 스트림에 쓰기

■ BinaryWriter 클래스의 Write 메소드를 사용해 바이트 배열을 스트림에 쓰는 방법을 보여준다. ▶ 예제 코드 (C#)


#region 쓰기 - Write(targetStream, sourceByteArray)

/// <summary>
/// 쓰기
/// </summary>
/// <param name="targetStream">타겟 스트림</param>
/// <param name="sourceByteArray">소스 바이트 배열</param>
public void Write(Stream targetStream, byte[] sourceByteArray)
{
    using(BinaryWriter writer = new BinaryWriter(targetStream))
    {
        writer.Write(sourceByteArray);
    }
}

#endregion

#region 쓰기 - Write(targetStream, sourceByteArray)

/// <summary>

/// 쓰기

/// </summary>

/// <param name="targetStream">타겟 스트림</param>

/// <param name="sourceByteArray">소스 바이트 배열</param>

public void Write(Stream targetStream, byte[] sourceByteArray)

{

using(BinaryWriter writer = new BinaryWriter(targetStream))

{

writer.Write(sourceByteArray);

}

#endregion

[C#/COMMON] Stream 클래스 : CopyTo 메소드를 사용해 스트림 복사하기

■ Stream 클래스의 CopyTo 메소드를 사용해 스트림을 복사하는 방법을 보여준다. ▶ 예제 코드 (C#)


using System.IO;

#region 스트림 복사하기 - CopyStream(source, target)

/// <summary>
/// 스트림 복사하기
/// </summary>
/// <param name="source">소스 스트림</param>
/// <param name="target">타겟 스트림</param>
public void CopyStream(Stream source, Stream target)
{
    source.CopyTo(target);
}

#endregion

using System.IO;

#region 스트림 복사하기 - CopyStream(source, target)

/// <summary>

/// 스트림 복사하기

/// </summary>

/// <param name="source">소스 스트림</param>

/// <param name="target">타겟 스트림</param>

public void CopyStream(Stream source, Stream target)

{

source.CopyTo(target);

}

#endregion

[C#/COMMON/.NET6] MemoryStream 클래스 : Write 메소드를 사용해 바이트 배열 병합하기

■ MemoryStream 클래스의 Write 메소드를 사용해 바이트 배열을 병합하는 방법을 보여준다. ▶ 예제 코드 (C#)


#region 병합하기 - Merage(sourceByteArrayArray)

/// <summary>
/// 병합하기
/// </summary>
/// <param name="sourceByteArrayArray">소스 바이트 배열 배열</param>
/// <returns>병합 바이트 배열</returns>
public byte[] Merge(params byte[][]  sourceByteArrayArray)
{
    using(MemoryStream targetStream = new MemoryStream())
    {
        foreach(byte[] sourceByteArray in sourceByteArrayArray)
        {
            targetStream.Write(sourceByteArray);
        }

        return targetStream.ToArray();
    }
}

#endregion

#region 병합하기 - Merage(sourceByteArrayArray)

/// <summary>

/// 병합하기

/// </summary>

/// <param name="sourceByteArrayArray">소스 바이트 배열 배열</param>

/// <returns>병합 바이트 배열</returns>

public byte[] Merge(params byte[][] sourceByteArrayArray)

{

using(MemoryStream targetStream = new MemoryStream())

{

foreach(byte[] sourceByteArray in sourceByteArrayArray)

{

targetStream.Write(sourceByteArray);

}

return targetStream.ToArray();

}

#endregion

[C#/COMMON] Stream 클래스 : 스트림 복사하기

■ Stream 클래스를 사용해 스트림을 복사하는 방법을 보여준다. ▶ 예제 코드 (C#)


using System;
using System.IO;

#region 스트림 복사하기 - CopyStream(source, target, copySize)

/// <summary>
/// 스트림 복사하기
/// </summary>
/// <param name="source">소스 스트림</param>
/// <param name="target">타겟 스트림</param>
/// <param name="copySize">복사 크기</param>
/// <returns>복사 바이트 수</returns>
public int CopyStream(Stream source, Stream target, int copySize)
{
    byte[] bufferByteArray = new byte[32763];

    int byteCountCopied = 0;

    while(copySize > 0)
    {
        int byteCountRead = source.Read(bufferByteArray, 0, Math.Min(bufferByteArray.Length, copySize));

        if(byteCountRead == 0)
        {
            break;
        }

        target.Write(bufferByteArray, 0, byteCountRead);

        copySize -= byteCountRead;

        byteCountCopied += byteCountRead;
    }

    return byteCountCopied;
}

#endregion

using System;

using System.IO;

#region 스트림 복사하기 - CopyStream(source, target, copySize)

/// <summary>

/// 스트림 복사하기

/// </summary>

/// <param name="source">소스 스트림</param>

/// <param name="target">타겟 스트림</param>

/// <param name="copySize">복사 크기</param>

/// <returns>복사 바이트 수</returns>

public int CopyStream(Stream source, Stream target, int copySize)

{

byte[] bufferByteArray = new byte[32763];

int byteCountCopied = 0;

while(copySize > 0)

{

int byteCountRead = source.Read(bufferByteArray, 0, Math.Min(bufferByteArray.Length, copySize));

if(byteCountRead == 0)

{

break;

}

target.Write(bufferByteArray, 0, byteCountRead);

copySize -= byteCountRead;

byteCountCopied += byteCountRead;

}

return byteCountCopied;

}

#endregion

[C#/WEB API/.NET5] 스트림을 사용해 대용량 파일 업로드 하기

■ 스트림을 사용해 대용량 파일을 업로드하는 방법을 보여준다. [TestServer 프로젝트] ▶ launchSettings.json


{
    "$schema"     : "http://json.schemastore.org/launchsettings.json",
    "iisSettings" :
    {
        "windowsAuthentication"   : false,
        "anonymousAuthentication" : true,
        "iisExpress"              :
        {
            "applicationUrl" : "http://localhost:41483",
            "sslPort"        : 44362
        }
    },
    "profiles" :
    {
        "IIS Express" :
        {
            "commandName"          : "IISExpress",
            "launchBrowser"        : true,
            "launchUrl"            : "swagger",
            "environmentVariables" :
            {
                "ASPNETCORE_ENVIRONMENT" : "Development"
            }
        },
        "TestServer" :
        {
            "commandName"          : "Project",
            "dotnetRunMessages"    : "true",
            "launchBrowser"        : true,
            "launchUrl"            : "swagger",
            "applicationUrl"       : "https://localhost:5001;http://localhost:5000",
            "environmentVariables" :
            {
                "ASPNETCORE_ENVIRONMENT" : "Development"
            }
        }
    }
}

{

"$schema" : "http://json.schemastore.org/launchsettings.json",

"iisSettings" :

{

"windowsAuthentication" : false,

"anonymousAuthentication" : true,

"iisExpress" :

{

"applicationUrl" : "http://localhost:41483",

"sslPort" : 44362

}

"profiles" :

{

"IIS Express" :

{

"commandName" : "IISExpress",

"launchBrowser" : true,

"launchUrl" : "swagger",

"environmentVariables" :

{

"ASPNETCORE_ENVIRONMENT" : "Development"

}

"TestServer" :

{

"commandName" : "Project",

"dotnetRunMessages" : "true",

"launchBrowser" : true,

"launchUrl" : "swagger",

"applicationUrl" : "https://localhost:5001;http://localhost:5000",

"environmentVariables" :

{

"ASPNETCORE_ENVIRONMENT" : "Development"

}

▶ appsettings.json


{
    "Logging" :
    {
        "LogLevel" :
        {
            "Default"                    : "Information",
            "Microsoft"                  : "Warning",
            "Microsoft.Hosting.Lifetime" : "Information"
        }
    },
    "AllowedHosts" : "*",
    "APIKey"       : "A7A54AD3-91E1-4C3A-98C6-A17E39C9EA3D"
}

{

"Logging" :

{

"LogLevel" :

{

"Default" : "Information",

"Microsoft" : "Warning",

"Microsoft.Hosting.Lifetime" : "Information"

}

"AllowedHosts" : "*",

"APIKey" : "A7A54AD3-91E1-4C3A-98C6-A17E39C9EA3D"

}

▶ web.config


<?xml version="1.0" encoding="utf-8"?>
<configuration>
    <system.webServer>
        <security>
            <requestFiltering>
                <requestLimits maxAllowedContentLength="2147483648" />
            </requestFiltering>
        </security>
    </system.webServer>
</configuration>

<?xml version="1.0" encoding="utf-8"?>

<system.webServer>

</requestFiltering>

</security>

</system.webServer>

</configuration>

※ IIS Express

[DART] Stream 클래스 : async*/yield 키워드 사용하기

■ Stream<T> 클래스에서 async*/yield 키워드를 사용하는 방법을 보여준다. ▶ 예제 코드 (DART)


import 'dart:async';

void main() async {
  Stream<String> stream = await getStream(['A', 'B', 'C', 'D', 'E']);

  int count = await Print(stream);

  print(count);
}

const Duration duration = Duration(seconds: 1);

Stream<String> getStream(Iterable<String> itemIterable) async* {
  for (String item in itemIterable) {
    await Future.delayed(duration);

    yield item;
  }
}

Future<int> Print(Stream<String> stream) async {
  int count = 0;

  await for (String item in stream) {
    print(item);

    count++;
  }

  return count;
}

import 'dart:async';

void main() async {

Stream<String> stream = await getStream(['A', 'B', 'C', 'D', 'E']);

int count = await Print(stream);

print(count);

}

const Duration duration = Duration(seconds: 1);

Stream<String> getStream(Iterable<String> itemIterable) async* {

for (String item in itemIterable) {

await Future.delayed(duration);

yield item;

}

Future<int> Print(Stream<String> stream) async {

int count = 0;

await for (String item in stream) {

print(item);

count++;

}

return count;

}

[DART] Stream 클래스 : length 속성 사용하기

■ Stream<T> 클래스의 length 속성을 사용하는 방법을 보여준다. ▶ 예제 코드 (DART)


import 'dart:async';

main() {
  Stream<int> stream = Stream.fromIterable([1, 2, 3, 4, 5]);

  stream.length.then((value) => print('항목 수 : $value'));
}

import 'dart:async';

main() {

Stream<int> stream = Stream.fromIterable([1, 2, 3, 4, 5]);

stream.length.then((value) => print('항목 수 : $value'));

}

[DART] Stream 클래스 : isEmpty 속성 사용하기

■ Stream<T> 클래스의 isEmpty 속성을 사용하는 방법을 보여준다. ▶ 예제 코드 (DART)


import 'dart:async';

main() {
  Stream<int> stream = Stream.fromIterable([1, 2, 3, 4, 5]);

  stream.isEmpty.then((value) => print('항목 존재 여부 : $value'));
}

import 'dart:async';

main() {

Stream<int> stream = Stream.fromIterable([1, 2, 3, 4, 5]);

stream.isEmpty.then((value) => print('항목 존재 여부 : $value'));

}

[DART] Stream 클래스 : last 속성 사용하기

■ Stream<T> 클래스의 last 속성을 사용하는 방법을 보여준다. ▶ 예제 코드 (DART)


import 'dart:async';

main() {
  Stream<int> stream = Stream.fromIterable([1, 2, 3, 4, 5]);

  stream.last.then((value) => print('두번째 항목 : $value'));
}

import 'dart:async';

main() {

Stream<int> stream = Stream.fromIterable([1, 2, 3, 4, 5]);

stream.last.then((value) => print('두번째 항목 : $value'));

}

[DART] Stream 클래스 사용하기

■ Stream<T> 클래스를 사용하는 방법을 보여준다. ▶ 예제 코드 (DART)


import 'dart:async';

Stream<int> countStream(int count) async* {
  for (int i = 1; i <= count; i++) {
    print('countStream : $i');
    yield i;
  }
}

Future<int> sumStream(Stream<int> stream) async {
  int summary = 0;
  await for (int value in stream) {
    print('sumStream : $value');
    summary += value;
  }
  return summary;
}

main() async {
  Stream<int> stream = countStream(10);
  int summary = await sumStream(stream);
  print(summary);
}

import 'dart:async';

Stream<int> countStream(int count) async* {

for (int i = 1; i <= count; i++) {

print('countStream : $i');

yield i;

}

Future<int> sumStream(Stream<int> stream) async {

int summary = 0;

await for (int value in stream) {

print('sumStream : $value');

summary += value;

}

return summary;

}

main() async {

Stream<int> stream = countStream(10);

int summary = await sumStream(stream);

print(summary);

}

[DART] Stream 클래스 : first 속성 사용하기

■ Stream<T> 클래스의 first 속성을 사용하는 방법을 보여준다. ▶ 예제 코드 (DART)


import 'dart:async';

main() {
  Stream<int> stream = Stream.fromIterable([1, 2, 3, 4, 5]);

  stream.first.then((value) => print('첫번째 항목 : $value'));
}

import 'dart:async';

main() {

Stream<int> stream = Stream.fromIterable([1, 2, 3, 4, 5]);

stream.first.then((value) => print('첫번째 항목 : $value'));

}