- 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 '
