forked from Archives/langchain
Compare commits
1 Commits
main
...
harrison/l
Author | SHA1 | Date |
---|---|---|
Harrison Chase | 844151605c | 1 year ago |
@ -0,0 +1,70 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Optional, Any
|
||||
from langchain.printing import print_text
|
||||
from pathlib import Path
|
||||
|
||||
CONTEXT_KEY = "__context__"
|
||||
|
||||
|
||||
class Logger(ABC):
|
||||
|
||||
@abstractmethod
|
||||
def log_start_of_chain(self, inputs):
|
||||
""""""
|
||||
|
||||
@abstractmethod
|
||||
def log_end_of_chain(self, outputs):
|
||||
""""""
|
||||
|
||||
@abstractmethod
|
||||
def log(self, text: str, context: dict, **kwargs):
|
||||
""""""
|
||||
|
||||
|
||||
class PrintLogger(Logger):
|
||||
def log_start_of_chain(self, inputs):
|
||||
""""""
|
||||
print("\n\n\033[1m> Entering new chain...\033[0m")
|
||||
|
||||
def log_end_of_chain(self, outputs):
|
||||
""""""
|
||||
print("\n\033[1m> Finished chain.\033[0m")
|
||||
|
||||
def log(self, text: str, context: dict, title: Optional[str ] =None ,**kwargs:Any):
|
||||
""""""
|
||||
if title is not None:
|
||||
print(title)
|
||||
print_text(text, **kwargs)
|
||||
|
||||
import json
|
||||
class JSONLogger(Logger):
|
||||
|
||||
def __init__(self, log_dir):
|
||||
self.log_dir = Path(log_dir)
|
||||
self.log_dir.mkdir(exist_ok=True)
|
||||
|
||||
def log_start_of_chain(self, inputs):
|
||||
""""""
|
||||
fname = self.log_dir / f"{inputs[CONTEXT_KEY]['id']}.json"
|
||||
if not fname.exists():
|
||||
with open(fname, 'w') as f:
|
||||
json.dump([], f)
|
||||
|
||||
def log_end_of_chain(self, outputs):
|
||||
""""""
|
||||
fname = self.log_dir / f"{outputs[CONTEXT_KEY]['id']}.json"
|
||||
with open(fname) as f:
|
||||
logs = json.load(f)
|
||||
logs.append(outputs)
|
||||
with open(fname, 'w') as f:
|
||||
json.dump(logs, f)
|
||||
|
||||
def log(self, text: str, context: dict, title: Optional[str ] =None ,**kwargs:Any):
|
||||
""""""
|
||||
fname = self.log_dir / f"{context['id']}.json"
|
||||
with open(fname) as f:
|
||||
logs = json.load(f)
|
||||
logs.append({"text": text, "title": title})
|
||||
with open(fname, 'w') as f:
|
||||
json.dump(logs, f)
|
||||
|
@ -0,0 +1,29 @@
|
||||
from typing import Optional, List, Dict
|
||||
|
||||
|
||||
def print_text(text: str, color: Optional[str] = None, end: str = "") -> None:
|
||||
"""Print text with highlighting and no end characters."""
|
||||
if color is None:
|
||||
print(text, end=end)
|
||||
else:
|
||||
color_str = _TEXT_COLOR_MAPPING[color]
|
||||
print(f"\u001b[{color_str}m\033[1;3m{text}\u001b[0m", end=end)
|
||||
|
||||
|
||||
def get_color_mapping(
|
||||
items: List[str], excluded_colors: Optional[List] = None
|
||||
) -> Dict[str, str]:
|
||||
"""Get mapping for items to a support color."""
|
||||
colors = list(_TEXT_COLOR_MAPPING.keys())
|
||||
if excluded_colors is not None:
|
||||
colors = [c for c in colors if c not in excluded_colors]
|
||||
color_mapping = {item: colors[i % len(colors)] for i, item in enumerate(items)}
|
||||
return color_mapping
|
||||
|
||||
|
||||
_TEXT_COLOR_MAPPING = {
|
||||
"blue": "36;1",
|
||||
"yellow": "33;1",
|
||||
"pink": "38;5;200",
|
||||
"green": "32;1",
|
||||
}
|
Loading…
Reference in New Issue