■ ChatOllama 클래스를 사용해 ollama에서 채팅 프로그램을 만드는 방법을 보여준다. (gemma2:2b 모델)
※ Ollama 설치와 gemma2:2b 모델이 다운로드 되어 있어야 한다.
▶ main.py
|
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 |
import asyncio from typing import List from typing import Tuple from typing import Dict from typing import Any from langchain_core.messages import HumanMessage from langchain_core.messages import AIMessage from langchain_core.prompts import ChatPromptTemplate from langchain_core.prompts import MessagesPlaceholder from langchain_community.chat_models import ChatOllama from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough def getMessageList(historyTupleList : List[Tuple[str, str]]) -> List[Dict[str, Any]]: messageList = [] for requestString, responseString in historyTupleList: messageList.append(HumanMessage(content = requestString )) messageList.append(AIMessage (content = responseString)) return messageList systemMessage = """당신은 한국의 유일한 만물 박사입니다. 한국어 사용자의 질문에 정중하고 단계별 상세히 반드시 한국어로 설명해줍니다.""" chatPromptTemplate = ChatPromptTemplate.from_messages( [ ("system", systemMessage), MessagesPlaceholder(variable_name = "history"), ("human", "{input}") ] ) chatOllama = ChatOllama(model = "gemma2:2b", streaming = True) runnableSequence = ( RunnablePassthrough.assign(history = lambda x : getMessageList(x["history"])) | chatPromptTemplate | chatOllama | StrOutputParser() ) def getHistoryTupleList(historyTupleList : List[Tuple[str, str]], requestString : str, responseString : str, maximumTurnCount : int = 5) -> List[Tuple[str, str]]: historyTupleList.append((requestString, responseString)) return historyTupleList[-maximumTurnCount:] async def getResponseString(asyncGenerator): responseString = "" async for chunkResponseString in asyncGenerator: responseString += chunkResponseString print(chunkResponseString, end = "", flush = True) print() return responseString async def main(): historyTupleList = [] turnCount = 0 print("채팅을 시작합니다. '종료'를 입력하면 대화를 마칩니다.") while True: requestString = input("사용자 : ") if requestString.lower() == "종료": print("채팅을 종료합니다.") break turnCount += 1 print(f"\nAI (Turn {turnCount}) : ", end="", flush = True) asyncGenerator = runnableSequence.astream({"input" : requestString, "history" : historyTupleList}) responseString = await getResponseString(asyncGenerator) historyTupleList = getHistoryTupleList(historyTupleList, requestString, responseString) print() if __name__ == "__main__": asyncio.run(main()) |
▶ requirements.txt
|
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 |
aiohappyeyeballs==2.4.0 aiohttp==3.10.5 aiosignal==1.3.1 annotated-types==0.7.0 anyio==4.4.0 attrs==24.2.0 certifi==2024.8.30 charset-normalizer==3.3.2 dataclasses-json==0.6.7 frozenlist==1.4.1 greenlet==3.0.3 h11==0.14.0 httpcore==1.0.5 httpx==0.27.2 idna==3.8 jsonpatch==1.33 jsonpointer==3.0.0 langchain==0.2.16 langchain-community==0.2.16 langchain-core==0.2.38 langchain-text-splitters==0.2.4 langsmith==0.1.111 marshmallow==3.22.0 multidict==6.0.5 mypy-extensions==1.0.0 numpy==1.26.4 orjson==3.10.7 packaging==24.1 pydantic==2.8.2 pydantic_core==2.20.1 PyYAML==6.0.2 requests==2.32.3 sniffio==1.3.1 SQLAlchemy==2.0.33 tenacity==8.5.0 typing-inspect==0.9.0 typing_extensions==4.12.2 urllib3==2.2.2 yarl==1.9.8 |
※ pip install langchain-community 명령을 실행했다.