Adding Expertimental Notebook 11.5 with LangGraph Supervisor agent

Author: pablomarin

06-First-RAG.ipynb

@@ -169,7 +169,7 @@ def get_session_history(session_id: str, user_id: str) -> CosmosDBChatMessageHis
 
 chatgpt_search = ChatGPTTool(llm=llm,
                      name="chatgpt",
-                    description="use for general questions, profile, greeting-like questions and when the questions includes the term: chatgpt",
+                    description="useful when the questions includes the term: chatgpt",
                     verbose=False)
 
 tools = [doc_search, book_search, www_search, sql_search, chatgpt_search]

09-BingChatClone.ipynb

@@ -0,0 +1,194 @@
+from langchain.pydantic_v1 import BaseModel, Field
+from sqlalchemy import create_engine, Column, Integer, String, LargeBinary, Table, MetaData, PrimaryKeyConstraint, select
+from sqlalchemy.orm import sessionmaker, Session, scoped_session
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.engine import URL
+from sqlalchemy import Engine
+from typing import Iterator, Optional, Any
+from types import TracebackType
+import pickle
+from contextlib import AbstractContextManager, contextmanager
+from langchain_core.runnables import RunnableConfig
+from typing_extensions import Self
+
+from langgraph.checkpoint.base import (
+    BaseCheckpointSaver,
+    Checkpoint,
+    CheckpointAt,
+    CheckpointTuple,
+    Serializable,
+)
+
+
+metadata = MetaData()
+
+# Adjusting the column type from String (which defaults to VARCHAR(max)) to a specific length
+checkpoints_table = Table(
+    'checkpoints', metadata,
+    Column('thread_id', String(255), primary_key=True),  # String(255) specifies the max length
+    Column('thread_ts', String(255), primary_key=True),
+    Column('parent_ts', String(255)),  # Optional: Specify length here if it's a commonly used field
+    Column('checkpoint', LargeBinary),  # VARBINARY(max) is fine for non-indexed columns
+    PrimaryKeyConstraint('thread_id', 'thread_ts')
+)
+
+class BaseCheckpointSaver(BaseModel):
+    
+    engine: Optional[Engine] = None
+    Session: Optional[scoped_session] = None
+    is_setup: bool = Field(default=False)
+    session: Any = Field(default=None) 
+
+    class Config:
+        arbitrary_types_allowed = True
+
+class SQLAlchemyCheckpointSaver(BaseCheckpointSaver, AbstractContextManager):
+    
+    def __init__(self, engine: Engine, *, serde: Optional[Serializable] = None, at: Optional[CheckpointAt] = None):
+        # Call super with all expected fields by Pydantic
+        super().__init__(serde=serde or pickle, at=at or CheckpointAt.END_OF_STEP, is_setup=False)
+        self.engine = engine
+        self.Session = scoped_session(sessionmaker(bind=self.engine))
+
+
+    @classmethod
+    def from_db_config(cls, db_config):
+        db_url = URL.create(
+            drivername=db_config['drivername'],
+            username=db_config['username'],
+            password=db_config['password'],
+            host=db_config['host'],
+            port=db_config['port'],
+            database=db_config['database'],
+            query=db_config['query']
+        )
+        engine = create_engine(db_url)
+        return cls(engine)
+    
+    def __enter__(self):
+        self.session = self.Session()
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        try:
+            if exc_type:
+                self.session.rollback()
+            else:
+                self.session.commit()
+        finally:
+            self.Session.remove()
+            self.session.close()
+
+        
+    def setup(self):
+        if not self.is_setup:
+            # Create all tables if they don't exist
+            metadata.create_all(self.engine)
+            self.is_setup = True
+
+    def get(self, config: RunnableConfig) -> Optional[Checkpoint]:
+        if value := self.get_tuple(config):
+            return value['checkpoint']
+
+    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
+        print("SQLAlchemyCheckpointSaver.get_tuple properly called")
+        with self.Session() as session:
+            thread_id = config["configurable"].get("thread_id")
+            thread_ts = config["configurable"].get("thread_ts")
+
+            query = select(checkpoints_table)
+            if thread_ts:
+                query = query.where(
+                    (checkpoints_table.c.thread_id == thread_id) &
+                    (checkpoints_table.c.thread_ts == thread_ts)
+                )
+            else:
+                query = query.where(
+                    checkpoints_table.c.thread_id == thread_id
+                ).order_by(checkpoints_table.c.thread_ts.desc()).limit(1)
+
+            result = session.execute(query).fetchone()
+            if result:
+                # Handling both potential types of result objects
+                if isinstance(result, tuple):
+                    # Convert tuple to dictionary using column keys if result is a tuple
+                    result = dict(zip(result.keys(), result))
+                elif hasattr(result, '_mapping'):
+                    # Convert SQLAlchemy RowProxy to dictionary directly if available
+                    result = dict(result._mapping)
+
+                return {
+                    'config': config,
+                    'checkpoint': pickle.loads(result['checkpoint']),
+                    'additional_info': {
+                        "thread_id": result['thread_id'],
+                        "thread_ts": result['parent_ts'] if result['parent_ts'] else None
+                    }
+                }
+            return None
+
+
+
+    def list(self, config: RunnableConfig):
+        with self.Session() as session:
+            query = select(checkpoints_table).where(
+                checkpoints_table.c.thread_id == config["configurable"]["thread_id"]
+            ).order_by(checkpoints_table.c.thread_ts.desc())
+            results = session.execute(query).fetchall()
+
+            return [
+                {
+                    "configurable": {
+                        "thread_id": result['thread_id'],
+                        "thread_ts": result['thread_ts']
+                    },
+                    "checkpoint": pickle.loads(result['checkpoint']),
+                    "additional_info": {
+                        "thread_id": result['thread_id'],
+                        "thread_ts": result['parent_ts'] or None
+                    }
+                }
+                for result in results
+            ]
+
+
+    def put(self, config: RunnableConfig, checkpoint: Checkpoint):
+        print("Attempting to connect with engine:", self.engine.url)  # Check the engine URL
+        with self.Session() as session:
+            print("Session started for put operation.")
+            try:
+                session.execute(
+                    checkpoints_table.insert().values(
+                        thread_id=config["configurable"]["thread_id"],
+                        thread_ts=checkpoint["ts"],
+                        parent_ts=config["configurable"].get("thread_ts"),
+                        checkpoint=pickle.dumps(checkpoint)
+                    )
+                )
+                session.commit()
+                print("Data inserted and committed successfully.")
+            except Exception as e:
+                print("Error during database operation:", e)
+                session.rollback()
+                raise
+            finally:
+                print("Session closed after put operation.")
+        return {
+            "configurable": {
+                "thread_id": config["configurable"]["thread_id"],
+                "thread_ts": checkpoint["ts"]
+            }
+        }
+
+    async def aget(self, config: RunnableConfig) -> Optional[Checkpoint]:
+        return await asyncio.get_running_loop().run_in_executor(None, self.get, config)
+
+    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
+        return await asyncio.get_running_loop().run_in_executor(None, self.get_tuple, config)
+
+    async def alist(self, config: RunnableConfig) -> Iterator[CheckpointTuple]:
+        return await asyncio.get_running_loop().run_in_executor(None, self.list, config)
+
+    async def aput(self, config: RunnableConfig, checkpoint: Checkpoint) -> RunnableConfig:
+        return await asyncio.get_running_loop().run_in_executor(None, self.put, config, checkpoint)
+

11-Smart_Agent.ipynb

@@ -363,6 +363,10 @@ def get_answer(llm: AzureChatOpenAI,
 
 class SearchInput(BaseModel):
     query: str = Field(description="should be a search query")
+    return_direct: bool = Field(
+        description="Whether or the result of this should be returned directly to the user without you seeing what it is",
+        default=False,
+    )
 
 class GetDocSearchResults_Tool(BaseTool):
     name = "docsearch"
@@ -375,7 +379,7 @@ class GetDocSearchResults_Tool(BaseTool):
     sas_token: str = "" 
 
     def _run(
-        self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None
+        self, query: str,  return_direct = False, run_manager: Optional[CallbackManagerForToolRun] = None
     ) -> str:
 
         retriever = CustomAzureSearchRetriever(indexes=self.indexes, topK=self.k, reranker_threshold=self.reranker_th, 
@@ -385,7 +389,7 @@ def _run(
         return results
 
     async def _arun(
-        self, query: str, run_manager: Optional[AsyncCallbackManagerForToolRun] = None
+        self, query: str, return_direct = False, run_manager: Optional[AsyncCallbackManagerForToolRun] = None
     ) -> str:
         """Use the tool asynchronously."""
 
@@ -424,15 +428,15 @@ def __init__(self, **data):
         self.agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=self.verbose, callback_manager=self.callbacks, handle_parsing_errors=True)
 
 
-    def _run(self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
+    def _run(self, query: str,  return_direct = False, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
         try:
             result = self.agent_executor.invoke({"question": query})
             return result['output']
         except Exception as e:
             print(e)
             return str(e)  # Return an empty string or some error indicator
 
-    async def _arun(self, query: str, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
+    async def _arun(self, query: str,  return_direct = False, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
         try:
             result = await self.agent_executor.ainvoke({"question": query})
             return result['output']
@@ -465,7 +469,7 @@ def __init__(self, **data):
                                                callback_manager=self.callbacks,
                                                )
 
-    def _run(self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
+    def _run(self, query: str, return_direct = False, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
         try:
             # Use the initialized agent_executor to invoke the query
             result = self.agent_executor.invoke(query)
@@ -474,7 +478,7 @@ def _run(self, query: str, run_manager: Optional[CallbackManagerForToolRun] = No
             print(e)
             return str(e)  # Return an error indicator
 
-    async def _arun(self, query: str, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
+    async def _arun(self, query: str, return_direct = False, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
         # Note: Implementation assumes the agent_executor and its methods support async operations
         try:
             # Use the initialized agent_executor to asynchronously invoke the query
@@ -528,7 +532,7 @@ def get_db_config(self):
             'query': {'driver': 'ODBC Driver 17 for SQL Server'}
         }
 
-    def _run(self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
+    def _run(self, query: str, return_direct = False, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
         try:
             # Use the initialized agent_executor to invoke the query
             result = self.agent_executor.invoke(query)
@@ -537,7 +541,7 @@ def _run(self, query: str, run_manager: Optional[CallbackManagerForToolRun] = No
             print(e)
             return str(e)  # Return an error indicator
 
-    async def _arun(self, query: str, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
+    async def _arun(self, query: str, return_direct = False, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
         # Note: Implementation assumes the agent_executor and its methods support async operations
         try:
             # Use the initialized agent_executor to asynchronously invoke the query
@@ -567,15 +571,15 @@ def __init__(self, **data):
         output_parser = StrOutputParser()
         self.chatgpt_chain = CHATGPT_PROMPT | self.llm | output_parser
 
-    def _run(self, query: str) -> str:
+    def _run(self, query: str, return_direct = False, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
         try:
             response = self.chatgpt_chain.invoke({"question": query})
             return response
         except Exception as e:
             print(e)
             return str(e)  # Return an error indicator
 
-    async def _arun(self, query: str, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
+    async def _arun(self, query: str, return_direct = False, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
         """Implement the tool to be used asynchronously."""
         try:
             response = await self.chatgpt_chain.ainvoke({"question": query})
@@ -595,14 +599,14 @@ class GetBingSearchResults_Tool(BaseTool):
 
     k: int = 5
 
-    def _run(self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
+    def _run(self, query: str,  return_direct = False, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
         bing = BingSearchAPIWrapper(k=self.k)
         try:
             return bing.results(query,num_results=self.k)
         except:
             return "No Results Found"
 
-    async def _arun(self, query: str, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
+    async def _arun(self, query: str, return_direct = False, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
         bing = BingSearchAPIWrapper(k=self.k)
         loop = asyncio.get_event_loop()
         try:
@@ -635,7 +639,9 @@ def __init__(self, **data):
             description="useful to fetch the content of a url"
         )
 
-        tools = [GetBingSearchResults_Tool(k=self.k), web_fetch_tool]
+        tools = [GetBingSearchResults_Tool(k=self.k)]
+        # tools = [GetBingSearchResults_Tool(k=self.k), web_fetch_tool] # Uncomment if using GPT-4
+        
         agent = create_openai_tools_agent(self.llm, tools, BINGSEARCH_PROMPT)
 
         self.agent_executor = AgentExecutor(agent=agent, tools=tools,
@@ -656,15 +662,15 @@ def fetch_web_page(self, url: str) -> str:
         response = requests.get(url, headers=HEADERS)
         return self.parse_html(response.content)
 
-    def _run(self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
+    def _run(self, query: str,  return_direct = False, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
         try:
             response = self.agent_executor.invoke({"question": query})
             return response['output']
         except Exception as e:
             print(e)
             return str(e)  # Return an error indicator
 
-    async def _arun(self, query: str, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
+    async def _arun(self, query: str, return_direct = False, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
         """Implements the tool to be used asynchronously."""
         try:
             response = await self.agent_executor.ainvoke({"question": query})
@@ -701,7 +707,7 @@ def __init__(self, **data):
             limit_to_domains=self.limit_to_domains
         )
 
-    def _run(self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
+    def _run(self, query: str, return_direct = False, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
         try:
             # Optionally sleep to avoid possible TPM rate limits
             sleep(2)
@@ -711,7 +717,7 @@ def _run(self, query: str, run_manager: Optional[CallbackManagerForToolRun] = No
 
         return response
 
-    async def _arun(self, query: str, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
+    async def _arun(self, query: str, return_direct = False, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
         """Use the tool asynchronously."""
         loop = asyncio.get_event_loop()
         try:
@@ -757,7 +763,7 @@ def __init__(self, **data):
                                             return_intermediate_steps=True,
                                             callback_manager=self.callbacks)
 
-    def _run(self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
+    def _run(self, query: str, return_direct = False, run_manager: Optional[CallbackManagerForToolRun] = None) -> str:
         try:
             # Use the initialized agent_executor to invoke the query
             response = self.agent_executor.invoke({"question":query})
@@ -766,7 +772,7 @@ def _run(self, query: str, run_manager: Optional[CallbackManagerForToolRun] = No
             print(e)
             return str(e)  # Return an error indicator
 
-    async def _arun(self, query: str, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
+    async def _arun(self, query: str, return_direct = False, run_manager: Optional[AsyncCallbackManagerForToolRun] = None) -> str:
         # Note: Implementation assumes the agent_executor and its methods support async operations
         try:
             # Use the initialized agent_executor to asynchronously invoke the query

11.5-Smart_Agent-LangGraph.ipynb

@@ -31,4 +31,5 @@ AZURE_COSMOSDB_NAME="ENTER YOUR VALUE"
 AZURE_COSMOSDB_CONTAINER_NAME="ENTER YOUR VALUE"
 AZURE_COMOSDB_CONNECTION_STRING="ENTER YOUR VALUE" # Find this in the Keys section
 BOT_ID="ENTER YOUR VALUE" # This is the name of your bot service created in Notebook 12
-BOT_SERVICE_DIRECT_LINE_SECRET="ENTER YOUR VALUE"  # Find this in Azure Bot Service -> Channels -> Direct Line
+BOT_SERVICE_DIRECT_LINE_SECRET="ENTER YOUR VALUE"  # Find this in Azure Bot Service -> Channels -> Direct Line
+