ashen007

from typing import Any
from langchain_core.messages import ToolMessage
from langgraph.prebuilt import ToolNode
from langchain_core.runnables import RunnableLambda, RunnableWithFallbacks

def handle_tool_error(state) -> dict:
    error = state.get('error')
    tool_calls = state['messages'][-1].tool_calls

    return {

ashen007

from utils.db_tools import get_engine
from langchain_community.utilities import SQLDatabase

def sql_database():
    return SQLDatabase(get_engine())

def db_query_tool(query: str) -> str:
    """
          execute SQL query.
        """

ashen007

from langchain_ollama import ChatOllama

def init_llm():
    # here you can use any of supported chat models by
    # langchain ex: ChatOpenAI, ...
    return ChatOllama(model='llama3.1:latest')

ashen007

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import AnyMessage, add_messages

class State(TypedDict):
    answer: list[dict]
    messages: Annotated[list[AnyMessage], add_messages]

ashen007

def query_gen_node(state: State):
    llm = init_llm()
    query_gen_chain = query_generation | llm.bind_tools([db_query_tool], tool_choice='db_query_tool')
    message = query_gen_chain.invoke(state)

    return {
        'messages': [message]
    }

ashen007

2023-02-27 13:43:37 - [DEBUG] __main__ [main.main:22]: start execution
2023-02-27 13:43:37 - [INFO] __main__ [main.main:24]: main is calling do_something from module_x
2023-02-27 13:43:37 - [DEBUG] package.module_x.lib_x [lib_x.do_something:19]: do somthing executed in module_x-lib.
2023-02-27 13:43:37 - [ERROR] package.module_x.lib_x [lib_x.do_something:21]: this is a random error massage from lib_x.
2023-02-27 13:43:37 - [INFO] __main__ [main.main:27]: main is calling do_something from module_y
2023-02-27 13:43:37 - [DEBUG] package.module_y.lib_y [lib_y.do_something:19]: do somthing executed in module_y-lib.
2023-02-27 13:43:37 - [DEBUG] __main__ [main.main:30]: end execution
2023-02-27 13:43:37 - [ERROR] __main__ [main.main:32]: this is a random error massage

ashen007

{
  "version": 1,
  "disable_existing_loggers": false,
  "formatters": {
    "brief": {
      "format": "%(asctime)s - [%(levelname)s]: %(message)s"
    },
    "default": {
      "format": "%(asctime)s - [%(levelname)s] %(name)s [%(module)s.%(funcName)s:%(lineno)d]: %(message)s",
      "datefmt": "%Y-%m-%d %H:%M:%S"

ashen007

import logging

if __name__ == "__main__":
    # modify the log massages to show tracking level and
    # date and time of the log entry creation
    logging.basicConfig(format='%(asctime)s %(levelname)s:%(message)s',
                        level=logging.DEBUG)

    logging.debug("start execution")
    logging.info("main is calling do_something from module_x")

ashen007

# config file configure
fileConfig('package/.config',
           defaults={'logfilename': '.log'})

# get the named logger object
logger = logging.getLogger(__name__)

ashen007

[loggers]
keys=root

[handlers]
keys=fileHandler

[formatters]
keys=defaultFormatter

[logger_root]