READ THIS TO UNDERSTAND RAW :
https://chatgpt.com/share/6a168d70-a25c-8321-9244-16402d447dd4
IN LANGCHAIN TRANSITION : https://chatgpt.com/share/6a1699e5-0544-8320-8648-1b9e1a89001b
take-this github to learn langchain : https://github.com/krishnaik06/Langchain-V1-Crash-Course/blob/main
check at the end of chat : https://chatgpt.com/share/6a1699e5-0544-8320-8648-1b9e1a89001b
full agentic workflow ofopeai chatbot :
import os
from dotenv import load_dotenv
from langchain.agents import create_agent
load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY")
def get_weather(city: str) -> str: """Get weather of a city."""
return f"The weather in {city} is 32°C and sunny."
def add_numbers(a: int, b: int) -> str: """Add two numbers."""
return f"Sum is {a + b}"
def recommend_movie(genre: str) -> str: """Recommend movie based on genre."""
return f"Recommended {genre} movie: Inception"
agent = create_agent(
model="gpt-5",
tools=[
get_weather,
add_numbers,
recommend_movie
],
system_prompt="You are a helpful AI assistant."
)
response = agent.invoke({
"messages": [
{
"role": "user",
"content":
"""
What is the weather in Mysore?
Also add 10 and 20.
Recommend a sci-fi movie.
"""
}
]
})
print(response["messages"][-1].content)
import os from dotenv import load_dotenv
load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY") os.environ["GOOGLE_API_KEY"] = os.getenv("GOOGLE_API_KEY") os.environ["GROQ_API_KEY"] = os.getenv("GROQ_API_KEY")
from langchain.chat_models import init_chat_model
model = init_chat_model("gpt-4.1")
response = model.invoke("Hello how are you?")
print(response.content)
from langchain_openai import ChatOpenAI
model = ChatOpenAI(
model="gpt-4.1",
temperature=0.7
)
response = model.invoke("Explain AI simply")
print(response.content)
from langchain_google_genai import ChatGoogleGenerativeAI
model = ChatGoogleGenerativeAI(
model="gemini-2.5-flash-lite"
)
response = model.invoke("Why do parrots talk?")
print(response.content)
from langchain_groq import ChatGroq
model = ChatGroq(
model="qwen/qwen3-32b"
)
response = model.invoke("What is quantum computing?")
print(response.content)
for chunk in model.stream(
"Write 100 words on Artificial Intelligence"
):
# Print chunks in real-time
print(chunk.text, end="", flush=True)
responses = model.batch([
"What is AI?",
"What is Machine Learning?",
"What is Deep Learning?"
])
for response in responses:
print("\n")
print(response.content)
""" OPENAI_API_KEY=your_openai_key
GOOGLE_API_KEY=your_google_key
GROQ_API_KEY=your_groq_key """ giving structured response so that other ai database can read
from pydantic import BaseModel, Field from langchain_ollama import ChatOllama
class TaskPlan(BaseModel): tasks: list[str] = Field(description="Task list") priority: str = Field(description="Overall priority") deadline: str = Field(description="Completion deadline")
llm = ChatOllama(model="llama3")
structured_llm = llm.with_structured_output( TaskPlan )
prompt = """ I need to prepare for interviews, learn LangChain, and complete ML project in 10 days. """
response = structured_llm.invoke(prompt)
print(response)
| Concept | Meaning |
|---|---|
| Middleware | Logic layer around LLM |
| SummarizationMiddleware | Compress old history |
| trigger | When summarization starts |
| keep | Recent messages preserved |
| checkpointer | Stores conversation state |
| thread_id | Unique conversation ID |
| ToolMessage | Result from tool |
full healthcare with all guardlayers implied
Layer 1 ContentFilterMiddleware()
Blocks harmful keywords.
Layer 2 PIIMiddleware()
Masks private data.
Layer 3 HumanInTheLoopMiddleware()
Human approval.
Layer 4 PIIMiddleware(output)
Protects outgoing responses.
Layer 5 SafetyGuardrailMiddleware()
Final AI safety inspection.
from typing import Any
from langchain.agents import create_agent from langchain.agents.middleware import ( PIIMiddleware, HumanInTheLoopMiddleware, AgentMiddleware, AgentState, hook_config )
from langchain_core.tools import tool from langchain_core.messages import AIMessage
from langgraph.runtime import Runtime from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command
from langchain_openai import ChatOpenAI
class HealthcareSafetyFilter(AgentMiddleware):
# banned dangerous topics
BLOCKED_TOPICS = [
"hack",
"weapon",
"drug synthesis",
"suicide",
"malware"
]
@hook_config(can_jump_to=["end"])
def before_agent(
self,
state: AgentState,
runtime: Runtime
) -> dict[str, Any] | None:
# if no messages then skip
if not state["messages"]:
return None
# get first user message
first_msg = state["messages"][0]
# ensure message is from human
if first_msg.type != "human":
return None
# convert input to lowercase
content = first_msg.content.lower()
# check banned topics
for topic in self.BLOCKED_TOPICS:
if topic in content:
print(f"🚫 BLOCKED TOPIC: {topic}")
# stop workflow immediately
return {
"messages": [
{
"role": "assistant",
"content": (
"I can only help with safe healthcare-related requests."
)
}
],
# terminate agent execution
"jump_to": "end"
}
return None
class MedicalOutputValidator(AgentMiddleware):
DISCLAIMER = (
"\n\n⚕️ This is general health information."
" Please consult a doctor."
)
@hook_config(can_jump_to=["end"])
def after_agent(
self,
state: AgentState,
runtime: Runtime
) -> dict[str, Any] | None:
# skip if no messages
if not state["messages"]:
return None
# get last AI response
last_message = state["messages"][-1]
# ensure message is AI generated
if not isinstance(last_message, AIMessage):
return None
# append disclaimer to response
last_message.content += self.DISCLAIMER
return None
@tool def search_symptoms(symptoms: str) -> str: """ Search medical symptoms. """
return (
f"Symptoms related to {symptoms} found."
)
@tool def medication_info(medicine: str) -> str: """ Get medicine details. """
return (
f"{medicine} is commonly used in healthcare."
)
@tool def book_appointment( patient_name: str, doctor: str, date: str ) -> str: """ Book hospital appointment. """
return (
f"Appointment booked for "
f"{patient_name} with Dr.{doctor} on {date}"
)
healthcare_agent = create_agent(
# main LLM
model="gpt-4o",
# available tools
tools=[
search_symptoms,
medication_info,
book_appointment
],
# middleware layers
middleware=[
# =====================================
# LAYER 1 : INPUT FILTER
# blocks harmful requests
# =====================================
HealthcareSafetyFilter(),
# =====================================
# LAYER 2 : INPUT PII REDACTION
# hides emails before model sees them
# =====================================
PIIMiddleware(
"email",
strategy="redact",
apply_to_input=True
),
# =====================================
# LAYER 3 : INPUT CREDIT CARD MASKING
# masks sensitive card numbers
# =====================================
PIIMiddleware(
"credit_card",
strategy="mask",
apply_to_input=True
),
# =====================================
# LAYER 4 : HUMAN APPROVAL
# pauses before appointment booking
# =====================================
HumanInTheLoopMiddleware(
interrupt_on={
# approval required
"book_appointment": True,
# auto approved
"search_symptoms": False,
"medication_info": False
}
),
# =====================================
# LAYER 5 : OUTPUT VALIDATION
# adds medical disclaimer
# =====================================
MedicalOutputValidator()
],
# stores paused state/memory
checkpointer=InMemorySaver(),
# system instructions for AI
system_prompt=(
"You are a healthcare assistant. "
"Help users with symptoms, medicines, "
"and appointments safely."
)
)
config = { "configurable": { "thread_id": "health_session_1" } }
result = healthcare_agent.invoke(
{
"messages": [
{
"role": "user",
"content": (
"What are symptoms of diabetes?"
)
}
]
},
config=config
)
print("\n=== SAFE RESPONSE ===\n")
print(result["messages"][-1].content)
Email gets hidden automatically
result = healthcare_agent.invoke(
{
"messages": [
{
"role": "user",
"content": (
"My email is john@gmail.com "
"What medicine helps headache?"
)
}
]
},
config=config
)
print("\n=== PII REDACTION ===\n")
print(result["messages"][-1].content)
result = healthcare_agent.invoke(
{
"messages": [
{
"role": "user",
"content": (
"How to synthesize drugs?"
)
}
]
},
config=config
)
print("\n=== BLOCKED REQUEST ===\n")
print(result["messages"][-1].content)
appointment_config = { "configurable": { "thread_id": "appointment_1" } }
result = healthcare_agent.invoke(
{
"messages": [
{
"role": "user",
"content": (
"Book appointment with "
"Dr Sharma on Monday"
)
}
]
},
config=appointment_config
)
print("\n=== WAITING FOR APPROVAL ===\n")
print(result)
approved = healthcare_agent.invoke(
Command(
resume={
"decisions": [
{
"type": "approve"
}
]
}
),
config=appointment_config
)
print("\n=== APPROVED RESPONSE ===\n")
print(approved["messages"][-1].content)