LangChain과 W&B를 이용한 프롬프트 엔지니어링 LLM

class BaseCallbackHandler(ABC):
    """langchain으로부터의 콜백을 처리하는데 사용할 수 있는 베이스 콜백 핸들러."""

    # (...) 일부 함수 누락됨

    @abstractmethod
    def on_llm_start(
        self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
    ) -> Any:
        """LLM이 실행 시작할 때 실행."""

    @abstractmethod
    def on_llm_new_token(self, token: str, **kwargs: Any) -> Any:
        """새로운 LLM 토큰에 대해 실행. 스트리밍이 활성화될 때에만 이용 가능."""

    @abstractmethod
    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> Any:
        """LLM 실행이 종료될 때 실행."""

    @abstractmethod
    def on_llm_error(
        self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
    ) -> Any:
        """LLM에 오류가 있을 때 실행."""

    @abstractmethod
    def on_chain_start(
        self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any
    ) -> Any:
        """체인이 실행 시작할 때 실행."""

    @abstractmethod
    def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> Any:
        """체인이 실행 종료될 때 실행."""

    @abstractmethod
    def on_chain_error(
        self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
    ) -> Any:
        """체인에 오류가 있을 때 실행."""

    @abstractmethod
    def on_tool_start(
        self, serialized: Dict[str, Any], input_str: str, **kwargs: Any
    ) -> Any:
        """도구가 실행 시작할 때 실행."""

    @abstractmethod
    def on_tool_end(self, output: str, **kwargs: Any) -> Any:
        """도구가 실행 종료될 때 실행."""

    @abstractmethod
    def on_tool_error(
        self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
    ) -> Any:
        """도구에 오류가 있을 때 실행."""

    @abstractmethod
    def on_text(self, text: str, **kwargs: Any) -> Any:
        """임의의 텍스트에 ���해 실행."""

    @abstractmethod
    def on_agent_action(self, action: AgentAction, **kwargs: Any) -> Any:
        """에이전트 동작 시 실행."""

    @abstractmethod
    def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) -> Any:
        """에이전트 종료 시 실행."""

def on_llm_start(
    self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
) -> None:
    """LLM이 시작할 때 실행."""
    self.step += 1
    self.llm_starts += 1
    self.starts += 1

    resp = self._init_resp()
    resp.update({"action": "on_llm_start"})
    resp.update(flatten_dict(serialized))
    resp.update(self.get_custom_callback_meta())

    for prompt in prompts:
        prompt_resp = deepcopy(resp)
        prompt_resp["prompts"] = prompt
        self.on_llm_start_records.append(prompt_resp)
        self.action_records.append(prompt_resp)
        if self.stream_logs:
            self.run.log(prompt_resp)

def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
    """LLM 실행이 종료될 때 실행."""
    self.step += 1
    self.llm_ends += 1
    self.ends += 1

    resp = self._init_resp()
    resp.update({"action": "on_llm_end"})
    resp.update(flatten_dict(response.llm_output))
    resp.update(self.get_custom_callback_meta())

    for generations in response.generations:
        for generation in generations:
            generation_resp = deepcopy(resp)
            generation_resp.update(flatten_dict(generation.to_dict()))
            generation_resp.update(
                analyze_text(
                    generation.text,
                    complexity_metrics=self.complexity_metrics,
                    visualize=self.visualize,
                    nlp=self.nlp,
                    output_dir=self.temp_dir.name,
                )
            )
            self.on_llm_end_records.append(generation_resp)
            self.action_records.append(generation_resp)
            if self.stream_logs:
                self.run.log(generation_resp)

class WandbCallbackHandler(BaseMetadataCallbackHandler, BaseCallbackHandler):
"""Weights and Biases에 로그되는 콜백 핸들러.

파라미터:
    job_type (str): 작업의 유형.
    project (str): 기록할 프로젝트.
    entity (str): 기록할 엔티티.
    tags (list): 기록할 태그.
    group (str): 기록할 그룹.
    name (str): 실행의 이름.
    notes (str): 기록할 메모.
    visualize (bool): 실행을 시각화할 지 여부.
    complexity_metrics (bool): 복잡성 척도를 기록할 지 여부.
    stream_logs (bool): 콜백 작업을 W&B에 스트림할 지 여부
"""

# Default values for WandbCallbackHandler(...)
visualize: bool = False,
complexity_metrics: bool = False,
stream_logs: bool = False,

from datetime import datetime
from langchain.callbacks import WandbCallbackHandler, StdOutCallbackHandler
from langchain.callbacks.base import CallbackManager
from langchain.llms import OpenAI

session_group = datetime.now().strftime("%m.%d.%Y_%H.%M.%S")
wandb_callback = WandbCallbackHandler(
    job_type="inference",
    project="langchain_callback_demo2",
    group=f"minimal_{session_group}",
    name="llm",
    tags=["test"],
)
manager = CallbackManager([StdOutCallbackHandler(), wandb_callback])
llm = OpenAI(temperature=0, callback_manager=manager, verbose=True)

# Defaults for WandbCallbackHandler.flush_tracker(...)
reset: bool = True,
finish: bool = False,

# 시나리오 1 - LLM
llm_result = llm.generate(["Tell me a joke", "Tell me a poem"] * 3)
wandb_callback.flush_tracker(llm, name="simple_sequential")

from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain, SimpleSequentialChain

# 시나리오 2 - 사슬
template = """You are a playwright. Given the title of play, it is your job to write a synopsis for that title.
Title: {title}
Playwright: This is a synopsis for the above play:"""
prompt_template = PromptTemplate(input_variables=["title"], template=template)
synopsis_chain = LLMChain(llm=llm, prompt=prompt_template, callback_manager=manager)

template = """You are a play critic from the New York Times. Given the synopsis of play, it is your job to write a review for that play.
Play Synopsis:
{synopsis}
Review from a New York Times play critic of the above play:"""
prompt_template = PromptTemplate(input_variables=["synopsis"], template=template)
review_chain = LLMChain(llm=llm, prompt=prompt_template, callback_manager=manager)

overall_chain = SimpleSequentialChain(
    chains=[synopsis_chain, review_chain], verbose=True, callback_manager=manager
)

test_prompts = [
    {
        "input": "documentary about good video games that push the boundary of game design"
    },
    {"input": "cocaine bear vs heroin wolf"},
    {"input": "the best in class mlops tooling"},
]
overall_chain.apply(test_prompts)
wandb_callback.flush_tracker(overall_chain, name="agent")

from langchain.agents import initialize_agent, load_tools

# 시나리오 3 - 도구가 있는 에이전트
tools = load_tools(["serpapi", "llm-math"], llm=llm, callback_manager=manager)
agent = initialize_agent(
    tools,
    llm,
    agent="zero-shot-react-description",
    callback_manager=manager,
    verbose=True,
)
agent.run(
    "Who is Leo DiCaprio's girlfriend? What is her current age raised to the 0.43 power?"
)
wandb_callback.flush_tracker(agent, reset=False, finish=True)

@article{Saravia_Prompt_Engineering_Guide_2022,
author = {Saravia, Elvis},
journal = {https://github.com/dair-ai/Prompt-Engineering-Guide},
month = {12},
title = {{Prompt Engineering Guide}},
year = {2022}
}