LLM이 더 똑똑해 지게 RAG가 나오고, 이어서 내게 필요한 실제 일을 할 수 있는 AI Agent가 나온 다음 이제는 좀 더 복한한 문제를 처리할 수 있는 Agentic AI가 회자되고 있다.
AI Agent 는
추론하고 도구를 사용하여 답을 생성한다. 여기에 ReAct (Resoning 추론 + Act 실행) 패턴 또는 PlanningAct 패턴등이 사용된다. - Plan : 작업을 어떻게 수행할지 결정, 문제를 단계로 분해
- Act : 실행하기. 도구호출, API 요청, 행동 실행 - Tool Calling, MCP를 통한 도구의 확장이 가능하다
- Observer : 실행 결과과 또는 외부 환경의 피드백을 관찰하고 이해
Tool Calling을 Enterprise 급으로 확장하기 위해 MCP가 필수로 보인다. 두뇌인 LLM이 일을 시킬 수 있는 팔/다리 역할을 하는 MCP 도구가 필요하다. AI Agent가 사용하는 도구로써 MCP는 Pluggable한 확장성 및 모듈화 배포가 필요하다.
Agentic AI 는
- 목표지향성 : 복잡한 과정을 거쳐야 하는 목표를 달성 한다
- 자율성 : 자율적으로 움직인다
- 협업 : 에이전트끼리 협업한다.
Agentic AI는 단순한 질의응답형 LLM을 넘어, 목표 지향적이며 자율적이고 협업 가능한 지능형 시스템을 구축하기 위한 접근입니다.
"에이전트들이 협업하며, 자율적으로 해동 계획을 수립하고, 목표를 달성하는 시스템 전체 또는 프레임워크이다". 보통 AI Agent 라는 부르는 것은 "특정 목표를 수행하는 개별 인공지능 객체"를 의미한다.
Agentic AI 에서 중요한 것은 3가지 이다.
- 자율적 자기주도적 사고
- 멀티에이전트 협업 능력
- 목표 달성 중심 시스템(프레임워크)
AI Engineering 롤을 두개로 나눈다면 AI Solution Engineer와 AI Data Engineer로 나눠 보았을 때, AI Solution Engineer가 Agentic AI 시스템을 구축하는 것이고, AI Data Engineer가 전통적인 ML 및 비즈니스 핵심 알고리즘을 구현하는 부분이라 본다.
그렇다면 AI Solution Engineering의 파트를 구체화 한다면 다음의 7 Layers로 나누어 볼 수 있겠다.
Where humans interact with the agent - through chatbots, dashboards, or voice assistants.
It’s the UI layer of agent interaction.
2. Discovery Layer
This is how the agent finds and injects relevant knowledge
- using RAG, vector databases, and memory recall techniques.
3. Agent Composition Layer
Defines the internal structure of your agent.
Combines modular sub-agents into complex, goal-driven workflows.
4. Reasoning & Planning Layer
The thinking brain of the agent. It plans actions, reflects, sets priorities,
and uses strategies like CoT or ReAct.
5. Tool & API Layer
Connects reasoning to real-world execution.
It handles file systems, APIs, shell commands, and function calls.
6. Memory & Feedback Layer
Where the agent learns and adapts.
Stores past interactions, builds self-awareness, and uses memory to improve.
7. Infrastructure Layer
The backend engine scales agents, manages models, deploys pipelines,
and secures the whole system.
인프라스트럭처의 LLMOps 에서 부터 A2A 또는 LangGraph와 같은 Framework을 활용하여 Multi Agent를 구현하고, UI를 제공하는 모든 부분을 Agentic AI Framework 또는 System 이라 말할 수 있다. 이부분을 AI Solution Engineer가 해야할 영역으로 본다.
Agentic AI Solution 은 "자기 주도적이면 자율적으로 멀티 에이전트가 협업하여 목표를 달성하는" 것을 목적으로 한다. 여기서 해당 솔루션을 비즈니스에 적용하기 위하여 최근 이야기 되고 있는 Context Engineering 관점의 접근도 필요하다.
Context Engineering 이란?
“Context Engineering”은 특히 대규모 언어 모델(LLM) 시대에 들어서면서 주목받는 개념으로, AI가 정확하게 이해하고 응답할 수 있도록 입력(프롬프트)의 맥락을 설계·조정하는 기술 또는 실천적 전략을 의미한다. 즉 AI Solution Engineer는 Agentic AI System 구축시 (AI Data Engineer가) Context Engineering을 유연하게 적용할 수 있는 시스템을 지향해야 한다.
Context Engineering = Prompting + Memory + Retrieval + Role Control + History 관리의 종합 아키텍처 설계 기술
프롬프트의 시대는 저물고 있습니다.
이제는 더욱 중요한 '컨텍스트'를 설계해야 할 때입니다.
해외 AI 업계에서는 최근 '컨텍스트 엔지니어링'이라는 개념이 주목받고 있습니다.
단순히 질문(프롬프트)을 잘 던지는 것만으로는 부족하다는 뜻입니다.
중요한 건, AI가 제대로 이해하고 일할 수 있도록 '맥락'을 설계하는 일입니다.
AI는 신입사원과 같습니다. 업무 지시만으로는 부족합니다.
조직의 문화, 일하는 방식, 기대 기준 등을 함께 알려줘야 제대로 일합니다.
'맥락'은 단순한 배경정보가 아닙니다.
우리가 평소 사용하는 리포트 포맷, 문서 스타일, 브랜드의 말투 등이 곧 조직의 언어이자 컨텍스트이죠.
결국 AI에게 일을 시키려면, 우리가 어떻게 일하는지를 먼저 정의해야 합니다.
AI를 잘 쓴다는 건, 우리 조직의 이상적인 일처리 기준을 정의하고, 그것을 AI에게 정확히 전달하는 것입니다.
이제는 ‘프롬프트’가 아닌 ‘컨텍스트’를 설계하는 시대입니다.
LLM을 도구 이상으로 활용하려면, 그 배경에 숨어 있는 이 섬세한 맥락 설계의 기술과 감각을 이해하여 사용한다.
Usage: uv [OPTIONS] <COMMAND>
Commands:
run Run a command or script
init Create a new project
add Add dependencies to the project
remove Remove dependencies from the project
sync Update the project's environment
lock Update the project's lockfile
export Export the project's lockfile to an alternate format
tree Display the project's dependency tree
tool Run and install commands provided by Python packages
python Manage Python versions and installations
pip Manage Python packages with a pip-compatible interface
venv Create a virtual environment
build Build Python packages into source distributions and wheels
publish Upload distributions to an index
cache Manage uv's cache
self Manage the uv executable
version Display uv's version
help Display documentation for a command
Cache options:
-n, --no-cache Avoid reading from or writing to the cache, instead using a temporary directory for the
duration of the operation [env: UV_NO_CACHE=]
--cache-dir <CACHE_DIR> Path to the cache directory [env: UV_CACHE_DIR=]
Python options:
--python-preference <PYTHON_PREFERENCE> Whether to prefer uv-managed or system Python installations [env:
UV_PYTHON_PREFERENCE=] [possible values: only-managed, managed, system,
only-system]
--no-python-downloads Disable automatic downloads of Python. [env: "UV_PYTHON_DOWNLOADS=never"]
Global options:
-q, --quiet
Do not print any output
-v, --verbose...
Use verbose output
--color <COLOR_CHOICE>
Control the use of color in output [possible values: auto, always, never]
--native-tls
Whether to load TLS certificates from the platform's native certificate store [env: UV_NATIVE_TLS=]
--offline
Disable network access [env: UV_OFFLINE=]
--allow-insecure-host <ALLOW_INSECURE_HOST>
Allow insecure connections to a host [env: UV_INSECURE_HOST=]
--no-progress
Hide all progress outputs [env: UV_NO_PROGRESS=]
--directory <DIRECTORY>
Change to the given directory prior to running the command
--project <PROJECT>
Run the command within the given project directory
--config-file <CONFIG_FILE>
The path to a `uv.toml` file to use for configuration [env: UV_CONFIG_FILE=]
--no-config
Avoid discovering configuration files (`pyproject.toml`, `uv.toml`) [env: UV_NO_CONFIG=]
-h, --help
Display the concise help for this command
-V, --version
Display the uv version
> npx create-nx-workspace
✔ Where would you like to create your workspace? · aip
✔ Which stack do you want to use? · none
✔ Would you like to use Prettier for code formatting? · Yes
✔ Which CI provider would you like to use? · azure
NX Creating your v20.5.0 workspace.
✔ Installing dependencies with npm
[2] NX 전용 파이썬 플러그인을 설치한다. @nxlv/python 은 build, publish 등 uv 패키지 메니져 사용시, poetry 의 장점을 보완한다.
NX의 모노레포 아키텍쳐에서는 Shell Library Pattern 방식을 지향한다. 이는 다양한 애플리케이션(서비스)안에서 사용하는 패키지(라이브러리)를 분리 관리하는 방식으로 확장성과 유지보수성을 높혀준다.
프로젝트에서 애플리케이션의 모든 페이지와 컴포넌트를 패키지에 담고, 애플리케이션은 이들의 조합하고 설정하는 역할만 수행한다. 이때 패키지는 Nexus Registry에 배포하여 버전관리를 하게되면 유지보수성이 좋아진다.
실 프로젝트에서 Frontend 파트를 보면 Micro Applications들은 각가의 Micro Frontend 로 구성하고, Portal 이 라는 곳에서 Module Federation을 통해 각 micro frontend를 통합하여 표현한다. Micro Frontend에는 비즈니스 로직이 없고, 애플리케이션 구성 환경설정만 존재한다. libs 폴더안에 있는 것이 패키지로 각 패키지는 NPM Registry에 배포되어 버전 관리를 하고, 애플리케이션은 libs에 있는 패키지를 통해 화면 및 비지니스 로직을 구현한다.
Peter real project folder structure
NX의 Shell Library Pattern도 유사하게 Feature Shell 이라는 Integrated Application을 제공한다.
Figure 1
A domain-wise application is the composition of every application that has the same routes, behavior, and functionalities. In the example in Figure 1, the Booking Application is the union of the Booking Web Application, the Booking Desktop Application, and the Booking Mobile Application.
UV 기반의 python 애플리케이션 또는 패키지를 생성(Generate)하기 위한 명령어.
npx nx g @nxlv/python:uv-project src --projectType=library --directory=packages/embedding --packageName=aip-embedding --publishable
- src 폴더 밑으로 템플릿 기반 소스 생성
- pyproject.toml 파일 생성
- project.json NX 환경파일 생성
- 그외 Unit test 용 tests 폴더 생성
▶ npx nx g @nxlv/python:uv-project src --projectType=library --directory=packages/embedding --packageName=aip-embedding --publishable
NX Generating @nxlv/python:uv-project
CREATE packages/embedding/project.json
CREATE packages/embedding/README.md
CREATE packages/embedding/.python-version
CREATE packages/embedding/src/__init__.py
CREATE packages/embedding/src/hello.py
CREATE packages/embedding/pyproject.toml
CREATE packages/embedding/tests/__init__.py
CREATE packages/embedding/tests/conftest.py
CREATE packages/embedding/tests/test_hello.py
=========
▶ npx nx g @nxlv/python:uv-project --help
NX generate @nxlv/python:uv-project [name] [options,...]
From: @nxlv/python (v20.7.0)
Name: uv-project
Options:
--name [string]
--projectType Project type [string] [choices: "application",
"library"] [default: "application"]
--buildBundleLocalDependencies Bundle local dependencies [boolean] [default: true]
--buildLockedVersions Use locked versions for build dependencies [boolean] [default: true]
--codeCoverage Generate code coverage report [boolean] [default: true]
--codeCoverageHtmlReport Generate html report for code coverage [boolean] [default: true]
--codeCoverageThreshold Code coverage threshold [number]
--codeCoverageXmlReport Generate Xml report for code coverage [boolean] [default: true]
--description Project short description [string]
--devDependenciesProject This approach installs all the missing dev [string]
dependencies in a separate project
(optional)
--directory A directory where the project is placed [string]
--linter Project linter [string] [choices: "flake8", "ruff",
"none"] [default: "ruff"]
--moduleName Python module name [string]
--packageName Python package name [string]
--projectNameAndRootFormat Whether to generate the project name and [string] [choices: "as-provided",
root directory as provided (`as-provided`) "derived"] [default: "as-provided"]
or generate them composing their values and
taking the configured layout into account
(`derived`).
--publishable Project is publishable [boolean]
--pyenvPythonVersion Pyenv .python-version content (default to [string]
current python version)
--pyprojectPythonDependency Pyproject python dependency version range [string] [default: ">=3.9,<4"]
--rootPyprojectDependencyGroup If a shared pyproject.toml is used, which [string] [default: "main"]
dependency group does this new project
should belong to
--tags, -t Add tags to the project (used for linting) [string]
--templateDir Custom template directory, this will [string]
override the default template, if not
provided the default template will be used
--unitTestHtmlReport Generate html report for unit tests [boolean] [default: true]
--unitTestJUnitReport Generate junit report for unit tests [boolean] [default: true]
--unitTestRunner Project unit test runner [string] [choices: "pytest", "none"]
[default: "pytest"]
- AI agents typically use language models as part of their software stack to interpret messages, perform reasoning, and execute actions.
- Each agent is a self-contained unit that can be developed, tested, and deployed independently.
[2] Multi Agent 특징
- Run within the same process or on the same machine
- Operate across different machines or organizational boundaries
- Be implemented in diverse programming languages and make use of different AI models or instructions
- Work together towards a shared goal, coordinating their actions through messaging
[3] Agent 실행 환경
the framework provides a runtime environment, which facilitates communication between agents, manages their identities and lifecycles, and enforce security and privacy boundaries.
Standalone Agent Runtime
- single process application
- all agents are implemented in the same language
- running in the same process.
Agent는 runtime 동안 메세지를 통해 통신을 하고, 런타임은 Agent의 LifeCycle을 관리한다.
Distrubuted Agent Runtime
- multi process applications
- Agents are implemented in different programming languages
- running on different machines
분산환경은 host servicer와 multiple workers를 갖는다.
- host servicer는 agent 사이의 통신과 연결상태를 관리한다.
- worker는 agent를 구동하고 host servicer와 gateway를 통해 통신한다.
[4] 애플리케이션 스택
AutoGen core는 다양한 multi-agent application 개발할 때 사용된다.
[5] Agent 식별 방법 & 라이프 사이클 관리
Agent 런타임은 에이젼트 identities와 lifecyles을 관리한다.
- Agent ID = Agent Type + Agent Key(instance)
런타임에서 Agent가 없으면 생성하고, 메세지로 agent type & key를 전달한다.
[6] Topic & Subscription
message를 broadcast할 때 방법을 설명한다.
- Topic : pubishing messages with Topic Type, Topic Source
- Subscription : topic 과 agentID와 맵팽한다. 런타임 환경에 맵핑을 만들고 삭제할 수 있다.
Type-based subscription
Topic Type -> Agent Type 전파
- Single Tenant & Single Topic
- Single Tenant & Multi Topics
- Multi Tenant
in single tenant, the topic source is "default". for multi tenant, it become data-dependent.
with get_openai_callback() as cb:는 Python의 컨텍스트 관리자(context manager)를 사용하여 get_openai_callback 함수가 반환하는 객체(cb)를 생성하고, 그 객체를 사용하는 블록을 정의하는 구문입니다. 이 구문을 이해하기 위해서는 Python의 컨텍스트 관리자가 어떻게 작동하는지와 get_openai_callback이 어떤 역할을 하는지를 아는 것이 중요합니다.
1. 컨텍스트 관리자 (Context Manager)
컨텍스트 관리자는 with 블록의 시작과 종료 시 특정 코드를 자동으로 실행하게 해줍니다. 일반적으로, 컨텍스트 관리자는 자원(resource)을 할당하고 해제하는 작업에 사용됩니다. 예를 들어, 파일을 열고 작업을 한 후 자동으로 파일을 닫는 데 사용할 수 있습니다.
•__enter__(): with 블록이 시작될 때 호출됩니다. 이 메서드는 일반적으로 어떤 자원을 할당하거나 초기화합니다.
•__exit__(): with 블록이 끝날 때 호출됩니다. 이 메서드는 자원을 해제하거나, 예외가 발생했을 때 이를 처리합니다.
2. get_openai_callback의 역할
get_openai_callback은 OpenAI API 호출과 관련된 메트릭을 수집하는 콜백 객체를 반환합니다. 이 콜백 객체는 컨텍스트 관리자에서 사용될 때 API 호출 동안의 토큰 사용량, 비용 등을 추적합니다.
3. with get_openai_callback() as cb:의 의미
•get_openai_callback()은 컨텍스트 관리자 역할을 하는 객체를 반환합니다.
•with 블록이 시작되면, cb 변수에 이 객체가 할당됩니다.
•with 블록 내에서 OpenAI API 호출이 이루어지면, cb 객체는 API 호출 관련 데이터를 수집합니다.
•with 블록이 종료되면, cb 객체는 수집한 데이터를 자동으로 정리하고, 필요한 경우 자원을 해제합니다.
예시 코드 분석
from langchain.llms import OpenAI
from langchain.callbacks import get_openai_callback
llm = OpenAI(model="text-davinci-003")
with get_openai_callback() as cb:
response = llm("What is the capital of France?")
print(response)
print(f"Total Tokens: {cb.total_tokens}")
print(f"Total Cost: {cb.total_cost}")
•get_openai_callback(): 콜백 객체를 생성하여 반환합니다.
•with ... as cb:: cb 변수에 콜백 객체를 할당하고, with 블록 내에서 이 객체를 사용합니다.
•cb.total_tokens, cb.total_cost: with 블록이 끝난 후, API 호출 동안 사용된 총 토큰 수와 총 비용을 출력합니다.
이 구문을 사용함으로써 개발자는 OpenAI API 호출의 성능을 모니터링하고 리소스 사용량을 효율적으로 관리할 수 있습니다.
get_openai_callback
소스: langchain_community/callbacks/manager.py
from contextlib import contextmanager
@contextmanager
def get_openai_callback() -> Generator[OpenAICallbackHandler, None, None]:
"""Get the OpenAI callback handler in a context manager.
which conveniently exposes token and cost information.
Returns:
OpenAICallbackHandler: The OpenAI callback handler.
Example:
>>> with get_openai_callback() as cb:
... # Use the OpenAI callback handler
"""
cb = OpenAICallbackHandler()
openai_callback_var.set(cb)
yield cb
openai_callback_var.set(None)
Pydantic은 Python에서 데이터 유효성 검사 및 설정 관리를 위한 라이브러리입니다. 주로 FastAPI와 같은 웹 프레임워크와 함께 사용되며, 데이터를 구조화하고 검증하는 데 유용합니다. BaseModel은 Pydantic의 핵심 클래스 중 하나로, 데이터 모델을 정의하는 데 사용됩니다.
Pydantic의 주요 기능
1.유효성 검사 및 변환: 필드에 대해 타입을 지정하면, 입력 데이터가 자동으로 그 타입으로 변환되며, 유효성 검사가 수행됩니다.
2.자동 완성 및 타입 힌팅 지원: IDE의 자동 완성과 타입 힌팅을 통해 개발 생산성을 높입니다.
3.데이터 직렬화 및 역직렬화: 모델 인스턴스를 JSON으로 직렬화하거나 JSON으로부터 역직렬화할 수 있습니다.
4.데이터 검증 오류 관리: 잘못된 데이터를 입력하면, Pydantic이 자동으로 유효성 검사 오류를 생성합니다.
BaseModel 사용 예시
다음은 Pydantic의 BaseModel을 사용하여 간단한 사용자 데이터를 관리하는 예제입니다.
from pydantic import BaseModel, EmailStr, Field
from typing import Optional
class User(BaseModel):
id: int
name: str = Field(..., min_length=3, max_length=50)
email: EmailStr
age: Optional[int] = Field(None, ge=18)
is_active: bool = True
# Example usage
user_data = {
"id": 1,
"name": "John Doe",
"email": "johndoe@example.com",
"age": 25,
}
user = User(**user_data)
print(user)
print(user.dict())
코드 설명
1.필드 정의:
•id: 정수형 필드.
•name: 길이가 3에서 50 사이인 문자열 필드.
•email: 이메일 형식의 문자열을 요구하는 필드. EmailStr 타입은 이메일 주소가 올바른 형식인지 검증합니다.
•age: 선택적 필드로, 값이 주어지면 18 이상이어야 합니다.
•is_active: 기본값이 True인 불리언 필드.
2.필드 유효성 검사:
•Field를 사용하여 각 필드에 대한 추가적인 제약 조건을 지정합니다.
3.데이터 생성 및 출력:
•user_data 딕셔너리를 통해 User 객체를 생성합니다. 생성된 객체를 출력하거나, .dict() 메서드를 사용하여 객체를 딕셔너리 형태로 변환할 수 있습니다.
이와 같이 Pydantic을 사용하면 데이터 모델을 간단하고 명확하게 정의할 수 있으며, 자동으로 타입 변환과 유효성 검사를 수행할 수 있습니다. 이를 통해 데이터 처리의 신뢰성과 안정성을 높일 수 있습니다.
BaseModel은 자동으로 __init__ 을 실행
Pydantic의 BaseModel을 사용하면 클래스 수준에서 필드를 정의할 수 있으며, 이러한 필드는 마치 __init__ 메서드에서 self.name과 같이 인스턴스 변수로 설정된 것처럼 동작합니다. Pydantic은 이러한 필드를 기반으로 자동으로 __init__ 메서드를 생성하고, 필드에 대한 타입 검사를 수행합니다.
이 방식은 일반적인 Python 클래스에서의 인스턴스 변수 설정과는 약간 다릅니다. 일반 Python 클래스에서는 인스턴스 변수를 __init__ 메서드 내에서 self를 통해 설정해야 하지만, Pydantic의 BaseModel을 사용하면 클래스 정의 시 필드의 타입과 기본값을 지정하여 더 간결하고 명확하게 모델을 정의할 수 있습니다.
예시 비교
일반 Python 클래스
class User:
def __init__(self, id: int, name: str, email: str, age: int, is_active: bool = True):
self.id = id
self.name = name
self.email = email
self.age = age
self.is_active = is_active
Pydantic BaseModel
from pydantic import BaseModel, EmailStr, Field
from typing import Optional
class User(BaseModel):
id: int
name: str = Field(..., min_length=3, max_length=50)
email: EmailStr
age: Optional[int] = Field(None, ge=18)
is_active: bool = True
차이점 설명
•일반 클래스에서는 __init__ 메서드 내에서 self를 사용하여 인스턴스 변수를 직접 설정합니다.
•Pydantic BaseModel에서는 클래스 정의 시 필드를 직접 설정하고, Pydantic이 자동으로 __init__ 메서드를 생성하여 필드 초기화, 타입 검사, 유효성 검사를 수행합니다.
이렇게 Pydantic의 BaseModel을 사용하면 코드가 더 간결해지며, 데이터 유효성 검사가 자동으로 처리되므로 안전하고 유지보수하기 쉬운 코드를 작성할 수 있습니다.
- Model 생성시 configurable_fileds 사설정 -> Model 인스턴스 with_config 런타임 설정
class RunnableSerializable(Serializable, Runnable[Input, Output]):
"""Runnable that can be serialized to JSON."""
def configurable_fields(
self, **kwargs: AnyConfigurableField
) -> RunnableSerializable[Input, Output]:
"""Configure particular Runnable fields at runtime.
Args:
**kwargs: A dictionary of ConfigurableField instances to configure.
Returns:
A new Runnable with the fields configured.
.. code-block:: python
from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI
model = ChatOpenAI(max_tokens=20).configurable_fields(
max_tokens=ConfigurableField(
id="output_token_number",
name="Max tokens in the output",
description="The maximum number of tokens in the output",
)
)
# max_tokens = 20
print(
"max_tokens_20: ",
model.invoke("tell me something about chess").content
)
# max_tokens = 200
print("max_tokens_200: ", model.with_config(
configurable={"output_token_number": 200}
).invoke("tell me something about chess").content
)
"""
from langchain_core.runnables.configurable import RunnableConfigurableFields
for key in kwargs:
if key not in self.__fields__:
raise ValueError(
f"Configuration key {key} not found in {self}: "
f"available keys are {self.__fields__.keys()}"
)
return RunnableConfigurableFields(default=self, fields=kwargs)
class Runnable(Generic[Input, Output], ABC):
... 중략 ...
def with_config(
self,
config: Optional[RunnableConfig] = None,
# Sadly Unpack is not well-supported by mypy so this will have to be untyped
**kwargs: Any,
) -> Runnable[Input, Output]:
"""
Bind config to a Runnable, returning a new Runnable.
Args:
config: The config to bind to the Runnable.
kwargs: Additional keyword arguments to pass to the Runnable.
Returns:
A new Runnable with the config bound.
"""
return RunnableBinding(
bound=self,
config=cast(
RunnableConfig,
{**(config or {}), **kwargs},
), # type: ignore[misc]
kwargs={},
)
실제 예제
- Model 생성시 ConfigurableField 통해 설정
- model 인스턴스 사용시 with_config 통해 설정
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI
model = ChatOpenAI(temperature=0).configurable_fields(
temperature=ConfigurableField(
id="llm_temperature",
name="LLM Temperature",
description="The temperature of the LLM",
)
)
//--- case-1
model.invoke("pick a random number")
// 결과
AIMessage(content='17', response_metadata={'token_usage': {'completion_tokens': 1, 'prompt_tokens': 11, 'total_tokens': 12}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None}, id='run-ba26a0da-0a69-4533-ab7f-21178a73d303-0')
//--- case-2
model.with_config(configurable={"llm_temperature": 0.9}).invoke("pick a random number")
// 결과
AIMessage(content='12', response_metadata={'token_usage': {'completion_tokens': 1, 'prompt_tokens': 11, 'total_tokens': 12}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None}, id='run-ba8422ad-be77-4cb1-ac45-ad0aae74e3d9-0')
//--- case-3
prompt = PromptTemplate.from_template("Pick a random number above {x}")
chain = prompt | model
chain.invoke({"x": 0})
// 결과
AIMessage(content='27', response_metadata={'token_usage': {'completion_tokens': 1, 'prompt_tokens': 14, 'total_tokens': 15}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None}, id='run-ecd4cadd-1b72-4f92-b9a0-15e08091f537-0')
//--- case-4
chain.with_config(configurable={"llm_temperature": 0.9}).invoke({"x": 0})
// 결과
AIMessage(content='35', response_metadata={'token_usage': {'completion_tokens': 1, 'prompt_tokens': 14, 'total_tokens': 15}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None}, id='run-a916602b-3460-46d3-a4a8-7c926ec747c0-0')
configurable_alternatives 메서드 이해
Chain에 연결할 Runnable을 교체할 수 있게 한다.
class RunnableSerializable(Serializable, Runnable[Input, Output]):
... 중략 ...
def configurable_alternatives(
self,
which: ConfigurableField,
*,
default_key: str = "default",
prefix_keys: bool = False,
**kwargs: Union[Runnable[Input, Output], Callable[[], Runnable[Input, Output]]],
) -> RunnableSerializable[Input, Output]:
"""Configure alternatives for Runnables that can be set at runtime.
Args:
which: The ConfigurableField instance that will be used to select the
alternative.
default_key: The default key to use if no alternative is selected.
Defaults to "default".
prefix_keys: Whether to prefix the keys with the ConfigurableField id.
Defaults to False.
**kwargs: A dictionary of keys to Runnable instances or callables that
return Runnable instances.
Returns:
A new Runnable with the alternatives configured.
.. code-block:: python
from langchain_anthropic import ChatAnthropic
from langchain_core.runnables.utils import ConfigurableField
from langchain_openai import ChatOpenAI
model = ChatAnthropic(
model_name="claude-3-sonnet-20240229"
).configurable_alternatives(
ConfigurableField(id="llm"),
default_key="anthropic",
openai=ChatOpenAI()
)
# uses the default model ChatAnthropic
print(model.invoke("which organization created you?").content)
# uses ChatOpenAI
print(
model.with_config(
configurable={"llm": "openai"}
).invoke("which organization created you?").content
)
"""
from langchain_core.runnables.configurable import (
RunnableConfigurableAlternatives,
)
return RunnableConfigurableAlternatives(
which=which,
default=self,
alternatives=kwargs,
default_key=default_key,
prefix_keys=prefix_keys,
)
실제 예제
- Anthropic 모델 생성시, OpenAI 모델을 alternative로 설정한다.
- 상황에 따라 OpenAI 모델을 사용한다.
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI
llm = ChatAnthropic(
model="claude-3-haiku-20240307", temperature=0
).configurable_alternatives(
# This gives this field an id
# When configuring the end runnable, we can then use this id to configure this field
ConfigurableField(id="llm"),
# This sets a default_key.
# If we specify this key, the default LLM (ChatAnthropic initialized above) will be used
default_key="anthropic",
# This adds a new option, with name `openai` that is equal to `ChatOpenAI()`
openai=ChatOpenAI(),
# This adds a new option, with name `gpt4` that is equal to `ChatOpenAI(model="gpt-4")`
gpt4=ChatOpenAI(model="gpt-4"),
# You can add more configuration options here
)
prompt = PromptTemplate.from_template("Tell me a joke about {topic}")
chain = prompt | llm
//--- case-1
# By default it will call Anthropic
chain.invoke({"topic": "bears"})
// 결과
AIMessage(content="Here's a bear joke for you:\n\nWhy don't bears wear socks? \nBecause they have bear feet!\n\nHow's that? I tried to come up with a simple, silly pun-based joke about bears. Puns and wordplay are a common way to create humorous bear jokes. Let me know if you'd like to hear another one!", response_metadata={'id': 'msg_018edUHh5fUbWdiimhrC3dZD', 'model': 'claude-3-haiku-20240307', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'input_tokens': 13, 'output_tokens': 80}}, id='run-775bc58c-28d7-4e6b-a268-48fa6661f02f-0')
//--- case-2
# We can use `.with_config(configurable={"llm": "openai"})` to specify an llm to use
chain.with_config(configurable={"llm": "openai"}).invoke({"topic": "bears"})
// 결과
AIMessage(content="Why don't bears like fast food?\n\nBecause they can't catch it!", response_metadata={'token_usage': {'completion_tokens': 15, 'prompt_tokens': 13, 'total_tokens': 28}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None}, id='run-7bdaa992-19c9-4f0d-9a0c-1f326bc992d4-0')
//--- case-3
# If we use the `default_key` then it uses the default
chain.with_config(configurable={"llm": "anthropic"}).invoke({"topic": "bears"})
// 결과
AIMessage(content="Here's a bear joke for you:\n\nWhy don't bears wear socks? \nBecause they have bear feet!\n\nHow's that? I tried to come up with a simple, silly pun-based joke about bears. Puns and wordplay are a common way to create humorous bear jokes. Let me know if you'd like to hear another one!", response_metadata={'id': 'msg_01BZvbmnEPGBtcxRWETCHkct', 'model': 'claude-3-haiku-20240307', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'input_tokens': 13, 'output_tokens': 80}}, id='run-59b6ee44-a1cd-41b8-a026-28ee67cdd718-0')
class Runnable(Generic[Input, Output], ABC):
... 중략 ...
def bind(self, **kwargs: Any) -> Runnable[Input, Output]:
"""
Bind arguments to a Runnable, returning a new Runnable.
Useful when a Runnable in a chain requires an argument that is not
in the output of the previous Runnable or included in the user input.
Args:
kwargs: The arguments to bind to the Runnable.
Returns:
A new Runnable with the arguments bound.
Example:
.. code-block:: python
from langchain_community.chat_models import ChatOllama
from langchain_core.output_parsers import StrOutputParser
llm = ChatOllama(model='llama2')
# Without bind.
chain = (
llm
| StrOutputParser()
)
chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two three four five.'
# With bind.
chain = (
llm.bind(stop=["three"])
| StrOutputParser()
)
chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two'
"""
return RunnableBinding(bound=self, kwargs=kwargs, config={})
- bind: Bind kwargs to pass to the underlying Runnable when running it. - with_config: Bind config to pass to the underlying Runnable when running it. - with_listeners: Bind lifecycle listeners to the underlying Runnable. - with_types: Override the input and output types of the underlying Runnable. - with_retry: Bind a retry policy to the underlying Runnable. - with_fallbacks: Bind a fallback policy to the underlying Runnable.
class RunnableBinding(RunnableBindingBase[Input, Output]):
"""Wrap a Runnable with additional functionality.
A RunnableBinding can be thought of as a "runnable decorator" that
preserves the essential features of Runnable; i.e., batching, streaming,
and async support, while adding additional functionality.
Any class that inherits from Runnable can be bound to a `RunnableBinding`.
Runnables expose a standard set of methods for creating `RunnableBindings`
or sub-classes of `RunnableBindings` (e.g., `RunnableRetry`,
`RunnableWithFallbacks`) that add additional functionality.
These methods include:
- `bind`: Bind kwargs to pass to the underlying Runnable when running it.
- `with_config`: Bind config to pass to the underlying Runnable when running it.
- `with_listeners`: Bind lifecycle listeners to the underlying Runnable.
- `with_types`: Override the input and output types of the underlying Runnable.
- `with_retry`: Bind a retry policy to the underlying Runnable.
- `with_fallbacks`: Bind a fallback policy to the underlying Runnable.
Example:
`bind`: Bind kwargs to pass to the underlying Runnable when running it.
.. code-block:: python
# Create a Runnable binding that invokes the ChatModel with the
# additional kwarg `stop=['-']` when running it.
from langchain_community.chat_models import ChatOpenAI
model = ChatOpenAI()
model.invoke('Say "Parrot-MAGIC"', stop=['-']) # Should return `Parrot`
# Using it the easy way via `bind` method which returns a new
# RunnableBinding
runnable_binding = model.bind(stop=['-'])
runnable_binding.invoke('Say "Parrot-MAGIC"') # Should return `Parrot`
Can also be done by instantiating a RunnableBinding directly (not recommended):
.. code-block:: python
from langchain_core.runnables import RunnableBinding
runnable_binding = RunnableBinding(
bound=model,
kwargs={'stop': ['-']} # <-- Note the additional kwargs
)
runnable_binding.invoke('Say "Parrot-MAGIC"') # Should return `Parrot`
"""
RAG 구성시에 마지막 chain을 만들 때 retriever를 설정할 때 RunnableParallel을 사용한다.
from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
vectorstore = FAISS.from_texts(
["harrison worked at kensho"], embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()
template = """Answer the question based only on the following context:
{context}
Question: {question}
"""
# The prompt expects input with keys for "context" and "question"
prompt = ChatPromptTemplate.from_template(template)
model = ChatOpenAI()
retrieval_chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt
| model
| StrOutputParser()
)
// 아래 3가지의 사용방식은 동일한다.
// ref: https://python.langchain.com/v0.2/docs/how_to/parallel/
// {"context": retriever, "question": RunnablePassthrough()}
// RunnableParallel({"context": retriever, "question": RunnablePassthrough()})
// RunnableParallel(context=retriever, question=runnablePassthrough())
retrieval_chain.invoke("where did harrison work?")
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel
from langchain_openai import ChatOpenAI
model = ChatOpenAI()
joke_chain = ChatPromptTemplate.from_template("tell me a joke about {topic}") | model
poem_chain = (
ChatPromptTemplate.from_template("write a 2-line poem about {topic}") | model
)
map_chain = RunnableParallel(joke=joke_chain, poem=poem_chain)
map_chain.invoke({"topic": "bear"})
// 결과
{'joke': AIMessage(content="Why don't bears like fast food? Because they can't catch it!", response_metadata={'token_usage': {'completion_tokens': 15, 'prompt_tokens': 13, 'total_tokens': 28}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_d9767fc5b9', 'finish_reason': 'stop', 'logprobs': None}, id='run-fe024170-c251-4b7a-bfd4-64a3737c67f2-0'),
'poem': AIMessage(content='In the quiet of the forest, the bear roams free\nMajestic and wild, a sight to see.', response_metadata={'token_usage': {'completion_tokens': 24, 'prompt_tokens': 15, 'total_tokens': 39}, 'model_name': 'gpt-3.5-turbo', 'system_fingerprint': 'fp_c2295e73ad', 'finish_reason': 'stop', 'logprobs': None}, id='run-2707913e-a743-4101-b6ec-840df4568a76-0')}
class RunnablePassthrough(RunnableSerializable[Other, Other]):
... 중략 ...
@classmethod
def assign(
cls,
**kwargs: Union[
Runnable[Dict[str, Any], Any],
Callable[[Dict[str, Any]], Any],
Mapping[
str,
Union[Runnable[Dict[str, Any], Any], Callable[[Dict[str, Any]], Any]],
],
],
) -> "RunnableAssign":
"""Merge the Dict input with the output produced by the mapping argument.
Args:
**kwargs: Runnable, Callable or a Mapping from keys to Runnables
or Callables.
Returns:
A Runnable that merges the Dict input with the output produced by the
mapping argument.
"""
return RunnableAssign(RunnableParallel(kwargs))
from operator import itemgetter
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import Runnable, RunnablePassthrough, chain
contextualize_instructions = """Convert the latest user question into a standalone question given the chat history. Don't answer the question, return the question and nothing else (no descriptive text)."""
contextualize_prompt = ChatPromptTemplate.from_messages(
[
("system", contextualize_instructions),
("placeholder", "{chat_history}"),
("human", "{question}"),
]
)
contextualize_question = contextualize_prompt | llm | StrOutputParser()
qa_instructions = (
"""Answer the user question given the following context:\n\n{context}."""
)
qa_prompt = ChatPromptTemplate.from_messages(
[("system", qa_instructions), ("human", "{question}")]
)
@chain
def contextualize_if_needed(input_: dict) -> Runnable:
if input_.get("chat_history"):
# NOTE: This is returning another Runnable, not an actual output.
return contextualize_question
else:
return RunnablePassthrough() | itemgetter("question")
@chain
def fake_retriever(input_: dict) -> str:
return "egypt's population in 2024 is about 111 million"
full_chain = (
RunnablePassthrough.assign(question=contextualize_if_needed).assign(
context=fake_retriever
)
| qa_prompt
| llm
| StrOutputParser()
)
full_chain.invoke(
{
"question": "what about egypt",
"chat_history": [
("human", "what's the population of indonesia"),
("ai", "about 276 million"),
],
}
)
// 결과
"According to the context provided, Egypt's population in 2024 is estimated to be about 111 million."
Custom Function을 Runnable로 사용하기 - @chain == RunnableLambda
chaining 시에 custom function은 RunnableLambda 를 사용하거나, @chain 데코레이터를 사용한다.
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import chain
prompt1 = ChatPromptTemplate.from_template("Tell me a joke about {topic}")
prompt2 = ChatPromptTemplate.from_template("What is the subject of this joke: {joke}")
@chain
def custom_chain(text):
prompt_val1 = prompt1.invoke({"topic": text})
output1 = ChatOpenAI().invoke(prompt_val1)
parsed_output1 = StrOutputParser().invoke(output1)
chain2 = prompt2 | ChatOpenAI() | StrOutputParser()
return chain2.invoke({"joke": parsed_output1})
custom_chain.invoke("bears")
// 결과
'The subject of the joke is the bear and his girlfriend.'
def chain(
func: Union[
Callable[[Input], Output],
Callable[[Input], Iterator[Output]],
Callable[[Input], Coroutine[Any, Any, Output]],
Callable[[Input], AsyncIterator[Output]],
],
) -> Runnable[Input, Output]:
"""Decorate a function to make it a Runnable.
Sets the name of the Runnable to the name of the function.
Any runnables called by the function will be traced as dependencies.
Args:
func: A callable.
Returns:
A Runnable.
Example:
.. code-block:: python
from langchain_core.runnables import chain
from langchain_core.prompts import PromptTemplate
from langchain_openai import OpenAI
@chain
def my_func(fields):
prompt = PromptTemplate("Hello, {name}!")
llm = OpenAI()
formatted = prompt.invoke(**fields)
for chunk in llm.stream(formatted):
yield chunk
"""
return RunnableLambda(func)
또는 | 오프레이터를 통해 자동 RunnableLambda를 적용할 수 있다.
prompt = ChatPromptTemplate.from_template("tell me a story about {topic}")
model = ChatOpenAI()
chain_with_coerced_function = prompt | model | (lambda x: x.content[:5])
chain_with_coerced_function.invoke({"topic": "bears"})
// 결과
'Once '
- invoke: abstractmethod로 반드시 구현해야 한다. call the chain on an input (ainvoke)
- stream: stream back chunks of the response (astream)
- batch: call the chain on a list of inputs (abatch)
invoke의 Input, Output 타입은 Generic으로 구현체의 타입을 따른다. 현재 Runnable의 invoke로 호출된 Output은 다음 Runnable의 Input 값이 된다.
class Runnable(Generic[Input, Output], ABC):
@abstractmethod
def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output:
"""Transform a single input into an output. Override to implement.
Args:
input: The input to the Runnable.
config: A config to use when invoking the Runnable.
The config supports standard keys like 'tags', 'metadata' for tracing
purposes, 'max_concurrency' for controlling how much work to do
in parallel, and other keys. Please refer to the RunnableConfig
for more details.
Returns:
The output of the Runnable.
"""
Generic[Input, Output] 의 타입표이고, LangChain의 컴포넌트들이 Runable을 상속받고 있다.
Runnable 체인의 연결은 pipe operatore( | ) 또는 Runnable의 .pipe() 메서드를 사용한다. 이는 마치 RxJS의 pipe 연결과 유사하다.
파이썬은 클래스 인스턴스를 + 기호로 쓸 수 있다. 이때 클래스의 __add__ 를 구현한다. 마찬가지로 | 기호를 사용하기 위해서 __or__ 와 __ror__ 을 구현하고 있다.
// a | b -> a.__or__(b)
def __or__(
self,
other: Union[
Runnable[Any, Other],
Callable[[Any], Other],
Callable[[Iterator[Any]], Iterator[Other]],
Mapping[str, Union[Runnable[Any, Other], Callable[[Any], Other], Any]],
],
) -> RunnableSerializable[Input, Other]:
"""Compose this Runnable with another object to create a RunnableSequence."""
return RunnableSequence(self, coerce_to_runnable(other))
// a | b (if not working) -> b.__ror_(a)
def __ror__(
self,
other: Union[
Runnable[Other, Any],
Callable[[Other], Any],
Callable[[Iterator[Other]], Iterator[Any]],
Mapping[str, Union[Runnable[Other, Any], Callable[[Other], Any], Any]],
],
) -> RunnableSerializable[Other, Output]:
"""Compose this Runnable with another object to create a RunnableSequence."""
return RunnableSequence(coerce_to_runnable(other), self)
// 끝에 dictionary 타입을 Runnable로 만들어 준다.
def coerce_to_runnable(thing: RunnableLike) -> Runnable[Input, Output]:
"""Coerce a Runnable-like object into a Runnable.
Args:
thing: A Runnable-like object.
Returns:
A Runnable.
Raises:
TypeError: If the object is not Runnable-like.
"""
if isinstance(thing, Runnable):
return thing
elif is_async_generator(thing) or inspect.isgeneratorfunction(thing):
return RunnableGenerator(thing)
elif callable(thing):
return RunnableLambda(cast(Callable[[Input], Output], thing))
elif isinstance(thing, dict):
return cast(Runnable[Input, Output], RunnableParallel(thing))
else:
raise TypeError(
f"Expected a Runnable, callable or dict."
f"Instead got an unsupported type: {type(thing)}"
)
이 코드에서 정의된 __or__ 메서드는 Python의 특수 메서드 중 하나로, 비트 연산자 | 를 오버로딩하여 객체 간의 조합을 가능하게 합니다. 이 메서드는 RunnableSerializable 클래스에 정의되어 있으며, 이 클래스의 인스턴스를 다른 객체와 조합하여 새로운 RunnableSequence를 생성합니다.
__or__ 메서드의 의미와 역할
1. __or__ 메서드 오버로딩
이 메서드는 | 연산자를 사용했을 때 호출됩니다. 예를 들어, a | b와 같이 사용할 때, a.__or__(b)가 호출됩니다.
from langchain_core.runnables import RunnableParallel
composed_chain_with_pipe = (
RunnableParallel({"joke": chain})
.pipe(analysis_prompt)
.pipe(model)
.pipe(StrOutputParser())
)
composed_chain_with_pipe.invoke({"topic": "battlestar galactica"})
// 결과
"I cannot reproduce any copyrighted material verbatim, but I can try to analyze the humor in the joke you provided without quoting it directly.\n\nThe joke plays on the idea that the Cylon raiders, who are the antagonists in the Battlestar Galactica universe, failed to locate the human survivors after attacking their home planets (the Twelve Colonies) due to using an outdated and poorly performing operating system (Windows Vista) for their targeting systems.\n\nThe humor stems from the juxtaposition of a futuristic science fiction setting with a relatable real-world frustration – the use of buggy, slow, or unreliable software or technology. It pokes fun at the perceived inadequacies of Windows Vista, which was widely criticized for its performance issues and other problems when it was released.\n\nBy attributing the Cylons' failure to locate the humans to their use of Vista, the joke creates an amusing and unexpected connection between a fictional advanced race of robots and a familiar technological annoyance experienced by many people in the real world.\n\nOverall, the joke relies on incongruity and relatability to generate humor, but without reproducing any copyrighted material directly."
// 또는 일렇게 사용도 가능하다.
composed_chain_with_pipe = RunnableParallel({"joke": chain}).pipe(
analysis_prompt, model, StrOutputParser()
)
파이프 예제 및 Runnable 구현체들
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
// model runnable
model = ChatOpenAI(model="gpt-4o-mini")
// prompt runnable
prompt = ChatPromptTemplate.from_template("tell me a joke about {topic}")
// output parser runnable
chain = prompt | model | StrOutputParser()
chain.invoke({"topic": "bears"})
// 결과
"Here's a bear joke for you:\n\nWhy did the bear dissolve in water?\nBecause it was a polar bear!"
Runnable 구현체들은 langchain_core.runnables 폴더의 base.py에 구현되어 있다.
- RunnableSerializable : Serialize 는 보통 파일, 네트워크 전송을 가능하게 한다. 여기서는 json 으로 serialization이 가능하다.
from langchain_core.runnables import RunnableLambda
def add_one(x: int) -> int:
return x + 1
def mul_two(x: int) -> int:
return x * 2
def mul_three(x: int) -> int:
return x * 3
runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
runnable_3 = RunnableLambda(mul_three)
sequence = runnable_1 | { # this dict is coerced to a RunnableParallel
"mul_two": runnable_2,
"mul_three": runnable_3,
}
# Or equivalently:
# sequence = runnable_1 | RunnableParallel(
# {"mul_two": runnable_2, "mul_three": runnable_3}
# )
# Also equivalently:
# sequence = runnable_1 | RunnableParallel(
# mul_two=runnable_2,
# mul_three=runnable_3,
# )
sequence.invoke(1)
await sequence.ainvoke(1)
// 결과
{'mul_two': 4, 'mul_three': 6}
sequence.batch([1, 2, 3])
await sequence.abatch([1, 2, 3])
// 결과
[{'mul_two': 4, 'mul_three': 6},
{'mul_two': 6, 'mul_three': 9},
{'mul_two': 8, 'mul_three': 12}]
LLM 호출 RunnableParallel 예제
from langchain_core.output_parsers import StrOutputParser
analysis_prompt = ChatPromptTemplate.from_template("is this a funny joke? {joke}")
composed_chain = {"joke": chain} | analysis_prompt | model | StrOutputParser()
// RunnableParallel 로 전환된다.
composed_chain.invoke({"topic": "bears"})
// 결과
'Haha, that\'s a clever play on words! Using "polar" to imply the bear dissolved or became polar/polarized when put in water. Not the most hilarious joke ever, but it has a cute, groan-worthy pun that makes it mildly amusing. I appreciate a good pun or wordplay joke.'
chain 의 output 을 lambda 의 input으로 받아 dictionary 를 만들어 체인을 구성할 수도 있다.
composed_chain_with_lambda = (
chain
| (lambda input: {"joke": input})
| analysis_prompt
| model
| StrOutputParser()
)
composed_chain_with_lambda.invoke({"topic": "beets"})
// 결과
"Haha, that's a cute and punny joke! I like how it plays on the idea of beets blushing or turning red like someone blushing. Food puns can be quite amusing. While not a total knee-slapper, it's a light-hearted, groan-worthy dad joke that would make me chuckle and shake my head. Simple vegetable humor!"
from typing import Any, AsyncIterator, Iterator
from langchain_core.runnables import RunnableGenerator
def gen(input: Iterator[Any]) -> Iterator[str]:
for token in ["Have", " a", " nice", " day"]:
yield token
runnable = RunnableGenerator(gen)
runnable.invoke(None) # "Have a nice day"
list(runnable.stream(None)) # ["Have", " a", " nice", " day"]
runnable.batch([None, None]) # ["Have a nice day", "Have a nice day"]
# Async version:
async def agen(input: AsyncIterator[Any]) -> AsyncIterator[str]:
for token in ["Have", " a", " nice", " day"]:
yield token
runnable = RunnableGenerator(agen)
await runnable.ainvoke(None) # "Have a nice day"
[p async for p in runnable.astream(None)] # ["Have", " a", " nice", " day"]
- 그외 RunnableEachBase, RunableEach, RunnableBindingBase, RunnableBinding
모든 Runnable은 input 과 output schema를 노출한다.
- input_schema: input pydantic model
- output_schema: output pydantic model
예제는 주로 invoke, ainvoke를 한번에 output을 받는 것이고, LLM을 통해 응답을 받을때는 UX개선을 위하 stream으로 받아 처리한다. Runnable stream 처리에 대한 다양한 접근법은 공식문서를 참조한다.
Runnable 개념 정리
LangChain Runnable과 LangChain Expression Language (LCEL)은 LLM(대형 언어 모델)의 기능을 활용하여 견고한 프로덕션 수준의 프로그램을 만들 수 있는 강력한 도구입니다. 다음은 이들의 동작 방식과 중요성에 대한 개요입니다.
LangChain Runnable
LangChain Runnable은 실행할 수 있는 작업이나 기능을 캡슐화한 개념으로, 일반적으로 일련의 계산 단계를 나타냅니다. Runnable은 LangChain에서 복잡한 워크플로우 또는 체인을 구성할 수 있는 기본 구성 요소입니다. 각 Runnable은 유연하게 설계되어 동기, 비동기, 배치, 스트리밍 작업을 처리할 수 있습니다.
LangChain Expression Language (LCEL)
LangChain Expression Language (LCEL)은 이러한 워크플로우 또는 프로그램을 구성할 수 있는 선언적 언어입니다. LCEL을 사용하면 여러 Runnable을 명확하고 구조적인 방식으로 결합할 수 있습니다. LCEL을 통해 복잡한 작업 체인을 정의하고 여러 Runnable을 하나의 프로그램으로 결합할 수 있습니다.
LCEL과 Runnables의 주요 기능
동기 작업: 전통적인 차단 작업으로, 각 단계가 완료된 후 다음 단계로 이동합니다.
비동기 작업: 차단되지 않는 작업으로, 특정 작업이 완료되기를 기다리는 동안 프로그램이 다른 작업을 처리할 수 있습니다. 이는 서버 환경에서 여러 요청을 동시에 처리할 때 특히 중요합니다.
배치 처리: 여러 입력을 동시에 처리할 수 있는 기능입니다. 대량의 데이터를 처리하거나 여러 사용자의 쿼리를 동시에 처리할 때 유용합니다.
스트리밍: 중간 결과가 생성됨에 따라 이를 스트리밍할 수 있는 기능으로, 사용자 경험을 개선할 수 있습니다. 예를 들어, LLM으로 텍스트를 생성할 때 전체 텍스트가 생성되기 전에 부분적인 응답을 사용자에게 스트리밍할 수 있습니다.
사용 예시
배치 처리: LLM을 통해 고객 쿼리 데이터를 분류해야 하는 시나리오를 상상해 보십시오. LCEL과 Runnables을 사용하여 이 쿼리들을 배치 처리하고 병렬로 처리할 수 있어 작업 속도가 크게 향상됩니다.
비동기 작업: 여러 클라이언트 요청을 처리하는 웹 서버에서 LCEL을 사용하여 이러한 요청을 비동기로 관리함으로써, 서버가 여러 트래픽을 동시에 처리할 수 있게 합니다.
스트리밍 결과: 실시간으로 사용자와 상호작용하는 챗봇의 경우, LCEL을 사용하여 부분적으로 생성된 텍스트로 바로 응답을 시작함으로써 더 인터랙티브하고 반응성이 좋은 사용자 경험을 제공할 수 있습니다.
프로덕션 수준 프로그램에서의 LCEL의 역할
LCEL의 설계는 LLM의 기능을 활용하는 복잡한 워크플로우를 보다 쉽게 구축하고 관리할 수 있도록 하는 데 중점을 둡니다. 이는 다양한 작업 모드(동기, 비동기, 배치, 스트리밍)를 처리하는 데 있어 복잡성을 추상화하여, 개발자가 워크플로우의 논리 정의에 집중할 수 있도록 합니다. 이러한 접근 방식은 특히 신뢰성, 확장성, 성능이 중요한 프로덕션 환경에서 매우 가치가 있습니다.
구현 및 스키마
LangChain Runnables 및 LCEL의 구현은 프로그램의 다양한 구성 요소가 어떻게 상호작용하는지를 설명하는 스키마를 정의하는 것입니다. 이러한 스키마는 체인의 각 구성 요소가 올바른 유형의 입력을 받고 예상되는 유형의 출력을 생성하도록 보장합니다. 이러한 구조화된 접근 방식은 신뢰할 수 있는 프로그램을 구축하는 데 도움이 될 뿐만 아니라, 프로그램을 유지하고 확장하는 데도 유리합니다.
결론
LangChain Runnables과 LCEL은 LLM을 통합한 워크플로우를 구축하고 관리할 수 있는 강력한 패러다임을 제공합니다. 이들은 다양한 작업 모드를 지원하는 유연하고 확장 가능한 프레임워크를 제공하여, 복잡한 데이터 처리 작업을 간단하게 구성하고 강력한 처리 체인을 만들 수 있습니다. LCEL과 Runnables을 활용하여 효율적이고 반응성이 뛰어난 확장 가능한 솔루션을 만들 수 있습니다.
위의 구조에서 my_package는 패키지입니다. 이 패키지에는 module1.py와 module2.py라는 두 개의 모듈과, sub_package라는 하위 패키지가 포함되어 있습니다. 하위 패키지 sub_package에는 module3.py라는 모듈이 포함되어 있습니다.
이러한 구조에서는 다음과 같이 모듈을 가져올 수 있습니다:
from my_package import module1
from my_package.sub_package import module3
패키지와 모듈의 차이점
•구조:
•모듈은 단일 .py 파일로 구성됩니다.
•패키지는 여러 모듈(그리고 하위 패키지)을 포함할 수 있는 디렉토리입니다.
•범위:
•모듈은 단일 파일에 포함된 코드 단위입니다.
•패키지는 모듈과 하위 패키지를 조직화하는 더 큰 단위입니다.
•내포 관계:
•모듈은 패키지 안에 포함될 수 있습니다.
•패키지는 모듈들을 그룹화하는 역할을 하며, 패키지 내에 또 다른 패키지를 포함할 수도 있습니다.
요약
•모듈: Python 코드가 담긴 단일 .py 파일. 모듈은 함수, 클래스, 변수 등의 코드를 포함할 수 있으며, 다른 Python 코드에서 import를 통해 재사용할 수 있습니다.
•패키지: 모듈들을 조직화하는 디렉토리 구조. 패키지는 모듈과 다른 하위 패키지를 포함할 수 있으며, 패키지로 인식되기 위해서는 일반적으로 __init__.py 파일이 필요합니다.
함수 정의에서 파라미터 앞에 붙는 *는 여러 가지 의미를 가질 수 있지만, 일반적으로 **위치 인수(variable-length positional arguments)**를 나타내는 데 사용됩니다. 이 기호는 함수가 호출될 때 임의의 개수의 위치 인수를 받아들이도록 하는 역할을 합니다.
*의 주요 의미
1.가변 위치 인수 (*args):
•함수 정의에서 *args와 같은 형태로 사용되며, 이 함수는 호출 시 여러 개의 위치 인수를 하나의 튜플로 받아들일 수 있습니다.
LangChain 소스 코드를 분석하면서 Python도 함께 배우고 있습니다. 이해되지 않는 것으 GPT를 통해 열심히 개념을 알아가고 있습니다.
1. Union에 대한 궁금증
Union은 Python의 타입 힌팅(type hinting)에서 사용되며, 함수의 매개변수나 반환 값이 둘 이상의 서로 다른 타입을 가질 수 있음을 나타낼 때 사용됩니다. Python 3.10 이전에는 typing 모듈의 Union을 사용했으며, Python 3.10부터는 더 간단한 | 연산자를 사용하여 동일한 기능을 제공할 수 있습니다.
예시
1.Python 3.9 이하에서의 Union 사용:
from typing import Union
def process_value(value: Union[int, float]) -> Union[int, float]:
return value * 2
이 함수 process_value는 int나 float 타입의 값을 받아서 두 배로 증가시킨 후 반환합니다. 반환 값도 int나 float이 될 수 있음을 Union을 통해 명시하고 있습니다.
2.Python 3.10 이후에서의 | 연산자 사용:
def process_value(value: int | float) -> int | float:
return value * 2
위와 동일한 기능을 | 연산자를 사용하여 간단히 표현할 수 있습니다.
Union 사용이 적합한 경우
•다양한 타입을 처리할 때: 함수나 메소드가 여러 타입을 받을 수 있고, 그 타입에 따라 다르게 동작할 때 Union을 사용합니다.
•점진적 타입 검사를 통해 오류를 줄일 때: 코드에 타입 힌트를 명확히 하여, 코드의 가독성과 안정성을 높이고, 코드 편집기나 정적 분석 도구가 더 나은 오류 검출을 할 수 있도록 도울 수 있습니다.
주의사항
•Union을 사용할 때, 가능한 한 사용 범위를 좁히는 것이 좋습니다. 너무 많은 타입을 허용하면 코드가 복잡해지고 유지보수가 어려워질 수 있습니다.
Union은 주로 함수의 입력과 출력에 다양한 타입을 허용할 때 매우 유용한 도구입니다. 필요에 따라 적절히 사용하는 것이 중요합니다.
2. ininstance 에 대한 궁금증
isinstance는 Python에서 특정 객체가 특정 클래스나 타입의 인스턴스인지 확인할 때 사용되는 내장 함수입니다. 이 함수는 객체와 클래스(또는 클래스의 튜플)를 인수로 받아, 객체가 해당 클래스의 인스턴스이거나 그 클래스의 서브클래스의 인스턴스인 경우 True를 반환하고, 그렇지 않으면 False를 반환합니다.
isinstance의 기본 사용법
result = isinstance(object, classinfo)
•object: 타입을 확인하려는 객체.
•classinfo: 클래스, 클래스의 튜플, 또는 타입들.
예시
1.단일 클래스에 대한 검사:
# 객체가 정수형인지 확인
x = 10
print(isinstance(x, int)) # True
2.여러 클래스에 대한 검사:
# 객체가 정수형 또는 실수형인지 확인
x = 10.5
print(isinstance(x, (int, float))) # True
4.상속 관계에서의 검사:
class Animal:
pass
class Dog(Animal):
pass
dog = Dog()
print(isinstance(dog, Animal)) # True
print(isinstance(dog, Dog)) # True
이 예제에서 Dog 클래스는 Animal 클래스를 상속받기 때문에, dog 객체는 Animal 클래스의 인스턴스이기도 합니다.
isinstance의 주요 용도
•타입 확인: 함수나 메소드 내부에서 인수의 타입을 확인하여, 타입에 따라 다른 로직을 수행할 때 유용합니다.
•상속 관계 검사: 객체가 특정 클래스에서 파생된 클래스의 인스턴스인지 확인할 수 있습니다.
•입력 검증: 함수에 전달된 인수의 타입이 예상한 타입과 일치하는지 검증하여, 타입 오류를 미리 방지할 수 있습니다.
isinstance와 type의 차이점
isinstance는 상속 관계를 고려하는 반면, type 함수는 객체의 정확한 타입을 확인할 때 사용됩니다. 예를 들어, 상속받은 클래스의 인스턴스는 type을 사용하면 부모 클래스와 매칭되지 않습니다.
class Animal:
pass
class Dog(Animal):
pass
dog = Dog()
# isinstance를 사용하면 True
print(isinstance(dog, Animal)) # True
# type을 사용하면 False
print(type(dog) == Animal) # False, Dog != Animal
이와 같이, isinstance는 더 유연하게 타입을 확인할 수 있는 반면, type은 정확한 타입 비교에 사용됩니다.
이 코드에서 kwargs 변수를 정의하는 방식은 Python에서 딕셔너리를 병합하는 기법 중 하나입니다. 이 코드의 주요 목적은 두 개의 딕셔너리를 결합하여 하나의 딕셔너리를 만드는 것입니다. 여기서 kwargs는 일반적으로 함수나 메소드에서 사용되는 키워드 인수들을 담고 있는 딕셔너리입니다.
•key 매개변수를 사용하여 정렬 기준을 설정할 수 있으며, reverse 매개변수를 사용하여 정렬 방향(오름차순/내림차순)을 설정할 수 있습니다.
•원본 데이터는 변경되지 않으며, 정렬된 새로운 리스트가 반환됩니다.
3. cast 에 대한 궁금증
cast는 Python의 타입 힌팅(type hinting)과 관련된 기능 중 하나로, typing 모듈에서 제공하는 함수입니다. cast는 특정 값을 지정된 타입으로 “캐스팅”한다고 표시하는 역할을 합니다. 그러나 실제로 값을 변환하거나 변경하지는 않으며, 주로 타입 힌팅을 통해 코드의 가독성을 높이고, 정적 분석 도구들이 올바른 타입을 추론하도록 돕기 위한 목적으로 사용됩니다.
cast의 기본 사용법
from typing import cast
cast(typ, val)
•typ: 캐스팅할 타입. Python의 타입(예: int, str, List[int] 등)이나 사용자 정의 클래스 등을 지정할 수 있습니다.
•val: 실제 값. typ로 캐스팅할 값입니다.
cast의 역할
•정적 분석 지원: cast는 Python 코드에서 변수나 표현식의 타입을 명시적으로 지정하는 데 사용됩니다. 정적 분석 도구나 IDE에서 코드의 타입을 더 잘 이해하고, 타입 관련 경고나 오류를 감지하는 데 도움이 됩니다.
•런타임에는 아무 영향 없음: cast는 런타임에 아무런 영향을 미치지 않습니다. 즉, cast를 사용해도 실제로 값의 타입이 변경되거나 변환되지 않습니다.
여기서 cast(List[str], get_items())는 get_items()의 반환값이 List[str] 타입임을 명시적으로 지정합니다.
2.변수 타입 힌트:
from typing import cast
value: int = cast(int, "42") # 정적 분석 도구에게 value가 int임을 알려줌
print(value) # "42", 실제로는 str 타입임
이 예시에서 cast(int, "42")는 "42"가 int 타입이라고 명시하지만, 실제로는 str 타입입니다. 이 코드에서 cast는 단지 정적 타입 힌팅을 위한 것이며, value의 타입이 실제로 int로 변환되지 않습니다.
3.사용자 정의 클래스:
from typing import cast
class Animal:
def speak(self):
pass
class Dog(Animal):
def speak(self):
return "Woof!"
class Cat(Animal):
def speak(self):
return "Meow!"
def get_animal() -> Animal:
return Dog()
animal = cast(Dog, get_animal())
print(animal.speak()) # "Woof!", 정적 분석 도구는 animal을 Dog로 간주함
여기서 cast(Dog, get_animal())는 get_animal()이 반환하는 객체가 Dog 타입임을 명시적으로 지정하여, animal 변수가 Dog로 간주되도록 합니다. 실제로는 get_animal()이 반환하는 객체가 Dog이기 때문에 문제가 없지만, 정적 분석 도구에 명시적으로 알려주기 위해 사용됩니다.
요약
•cast는 Python에서 타입 힌팅을 명시적으로 지정하기 위한 도구로, 런타임에는 영향을 미치지 않습니다.
•주로 정적 분석 도구나 IDE에서 타입 추론을 돕기 위해 사용됩니다.
•cast는 값의 실제 타입을 변경하지 않으며, 코드의 가독성과 안전성을 높이는 역할을 합니다.
return cls(messages, template_format=template_format)는 Python에서 클래스 메소드나 다른 클래스 내부 메소드에서 새로운 클래스 인스턴스를 생성하여 반환할 때 사용되는 패턴입니다. 이 문장에서 cls는 현재 클래스 자체를 가리키고, messages와 template_format은 그 클래스의 생성자 __init__ 메소드에 전달되는 인수입니다.
이 코드의 의미
cls:
cls는 클래스 메소드 또는 클래스 내부의 다른 메소드에서 해당 클래스 자체를 참조하는 키워드입니다. cls를 사용하면 현재 클래스의 인스턴스를 생성할 수 있습니다.
일반적으로 @classmethod 데코레이터가 붙은 메소드 내에서 사용되며, 이 메소드가 호출될 때 해당 클래스 자체가 첫 번째 인수로 cls에 전달됩니다.
cls(messages, template_format=template_format):
이 구문은 현재 클래스의 인스턴스를 생성하는 표현입니다.
messages와 template_format=template_format은 생성자에 전달되는 인수입니다. 여기서 messages는 위치 인수로, template_format은 키워드 인수로 전달됩니다.
이는 클래스의 __init__ 메소드가 messages와 template_format이라는 두 개의 인수를 받을 것으로 예상된다는 것을 의미합니다.
return:
새로 생성된 클래스 인스턴스를 반환합니다. 이 반환된 객체는 이 메소드를 호출한 코드에서 사용할 수 있게 됩니다.
예시 코드
아래는 이 패턴이 어떻게 사용되는지에 대한 예시입니다:
class PromptTemplate:
def __init__(self, messages, template_format=None):
self.messages = messages
self.template_format = template_format
@classmethod
def from_messages(cls, messages, template_format="default"):
# 새로운 PromptTemplate 인스턴스를 생성하여 반환
return cls(messages, template_format=template_format)
# 사용 예시
messages = ["Hello, World!", "How are you?"]
template = PromptTemplate.from_messages(messages, template_format="custom")
print(template.messages) # ['Hello, World!', 'How are you?']
print(template.template_format) # 'custom'
요약
return cls(messages, template_format=template_format)는 현재 클래스의 인스턴스를 생성하여 반환하는 코드입니다.
cls는 해당 클래스 자체를 참조하며, 이 코드 구문은 클래스 메소드 또는 클래스 내부의 다른 메소드에서 새로운 인스턴스를 생성하기 위해 사용됩니다.
이 패턴은 클래스의 __init__ 메소드가 해당 인수들을 받아들일 것으로 예상합니다. 따라서 이 코드가 사용되는 클래스는 __init__ 메소드가 messages와 template_format을 인수로 받아야 합니다.
class PromptTemplate(StringPromptTemplate):
@classmethod
def from_template(
cls,
template: str,
*,
template_format: str = "f-string",
partial_variables: Optional[Dict[str, Any]] = None,
**kwargs: Any,
) -> PromptTemplate:
"""Load a prompt template from a template.
*Security warning*:
Prefer using `template_format="f-string"` instead of
`template_format="jinja2"`, or make sure to NEVER accept jinja2 templates
from untrusted sources as they may lead to arbitrary Python code execution.
As of LangChain 0.0.329, Jinja2 templates will be rendered using
Jinja2's SandboxedEnvironment by default. This sand-boxing should
be treated as a best-effort approach rather than a guarantee of security,
as it is an opt-out rather than opt-in approach.
Despite the sand-boxing, we recommend never using jinja2 templates
from untrusted sources.
Args:
template: The template to load.
template_format: The format of the template. Use `jinja2` for jinja2,
and `f-string` or None for f-strings.
Defaults to `f-string`.
partial_variables: A dictionary of variables that can be used to partially
fill in the template. For example, if the template is
`"{variable1} {variable2}"`, and `partial_variables` is
`{"variable1": "foo"}`, then the final prompt will be
`"foo {variable2}"`. Defaults to None.
kwargs: Any other arguments to pass to the prompt template.
Returns:
The prompt template loaded from the template.
"""
input_variables = get_template_variables(template, template_format)
_partial_variables = partial_variables or {}
if _partial_variables:
input_variables = [
var for var in input_variables if var not in _partial_variables
]
return cls(
input_variables=input_variables,
template=template,
template_format=template_format, # type: ignore[arg-type]
partial_variables=_partial_variables,
**kwargs,
)
PromptTemplate 인스턴스의 invoke 호출하기
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
# ChatOpenAI 모델을 인스턴스화합니다.
model = ChatOpenAI()
# 주어진 토픽에 대한 농담을 요청하는 프롬프트 템플릿을 생성합니다.
prompt = PromptTemplate.from_template("{topic} 에 대하여 3문장으로 설명해줘.")
# invoke 호출시에 topic 입력값을 준다.
chain = prompt | model | StrOutputParser()
chain.invoke({"topic": "ChatGPT"})
출처: LangChain KR
LCEL에 적용할 수 있는 것은 전부 Runnable을 상속받아야 하고, Runnable의 abstractmethod인 invoke를 구현해야 한다. "chain.invoke" 호출은 최초 Runnable의 invoke를 구현하고 있는 BasePromptTemplate --> StringPromptTemplate --> PromptTemplate 상속 관계에서 Runnable을 상속받아 invoke 메서드를 구현하고 있는 BasePromptTemplate의 invoke 보자.
- 입력값: Dictionary 예) {"topic": "ChatGPT"}
- 반환값: PromptValue 인스턴스이다.
class BasePromptTemplate(
RunnableSerializable[Dict, PromptValue], Generic[FormatOutputType], ABC
):
... 중략 ...
def invoke(
self, input: Dict, config: Optional[RunnableConfig] = None
) -> PromptValue:
"""Invoke the prompt.
Args:
input: Dict, input to the prompt.
config: RunnableConfig, configuration for the prompt.
Returns:
PromptValue: The output of the prompt.
"""
config = ensure_config(config)
if self.metadata:
config["metadata"] = {**config["metadata"], **self.metadata}
if self.tags:
config["tags"] = config["tags"] + self.tags
return self._call_with_config(
self._format_prompt_with_error_handling,
input,
config,
run_type="prompt",
)
- from_messages 로 system, human mesage 를 구성 -> format_messages (또는 format) 으로 입력값 key=value 설정
- LLM 호출전에 format_messages 형태로
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
chat_prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"당신은 요약 전문 AI 어시스턴트입니다. 당신의 임무는 주요 키워드로 대화를 요약하는 것입니다.",
),
MessagesPlaceholder(variable_name="conversation"),
("human", "지금까지의 대화를 {word_count} 단어로 요약합니다."),
]
)
...
// 호출 case-1)
messages = chat_prompt.format_messages(
word_count=5,
conversation=[
("human", "안녕하세요! 저는 오늘 새로 입사한 테디 입니다. 만나서 반갑습니다."),
("ai", "반가워요! 앞으로 잘 부탁 드립니다."),
],
)
llm.invoke(messages)
// 호출 case-2) 또는 LCEL을 사용하여 chain 을 구성 후 입력값 설정하기
chain = chat_template | llm
chain.invoke({word_count=5,
conversation=[
("human", "안녕하세요! 저는 오늘 새로 입사한 테디 입니다. 만나서 반갑습니다."),
("ai", "반가워요! 앞으로 잘 부탁 드립니다."),
]})
// 출처: LangChain KR
예의 from_messages 호출 반환 chat_prompt 객체 내역
ChatPromptTemplate(input_variables=['conversation', 'word_count'], input_types={'conversation': typing.List[typing.Union[langchain_core.messages.ai.AIMessage, langchain_core.messages.human.HumanMessage, langchain_core.messages.chat.ChatMessage, langchain_core.messages.system.SystemMessage, langchain_core.messages.function.FunctionMessage, langchain_core.messages.tool.ToolMessage]]}, messages=[SystemMessagePromptTemplate(prompt=PromptTemplate(input_variables=[], template='당신은 요약 전문 AI 어시스턴트입니다. 당신의 임무는 주요 키워드로 대화를 요약하는 것입니다.')), MessagesPlaceholder(variable_name='conversation'), HumanMessagePromptTemplate(prompt=PromptTemplate(input_variables=['word_count'], template='지금까지의 대화를 {word_count} 단어로 요약합니다.'))])
예의 format_message 호출 반환 messages 객체 내역
[SystemMessage(content='당신은 요약 전문 AI 어시스턴트입니다. 당신의 임무는 주요 키워드로 대화를 요약하는 것입니다.'),
HumanMessage(content='안녕하세요! 저는 오늘 새로 입사한 테디 입니다. 만나서 반갑습니다.'),
AIMessage(content='반가워요! 앞으로 잘 부탁 드립니다.'),
HumanMessage(content='지금까지의 대화를 5 단어로 요약합니다.')]
반환 타입에서 messages 서브 패키지의 다양한 메세지타입이 보인다. 이는 다음의 상속관계를 갖는다.
langchain_core 패키지의 prompts 와 messages 서브패키지를 사용합니다. 아래 상속관계를 보면 BaseMessagePromptTemplate 는 개별 메시지 단위로 프롬프트를 구성하는 것이고, BasePromptTemplate를 상속받은 ChatPromptTemplate은 BaseMessagePromptTemplate 상속받은 클래스들과 str, tuple등의 메세지를 담을 수 있다.
class ChatPromptTemplate(BaseChatPromptTemplate):
"""Prompt template for chat models.
Use to create flexible templated prompts for chat models.
Examples:
.. versionchanged:: 0.2.24
You can pass any Message-like formats supported by
``ChatPromptTemplate.from_messages()`` directly to ``ChatPromptTemplate()``
init.
.. code-block:: python
from langchain_core.prompts import ChatPromptTemplate
template = ChatPromptTemplate([
("system", "You are a helpful AI bot. Your name is {name}."),
("human", "Hello, how are you doing?"),
("ai", "I'm doing well, thanks!"),
("human", "{user_input}"),
])
prompt_value = template.invoke(
{
"name": "Bob",
"user_input": "What is your name?"
}
)
# Output:
# ChatPromptValue(
# messages=[
# SystemMessage(content='You are a helpful AI bot. Your name is Bob.'),
# HumanMessage(content='Hello, how are you doing?'),
# AIMessage(content="I'm doing well, thanks!"),
# HumanMessage(content='What is your name?')
# ]
#)
Messages Placeholder:
.. code-block:: python
# In addition to Human/AI/Tool/Function messages,
# you can initialize the template with a MessagesPlaceholder
# either using the class directly or with the shorthand tuple syntax:
template = ChatPromptTemplate([
("system", "You are a helpful AI bot."),
# Means the template will receive an optional list of messages under
# the "conversation" key
("placeholder", "{conversation}")
# Equivalently:
# MessagesPlaceholder(variable_name="conversation", optional=True)
])
prompt_value = template.invoke(
{
"conversation": [
("human", "Hi!"),
("ai", "How can I assist you today?"),
("human", "Can you make me an ice cream sundae?"),
("ai", "No.")
]
}
)
# Output:
# ChatPromptValue(
# messages=[
# SystemMessage(content='You are a helpful AI bot.'),
# HumanMessage(content='Hi!'),
# AIMessage(content='How can I assist you today?'),
# HumanMessage(content='Can you make me an ice cream sundae?'),
# AIMessage(content='No.'),
# ]
#)
Single-variable template:
If your prompt has only a single input variable (i.e., 1 instance of "{variable_nams}"),
and you invoke the template with a non-dict object, the prompt template will
inject the provided argument into that variable location.
.. code-block:: python
from langchain_core.prompts import ChatPromptTemplate
template = ChatPromptTemplate([
("system", "You are a helpful AI bot. Your name is Carl."),
("human", "{user_input}"),
])
prompt_value = template.invoke("Hello, there!")
# Equivalent to
# prompt_value = template.invoke({"user_input": "Hello, there!"})
# Output:
# ChatPromptValue(
# messages=[
# SystemMessage(content='You are a helpful AI bot. Your name is Carl.'),
# HumanMessage(content='Hello, there!'),
# ]
# )
""" # noqa: E501
messages: List[MessageLike]
"""List of messages consisting of either message prompt templates or messages."""
validate_template: bool = False
"""Whether or not to try validating the template."""
def __init__(
self,
messages: Sequence[MessageLikeRepresentation],
*,
template_format: Literal["f-string", "mustache", "jinja2"] = "f-string",
**kwargs: Any,
) -> None:
"""Create a chat prompt template from a variety of message formats.
Args:
messages: sequence of message representations.
A message can be represented using the following formats:
(1) BaseMessagePromptTemplate, (2) BaseMessage, (3) 2-tuple of
(message type, template); e.g., ("human", "{user_input}"),
(4) 2-tuple of (message class, template), (4) a string which is
shorthand for ("human", template); e.g., "{user_input}".
template_format: format of the template. Defaults to "f-string".
input_variables: A list of the names of the variables whose values are
required as inputs to the prompt.
optional_variables: A list of the names of the variables that are optional
in the prompt.
partial_variables: A dictionary of the partial variables the prompt
template carries. Partial variables populate the template so that you
don't need to pass them in every time you call the prompt.
validate_template: Whether to validate the template.
input_types: A dictionary of the types of the variables the prompt template
expects. If not provided, all variables are assumed to be strings.
Returns:
A chat prompt template.
Examples:
Instantiation from a list of message templates:
.. code-block:: python
template = ChatPromptTemplate([
("human", "Hello, how are you?"),
("ai", "I'm doing well, thanks!"),
("human", "That's good to hear."),
])
Instantiation from mixed message formats:
.. code-block:: python
template = ChatPromptTemplate([
SystemMessage(content="hello"),
("human", "Hello, how are you?"),
])
"""
_messages = [
_convert_to_message(message, template_format) for message in messages
]
# Automatically infer input variables from messages
input_vars: Set[str] = set()
optional_variables: Set[str] = set()
partial_vars: Dict[str, Any] = {}
for _message in _messages:
if isinstance(_message, MessagesPlaceholder) and _message.optional:
partial_vars[_message.variable_name] = []
optional_variables.add(_message.variable_name)
elif isinstance(
_message, (BaseChatPromptTemplate, BaseMessagePromptTemplate)
):
input_vars.update(_message.input_variables)
kwargs = {
**dict(
input_variables=sorted(input_vars),
optional_variables=sorted(optional_variables),
partial_variables=partial_vars,
),
**kwargs,
}
cast(Type[ChatPromptTemplate], super()).__init__(messages=_messages, **kwargs)
def _convert_to_message(
message: MessageLikeRepresentation,
template_format: Literal["f-string", "mustache", "jinja2"] = "f-string",
) -> Union[BaseMessage, BaseMessagePromptTemplate, BaseChatPromptTemplate]:
"""Instantiate a message from a variety of message formats.
The message format can be one of the following:
- BaseMessagePromptTemplate
- BaseMessage
- 2-tuple of (role string, template); e.g., ("human", "{user_input}")
- 2-tuple of (message class, template)
- string: shorthand for ("human", template); e.g., "{user_input}"
Args:
message: a representation of a message in one of the supported formats.
template_format: format of the template. Defaults to "f-string".
Returns:
an instance of a message or a message template.
Raises:
ValueError: If unexpected message type.
ValueError: If 2-tuple does not have 2 elements.
"""
if isinstance(message, (BaseMessagePromptTemplate, BaseChatPromptTemplate)):
_message: Union[
BaseMessage, BaseMessagePromptTemplate, BaseChatPromptTemplate
] = message
elif isinstance(message, BaseMessage):
_message = message
elif isinstance(message, str):
_message = _create_template_from_message_type(
"human", message, template_format=template_format
)
elif isinstance(message, tuple):
if len(message) != 2:
raise ValueError(f"Expected 2-tuple of (role, template), got {message}")
message_type_str, template = message
if isinstance(message_type_str, str):
_message = _create_template_from_message_type(
message_type_str, template, template_format=template_format
)
else:
_message = message_type_str(
prompt=PromptTemplate.from_template(
cast(str, template), template_format=template_format
)
)
else:
raise NotImplementedError(f"Unsupported message type: {type(message)}")
return _message
API
클래스 메서드인 from_messages() 를 호출하면 cls를 반환하므로 ChatPromptTemplate 인스턴스를 반환한다.
- from_messages() -> ChatPromptTemplate
@classmethod
def from_messages(
cls,
messages: Sequence[MessageLikeRepresentation],
template_format: Literal["f-string", "mustache", "jinja2"] = "f-string",
) -> ChatPromptTemplate:
"""Create a chat prompt template from a variety of message formats.
Examples:
Instantiation from a list of message templates:
.. code-block:: python
template = ChatPromptTemplate.from_messages([
("human", "Hello, how are you?"),
("ai", "I'm doing well, thanks!"),
("human", "That's good to hear."),
])
Instantiation from mixed message formats:
.. code-block:: python
template = ChatPromptTemplate.from_messages([
SystemMessage(content="hello"),
("human", "Hello, how are you?"),
])
Args:
messages: sequence of message representations.
A message can be represented using the following formats:
(1) BaseMessagePromptTemplate, (2) BaseMessage, (3) 2-tuple of
(message type, template); e.g., ("human", "{user_input}"),
(4) 2-tuple of (message class, template), (4) a string which is
shorthand for ("human", template); e.g., "{user_input}".
template_format: format of the template. Defaults to "f-string".
Returns:
a chat prompt template.
"""
return cls(messages, template_format=template_format)
def format_messages(self, **kwargs: Any) -> List[BaseMessage]:
"""Format the chat template into a list of finalized messages.
Args:
**kwargs: keyword arguments to use for filling in template variables
in all the template messages in this chat template.
Returns:
list of formatted messages.
"""
kwargs = self._merge_partial_and_user_variables(**kwargs)
result = []
for message_template in self.messages:
if isinstance(message_template, BaseMessage):
result.extend([message_template])
elif isinstance(
message_template, (BaseMessagePromptTemplate, BaseChatPromptTemplate)
):
message = message_template.format_messages(**kwargs)
result.extend(message)
else:
raise ValueError(f"Unexpected input: {message_template}")
return result
PromptTemplate의 from_template 인스턴스
from langchain_core.prompts import PromptTemplate
prompt = PromptTemplate.from_template("tell me a joke about {topic}")
prompt
// 결과
PromptTemplate(input_variables=['topic'], template='tell me a joke about {topic}')
ChatPromptTemplate의 from_template 인스턴스
- HumanMessagePromptTemplate이 PromptTemplate을 한번 더 감싼 객체
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_template("tell me a joke about {topic}")
prompt
// 결과
ChatPromptTemplate(input_variables=['topic'], messages=[HumanMessagePromptTemplate(prompt=PromptTemplate(input_variables=['topic'], template='tell me a joke about {topic}'))])
Retriever은 VectorStore에서 데이터를 검색하는 인터페이스이다. Retriever는 VectorStore에서 as_retriever() 메서드 호출로 생성한다. 리트리버의 두가지 쿼리 방식에 대해 알아보자.
- Sematic Similarity
- k: relevance 관련성 있는 것을 결과에 반영할 갯수
- Maxium Marginal relavance
- k: relevance 관련성 있는 것을 결과에 반영할 갯수
- fetch_k: diverse 다양성 있는 것을 결과에 반영할 갯수
패키지
Semantic Similarity는 의미적으로 유사한 것을 찾아준다. Reriever를 VectorStore를 통해 얻게 되면 VectoreStore 클래스의 similarity_search() 를 사용한다. 그리고, VectoreStore 클래스를 상속받은 구현체는 similarity_search가 abstractmethod 이기에 구현을 해야 한다. 예로) Chroma는 VectorStore를 상속받아 구현하고 있다.
- 관련성있는 것을 찾기 위하여 k 갯수를 설정한다.
class Chroma(VectorStore):
def similarity_search(
self,
query: str,
k: int = DEFAULT_K,
filter: Optional[Dict[str, str]] = None,
**kwargs: Any,
) -> List[Document]:
"""Run similarity search with Chroma.
Args:
query (str): Query text to search for.
k (int): Number of results to return. Defaults to 4.
filter (Optional[Dict[str, str]]): Filter by metadata. Defaults to None.
Returns:
List[Document]: List of documents most similar to the query text.
"""
docs_and_scores = self.similarity_search_with_score(
query, k, filter=filter, **kwargs
)
return [doc for doc, _ in docs_and_scores]
class Chroma(VectorStore):
def max_marginal_relevance_search(
self,
query: str,
k: int = DEFAULT_K,
fetch_k: int = 20,
lambda_mult: float = 0.5,
filter: Optional[Dict[str, str]] = None,
where_document: Optional[Dict[str, str]] = None,
**kwargs: Any,
) -> List[Document]:
"""Return docs selected using the maximal marginal relevance.
Maximal marginal relevance optimizes for similarity to query AND diversity
among selected documents.
Args:
query: Text to look up documents similar to.
k: Number of Documents to return. Defaults to 4.
fetch_k: Number of Documents to fetch to pass to MMR algorithm.
lambda_mult: Number between 0 and 1 that determines the degree
of diversity among the results with 0 corresponding
to maximum diversity and 1 to minimum diversity.
Defaults to 0.5.
filter (Optional[Dict[str, str]]): Filter by metadata. Defaults to None.
Returns:
List of Documents selected by maximal marginal relevance.
"""
if self._embedding_function is None:
raise ValueError(
"For MMR search, you must specify an embedding function on" "creation."
)
embedding = self._embedding_function.embed_query(query)
docs = self.max_marginal_relevance_search_by_vector(
embedding,
k,
fetch_k,
lambda_mult=lambda_mult,
filter=filter,
where_document=where_document,
)
return docs
VectorStore에서 from_documents 또는 from_texts 메서도에 content와 embedding을 파라미터 설정하고 호출하면 VST 타입의 VectorStore 객체를 반환 받을 수 있다. 이를 통해 similarity_search와 max_marginal_relevance_search 를 사용할 수 있다.
class VectorStore(ABC):
@classmethod
@abstractmethod
def from_texts(
cls: Type[VST],
texts: List[str],
embedding: Embeddings,
metadatas: Optional[List[dict]] = None,
**kwargs: Any,
) -> VST:
"""Return VectorStore initialized from texts and embeddings.
Args:
texts: Texts to add to the vectorstore.
embedding: Embedding function to use.
metadatas: Optional list of metadatas associated with the texts.
Default is None.
**kwargs: Additional keyword arguments.
Returns:
VectorStore: VectorStore initialized from texts and embeddings.
"""
similarity_search와 max_marginal_relevance_search 를 사용예를 보자.
from langchain.vectorstores import Chroma
from langchain.openai import OpenAIEmbeddings
persistence_path = 'db/chroma'
embeddings = OpenAIEmbeddings()
vectorestore = Chroma(persist_directory=persistence_path, embedding_function=embeddings)
texts = [
"""홍길동을 홍길동이라 부르지 못하는 이유는 무엇인가""",
"""심청이가 물에 빠진것을 효라고 말할 수 있는가, 그것이 진심이었나""",
"""춘향이가 이몽룡을 기다른 것은 진심이었나""",
]
smalldb = Chroma.from_texts(texts, embedding=embedding)
question = "진심이라는 의미"
smalldb.similarity_search(question, k=1)
// 결과
[Document(page_content='춘향이가 이몽룡을 기다른 것은 진심이었나', metadata={})]
smalldb.max_marginal_relevance_search(question,k=2, fetch_k=3)
// 결과
[Document(page_content='춘향이가 이몽룡을 기다른 것은 진심이었나', metadata={}),
Document(page_content='심청이가 물에 빠진것을 효라고 말할 수 있는가, 그것이 진심이었나', metadata={})]
위의 두가지외에 다른 검색 타입까지 합쳐 놓은 것이 langchain_core VectorStore의 as_retriever() 이다. from_documents 또는 from_texts를 호출하면 VST (Vector Store Type) 인스턴스를 반환 받아 as_retriever()를 호출한다. as_retriever()는 langchain_community VectorStoreRetriever를 반환하고, 이는 langchain_core의 retrievers.py 파일의 BaseRetriever 클래스를 상속받아 구현하고 있다.
- 0.1.46 이후 get_relevant_documents 사용하지 않고 invoke로 대체한다.
- _get_relevant_documents 를 구현해야 한다.
- search_type
- similarity
- similarity_score_threshold : score_threshold 값 설정 이상만 결과에 반영
- mmr
- 결과값: List[Document]
class VectorStoreRetriever(BaseRetriever):
"""Base Retriever class for VectorStore."""
vectorstore: VectorStore
"""VectorStore to use for retrieval."""
search_type: str = "similarity"
"""Type of search to perform. Defaults to "similarity"."""
search_kwargs: dict = Field(default_factory=dict)
"""Keyword arguments to pass to the search function."""
allowed_search_types: ClassVar[Collection[str]] = (
"similarity",
"similarity_score_threshold",
"mmr",
)
...
class BaseRetriever(RunnableSerializable[RetrieverInput, RetrieverOutput], ABC):
def invoke(
self, input: str, config: Optional[RunnableConfig] = None, **kwargs: Any
) -> List[Document]:
"""Invoke the retriever to get relevant documents.
...
"""
...
// 하위 클래스 생성시 호출이 된다. cls는 하위 클래스이다. get_relevant_documents를 할당함.
def __init_subclass__(cls, **kwargs: Any) -> None:
super().__init_subclass__(**kwargs)
# Version upgrade for old retrievers that implemented the public
# methods directly.
if cls.get_relevant_documents != BaseRetriever.get_relevant_documents:
warnings.warn(
"Retrievers must implement abstract `_get_relevant_documents` method"
" instead of `get_relevant_documents`",
DeprecationWarning,
)
swap = cls.get_relevant_documents
cls.get_relevant_documents = ( # type: ignore[assignment]
BaseRetriever.get_relevant_documents
)
cls._get_relevant_documents = swap # type: ignore[assignment]
if (
hasattr(cls, "aget_relevant_documents")
and cls.aget_relevant_documents != BaseRetriever.aget_relevant_documents
):
warnings.warn(
"Retrievers must implement abstract `_aget_relevant_documents` method"
" instead of `aget_relevant_documents`",
DeprecationWarning,
)
aswap = cls.aget_relevant_documents
cls.aget_relevant_documents = ( # type: ignore[assignment]
BaseRetriever.aget_relevant_documents
)
cls._aget_relevant_documents = aswap # type: ignore[assignment]
parameters = signature(cls._get_relevant_documents).parameters
cls._new_arg_supported = parameters.get("run_manager") is not None
# If a V1 retriever broke the interface and expects additional arguments
cls._expects_other_args = (
len(set(parameters.keys()) - {"self", "query", "run_manager"}) > 0
)
@abstractmethod
def _get_relevant_documents(
self, query: str, *, run_manager: CallbackManagerForRetrieverRun
) -> List[Document]:
"""Get documents relevant to a query.
Args:
query: String to find relevant documents for.
run_manager: The callback handler to use.
Returns:
List of relevant documents.
"""
...
@deprecated(since="0.1.46", alternative="invoke", removal="0.3.0")
def get_relevant_documents(
....
import와 from 키워드는 파이썬에서 모듈과 그 모듈 내의 특정 항목을 가져오는 데 사용됩니다. 각각의 기능과 사용할 수 있는 것들을 정리하면 다음과 같습니다:
import 키워드
import는 전체 모듈을 가져옵니다. 가져온 모듈의 이름을 통해 해당 모듈에 정의된 함수, 클래스, 변수 등을 사용할 수 있습니다.
import module_name
import로 가져올 수 있는 것들:
파이썬 표준 라이브러리 모듈: 예를 들어, import os, import sys, import math 등.
사용자 정의 모듈: 사용자가 만든 .py 파일을 가져올 수 있습니다. 예를 들어, import my_module.
서드 파티 라이브러리 모듈: 설치된 서드 파티 패키지를 가져올 수 있습니다. 예를 들어, import numpy, import pandas.
from 키워드
from은 모듈 내의 특정 항목을 직접 가져올 때 사용됩니다. 이를 통해 모듈 이름 없이도 해당 항목을 직접 사용할 수 있습니다.
from module_name import specific_item
from으로 정의할 수 있는 것들:
모듈 내의 특정 함수: 예를 들어, from math import sqrt는 sqrt 함수를 직접 사용할 수 있게 합니다.
모듈 내의 특정 클래스: 예를 들어, from datetime import datetime은 datetime 클래스를 직접 사용할 수 있게 합니다.
모듈 내의 특정 변수: 예를 들어, from config import config_value는 config_value라는 변수를 가져올 수 있게 합니다.
모듈 내의 모든 항목: from module_name import *를 사용하면 모듈 내의 모든 공개된 항목을 가져올 수 있습니다. 하지만, 이 방식은 권장되지 않습니다.
예시
import math # math 모듈을 가져옴
print(math.sqrt(16)) # math 모듈을 통해 sqrt 함수 호출
from math import sqrt # math 모듈에서 sqrt 함수만 가져옴
print(sqrt(16)) # 모듈 이름 없이 sqrt 함수 호출
from my_module import my_function # my_module 모듈에서 my_function 함수만 가져옴
my_function() # 직접 함수 호출
import와 from 키워드를 적절히 사용하면 코드의 가독성과 효율성을 높일 수 있습니다. written by GPT
