Classification langchain example

examples/chat_history/memory.py

@@ -14,9 +14,9 @@
 import time
 import uuid
 
+from chat_history import history
 from dotenv import find_dotenv, load_dotenv
 
-from chat_history import history
 from zep_python import (
     APIError,
     NotFoundError,

examples/chat_history/memory_async.py

@@ -15,9 +15,9 @@
 import time
 import uuid
 
+from chat_history import history
 from dotenv import find_dotenv, load_dotenv
 
-from chat_history import history
 from zep_python import (
     APIError,
     NotFoundError,

examples/langchain-langserve/app/classification_chain.py

@@ -0,0 +1,124 @@
+import os
+
+from langchain.callbacks.tracers import ConsoleCallbackHandler
+from langchain_community.chat_models import ChatAnthropic
+from langchain_core.output_parsers import StrOutputParser
+from langchain_core.prompts.prompt import PromptTemplate
+from langchain_core.pydantic_v1 import BaseModel
+from langchain_core.runnables import RunnableBranch, RunnableLambda, RunnablePassthrough
+from langchain_core.runnables.history import RunnableWithMessageHistory
+
+from zep_python import ZepClient
+from zep_python.langchain import ZepChatMessageHistory
+
+ZEP_API_KEY = os.environ.get("ZEP_API_KEY")  # Required for Zep Cloud
+ZEP_API_URL = os.environ.get(
+    "ZEP_API_URL"
+)  # only required if you're using Zep Open Source
+
+if ZEP_API_KEY is None:
+    raise ValueError(
+        "ZEP_API_KEY is required for Zep Cloud. "
+        "Remove this check if using Zep Open Source."
+    )
+
+zep = ZepClient(
+    api_key=ZEP_API_KEY,
+    api_url=ZEP_API_URL,  # only required if you're using Zep Open Source
+)
+
+langchain_chain = (
+    PromptTemplate.from_template(
+        """You are an expert in langchain. \
+    Always answer questions starting with "As Harrison Chase told me". \
+    Respond to the following question:
+
+    Question: {question}
+    Answer:"""
+    )
+    | ChatAnthropic()
+)
+anthropic_chain = (
+    PromptTemplate.from_template(
+        """You are an expert in anthropic. \
+    Always answer questions starting with "As Dario Amodei told me". \
+    Respond to the following question:
+
+    Question: {question}
+    Answer:"""
+    )
+    | ChatAnthropic()
+)
+general_chain = (
+    PromptTemplate.from_template(
+        """Respond to the following question:
+
+    Question: {question}
+    Answer:"""
+    )
+    | ChatAnthropic()
+)
+
+branch = RunnableBranch(
+    (lambda x: "anthropic" in x["topic"].lower(), anthropic_chain),
+    (lambda x: "langchain" in x["topic"].lower(), langchain_chain),
+    general_chain,
+)
+
+topic_classifier = (
+    PromptTemplate.from_template(
+        """Given the user question below, classify it as either being about `LangChain`, `Anthropic`, or `Other`.
+
+    Do not respond with more than one word.
+
+    <question>
+    {question}
+    </question>
+
+    Classification:"""
+    )
+    | ChatAnthropic()
+    | StrOutputParser()
+)
+
+
+# User input
+class UserInput(BaseModel):
+    question: str
+    session_id: str
+
+
+def classify_session(session_id: str):
+    result = zep.memory.classify_session(
+        session_id, "intent", ["langchain", "anthropic", "none"], persist=True
+    )
+    return result.class_
+
+
+def invoke_chain(user_input: UserInput):
+    result_chain = RunnableWithMessageHistory(
+        RunnablePassthrough.assign(session_id=lambda x: user_input["session_id"])
+        | {
+            "question": lambda x: x["question"],
+            "topic": lambda x: classify_session(x["session_id"]),
+        }
+        | branch,
+        lambda session_id: ZepChatMessageHistory(
+            session_id=session_id,
+            zep_client=zep,
+            memory_type="perpetual",
+        ),
+        input_messages_key="question",
+        history_messages_key="chat_history",
+    )
+
+    return result_chain.invoke(
+        user_input, config={"configurable": {"session_id": user_input["session_id"]}}
+    )
+
+
+chain = (
+    RunnableLambda(invoke_chain)
+    .with_types(input_type=UserInput)
+    .with_config(callbacks=[ConsoleCallbackHandler()])
+)

examples/langchain-langserve/app/real_world_classification.py

@@ -0,0 +1,142 @@
+import os
+
+from langchain.agents import AgentExecutor, AgentType, initialize_agent, load_tools
+from langchain.callbacks.tracers import ConsoleCallbackHandler
+from langchain_community.chat_models import ChatOpenAI
+from langchain_community.tools import WikipediaQueryRun
+from langchain_community.tools.yahoo_finance_news import YahooFinanceNewsTool
+from langchain_community.utilities import WikipediaAPIWrapper
+from langchain_core.output_parsers import StrOutputParser
+from langchain_core.prompts.prompt import PromptTemplate
+from langchain_core.pydantic_v1 import BaseModel
+from langchain_core.runnables import RunnableBranch, RunnableLambda, RunnablePassthrough
+from langchain_core.runnables.history import RunnableWithMessageHistory
+
+from zep_python import ZepClient
+from zep_python.langchain import ZepChatMessageHistory
+
+ZEP_API_KEY = os.environ.get("ZEP_API_KEY")  # Required for Zep Cloud
+ZEP_API_URL = os.environ.get(
+    "ZEP_API_URL"
+)  # only required if you're using Zep Open Source
+
+llm = ChatOpenAI(temperature=0.0)
+yahoo_chain = initialize_agent(
+    [YahooFinanceNewsTool()],
+    llm,
+    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
+    verbose=True,
+    handle_parsing_errors=True,
+)
+
+stack_chain = initialize_agent(
+    load_tools(["stackexchange"]),
+    llm,
+    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
+    verbose=True,
+    handle_parsing_errors=True,
+)
+
+wiki_chain = initialize_agent(
+    [WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())],
+    llm,
+    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
+    verbose=True,
+    handle_parsing_errors=True,
+)
+
+if ZEP_API_KEY is None:
+    raise ValueError(
+        "ZEP_API_KEY is required for Zep Cloud. "
+        "Remove this check if using Zep Open Source."
+    )
+
+zep = ZepClient(
+    api_key=ZEP_API_KEY,
+    api_url=ZEP_API_URL,  # only required if you're using Zep Open Source
+)
+
+wikipedia_search_prompt = PromptTemplate.from_template(
+    """Please answer the question based only on the following context:
+Context: {context}
+Question: {question}
+Answer:"""
+)
+
+
+def invoke_agent_executor(agent: AgentExecutor, x: any):
+    result = agent.invoke(input=x["question"])
+    return result["output"]
+
+
+general_chain = PromptTemplate.from_template(
+    """Please say that you cannot answer the question, be sarcastic:
+
+Question: {question}
+Answer:"""
+)
+
+branch = RunnableBranch(
+    (
+        lambda x: "research" in x["topic"].lower(),
+        lambda x: invoke_agent_executor(wiki_chain, x),
+    ),
+    (
+        lambda x: "finance_news" in x["topic"].lower(),
+        lambda x: invoke_agent_executor(yahoo_chain, x),
+    ),
+    (
+        lambda x: "dev_question" in x["topic"].lower(),
+        lambda x: invoke_agent_executor(stack_chain, x),
+    ),
+    general_chain | StrOutputParser(),
+)
+
+
+# User input
+class UserInput(BaseModel):
+    question: str
+    session_id: str
+
+
+def classify_session(session_id: str):
+    result = zep.memory.classify_session(
+        session_id,
+        "intent",
+        [
+            "research",
+            "dev_question",
+            "finance_news",
+            "none",
+        ],
+    )
+    return result.class_
+
+
+def invoke_chain(user_input: UserInput):
+    result_chain = RunnableWithMessageHistory(
+        RunnablePassthrough.assign(session_id=lambda x: user_input["session_id"])
+        | {
+            "question": lambda x: x["question"],
+            "topic": lambda x: classify_session(x["session_id"]),
+        }
+        | branch,
+        lambda session_id: ZepChatMessageHistory(
+            session_id=session_id,
+            zep_client=zep,
+            memory_type="perpetual",
+        ),
+        input_messages_key="question",
+        history_messages_key="chat_history",
+    )
+
+    return result_chain.invoke(
+        user_input, config={"configurable": {"session_id": user_input["session_id"]}}
+    )
+
+
+chain = (
+    RunnableLambda(invoke_chain)
+    .with_types(input_type=UserInput)
+    .with_config(callbacks=[ConsoleCallbackHandler()])
+)

examples/langchain-langserve/app/server.py

@@ -2,6 +2,7 @@
 
 load_dotenv(dotenv_path=find_dotenv())
 
+from classification_chain import chain as classification_chain
 from fastapi import FastAPI
 from fastapi.responses import RedirectResponse
 from langserve import add_routes
@@ -10,6 +11,7 @@
     chain as message_history_vector_store_chain,
 )
 from rag_vector_store_chain import chain as rag_vector_store_chain
+from real_world_classification import chain as real_world_classification_chain
 
 app = FastAPI()
 
@@ -24,6 +26,8 @@ async def redirect_root_to_docs():
 add_routes(
     app, message_history_vector_store_chain, path="/message_history_vector_store"
 )
+add_routes(app, classification_chain, path="/classification")
+add_routes(app, real_world_classification_chain, path="/real_world_classification")
 
 if __name__ == "__main__":
     import uvicorn