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): runの名前。
    notes (str):記録先のメモ。
    visualize (bool): runを可視化するかどうか。
    complexity_metrics (bool): 複雑度メトリクスを記録するかどうか.
    stream_logs (bool):コールバックアクションを W&Bにストリームするかどうか
"""

# 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}
}