Bold Crumbs (#4876)

docker
Zander Chase 1 year ago committed by GitHub
parent 4c3ab55e94
commit 1ff7c958b0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

@ -3,7 +3,7 @@ from typing import Any, List
from langchain.callbacks.tracers.base import BaseTracer from langchain.callbacks.tracers.base import BaseTracer
from langchain.callbacks.tracers.schemas import Run from langchain.callbacks.tracers.schemas import Run
from langchain.input import get_colored_text from langchain.input import get_bolded_text, get_colored_text
def try_json_stringify(obj: Any, fallback: str) -> str: def try_json_stringify(obj: Any, fallback: str) -> str:
@ -54,24 +54,28 @@ class ConsoleCallbackHandler(BaseTracer):
crumbs = self.get_breadcrumbs(run) crumbs = self.get_breadcrumbs(run)
print( print(
f"{get_colored_text('[chain/start]', color='green')} " f"{get_colored_text('[chain/start]', color='green')} "
f"[{crumbs}] Entering Chain run with input:\n" + get_bolded_text(f"[{crumbs}] Entering Chain run with input:\n")
f"{try_json_stringify(run.inputs, '[inputs]')}" + f"{try_json_stringify(run.inputs, '[inputs]')}"
) )
def _on_chain_end(self, run: Run) -> None: def _on_chain_end(self, run: Run) -> None:
crumbs = self.get_breadcrumbs(run) crumbs = self.get_breadcrumbs(run)
print( print(
f"{get_colored_text('[chain/end]', color='blue')} " f"{get_colored_text('[chain/end]', color='blue')} "
f"[{crumbs}] [{elapsed(run)}] Exiting Chain run with output:\n" + get_bolded_text(
f"{try_json_stringify(run.outputs, '[outputs]')}" f"[{crumbs}] [{elapsed(run)}] Exiting Chain run with output:\n"
)
+ f"{try_json_stringify(run.outputs, '[outputs]')}"
) )
def _on_chain_error(self, run: Run) -> None: def _on_chain_error(self, run: Run) -> None:
crumbs = self.get_breadcrumbs(run) crumbs = self.get_breadcrumbs(run)
print( print(
f"{get_colored_text('[chain/error]', color='red')} " f"{get_colored_text('[chain/error]', color='red')} "
f"[{crumbs}] [{elapsed(run)}] Chain run errored with error:\n" + get_bolded_text(
f"{try_json_stringify(run.error, '[error]')}" f"[{crumbs}] [{elapsed(run)}] Chain run errored with error:\n"
)
+ f"{try_json_stringify(run.error, '[error]')}"
) )
def _on_llm_start(self, run: Run) -> None: def _on_llm_start(self, run: Run) -> None:
@ -83,32 +87,36 @@ class ConsoleCallbackHandler(BaseTracer):
) )
print( print(
f"{get_colored_text('[llm/start]', color='green')} " f"{get_colored_text('[llm/start]', color='green')} "
f"[{crumbs}] Entering LLM run with input:\n" + get_bolded_text(f"[{crumbs}] Entering LLM run with input:\n")
f"{try_json_stringify(inputs, '[inputs]')}" + f"{try_json_stringify(inputs, '[inputs]')}"
) )
def _on_llm_end(self, run: Run) -> None: def _on_llm_end(self, run: Run) -> None:
crumbs = self.get_breadcrumbs(run) crumbs = self.get_breadcrumbs(run)
print( print(
f"{get_colored_text('[llm/end]', color='blue')} " f"{get_colored_text('[llm/end]', color='blue')} "
f"[{crumbs}] [{elapsed(run)}] Exiting LLM run with output:\n" + get_bolded_text(
f"{try_json_stringify(run.outputs, '[response]')}" f"[{crumbs}] [{elapsed(run)}] Exiting LLM run with output:\n"
)
+ f"{try_json_stringify(run.outputs, '[response]')}"
) )
def _on_llm_error(self, run: Run) -> None: def _on_llm_error(self, run: Run) -> None:
crumbs = self.get_breadcrumbs(run) crumbs = self.get_breadcrumbs(run)
print( print(
f"{get_colored_text('[llm/error]', color='red')} " f"{get_colored_text('[llm/error]', color='red')} "
f"[{crumbs}] [{elapsed(run)}] LLM run errored with error:\n" + get_bolded_text(
f"{try_json_stringify(run.error, '[error]')}" f"[{crumbs}] [{elapsed(run)}] LLM run errored with error:\n"
)
+ f"{try_json_stringify(run.error, '[error]')}"
) )
def _on_tool_start(self, run: Run) -> None: def _on_tool_start(self, run: Run) -> None:
crumbs = self.get_breadcrumbs(run) crumbs = self.get_breadcrumbs(run)
print( print(
f'{get_colored_text("[tool/start]", color="green")} ' f'{get_colored_text("[tool/start]", color="green")} '
f"[{crumbs}] Entering Tool run with input:\n" + get_bolded_text(f"[{crumbs}] Entering Tool run with input:\n")
f'"{run.inputs["input"].strip()}"' + f'"{run.inputs["input"].strip()}"'
) )
def _on_tool_end(self, run: Run) -> None: def _on_tool_end(self, run: Run) -> None:
@ -116,15 +124,17 @@ class ConsoleCallbackHandler(BaseTracer):
if run.outputs: if run.outputs:
print( print(
f'{get_colored_text("[tool/end]", color="blue")} ' f'{get_colored_text("[tool/end]", color="blue")} '
f"[{crumbs}] [{elapsed(run)}] Exiting Tool run with output:\n" + get_bolded_text(
f'"{run.outputs["output"].strip()}"' f"[{crumbs}] [{elapsed(run)}] Exiting Tool run with output:\n"
)
+ f'"{run.outputs["output"].strip()}"'
) )
def _on_tool_error(self, run: Run) -> None: def _on_tool_error(self, run: Run) -> None:
crumbs = self.get_breadcrumbs(run) crumbs = self.get_breadcrumbs(run)
print( print(
f"{get_colored_text('[tool/error]', color='red')} " f"{get_colored_text('[tool/error]', color='red')} "
f"[{crumbs}] [{elapsed(run)}] " + get_bolded_text(f"[{crumbs}] [{elapsed(run)}] ")
f"Tool run errored with error:\n" + f"Tool run errored with error:\n"
f"{run.error}" f"{run.error}"
) )

@ -27,6 +27,11 @@ def get_colored_text(text: str, color: str) -> str:
return f"\u001b[{color_str}m\033[1;3m{text}\u001b[0m" return f"\u001b[{color_str}m\033[1;3m{text}\u001b[0m"
def get_bolded_text(text: str) -> str:
"""Get bolded text."""
return f"\033[1m{text}\033[0m"
def print_text(text: str, color: Optional[str] = None, end: str = "") -> None: def print_text(text: str, color: Optional[str] = None, end: str = "") -> None:
"""Print text with highlighting and no end characters.""" """Print text with highlighting and no end characters."""
if color is None: if color is None:

Loading…
Cancel
Save