HighScienceHub

Bridging Curiosity to Clarity in Tech, Science & AI

Deep Dive into Building AI Agents with LangChain and LangGraph

@09 mins

1. Understanding LangChain: Core Components and Architecture

LangChain is a framework designed to integrate large language models (LLMs) with external data sources and applications. Its modular architecture allows developers to create sophisticated AI workflows. Here’s a breakdown of its core components:

1.1 Document Loaders

  • Purpose: Load data from various formats (PDFs, CSVs, websites, databases).

Tools:

  • PyPDFLoader: Extracts text from PDFs.
  • WebBaseLoader: Scrapes web content.
  • NotionLoader: Imports Notion pages.

Example

from langchain.document_loaders import PyPDFLoader loader = PyPDFLoader("research_paper.pdf") documents = loader.load()

1.2 Text Splitters

  • Purpose: Break documents into manageable chunks for processing.

Strategies

  • RecursiveCharacterTextSplitter: Splits text by characters (e.g., paragraphs, sentences).
  • TokenTextSplitter: Splits by token count (useful for models with token limits).

Example

from langchain.text_splitter import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200) texts = splitter.split_documents(documents)

1.3 Vector Stores

  • Purpose: Convert text chunks into embeddings (vector representations) for semantic search.

Tools

  • FAISS: Local vector store for fast similarity search.
  • Pinecone: Cloud-based vector database for scalability.

Example

from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import FAISS embeddings = OpenAIEmbeddings(api_key="your-key") vector_store = FAISS.from_documents(texts, embeddings) retriever = vector_store.as_retriever()

1.4 Chains

  • Purpose: Sequence operations to build complex workflows.

Types

  • RetrievalQA: Combines retrieval and question-answering.
  • ConversationalRetrievalChain: Adds memory for multi-turn chats.

Exampl

from langchain.chains import RetrievalQA from langchain.llms import OpenAI llm = OpenAI(temperature=0) qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=retriever, return_source_documents=True ) response = qa_chain("What are the key findings?")

2. LangGraph: Building Stateful, Multi-Agent Workflows

LangGraph extends LangChain by enabling graph-based workflows, where nodes represent agents or functions, and edges define transitions based on conditions.

2.1 Key Concepts

  • State: A shared dictionary passed between nodes, evolving as the workflow progresses.
  • Nodes: Functions or agents that modify the state.
  • Edges: Define transitions between nodes (conditional or unconditional).

2.2 Example: Medical Triage System

Workflow:

  1. Symptom Checker: Assesses patient symptoms.
  2. Triage Agent: Determines urgency (Emergency/Non-Urgent).
  3. Lab Recommender: Suggests tests based on triage.
  4. Alert System: Notifies staff if critical.

Code Implementation:from langgraph.graph import StateGraph, END # Define State Schema from typing import TypedDict, List class MedicalState(TypedDict): symptoms: str triage_level: str tests: List[str] alert: str # Nodes (Agents) def symptom_checker(state: MedicalState): return {"symptoms": state["symptoms"]} def triage_agent(state: MedicalState): symptoms = state["symptoms"].lower() if "chest pain" in symptoms or "shortness of breath" in symptoms: return {"triage_level": "Emergency"} return {"triage_level": "Non-Urgent"} def lab_agent(state: MedicalState): if state["triage_level"] == "Emergency": return {"tests": ["ECG", "Troponin Test", "Chest X-Ray"]} return {"tests": ["Complete Blood Count", "Basic Metabolic Panel"]} def alert_agent(state: MedicalState): if state["triage_level"] == "Emergency": return {"alert": "STAT: Cardiac team required in ER"} return {"alert": "Routine checkup needed"} # Build Graph workflow = StateGraph(MedicalState) workflow.add_node("symptom_checker", symptom_checker) workflow.add_node("triage", triage_agent) workflow.add_node("labs", lab_agent) workflow.add_node("alerts", alert_agent) # Define Edges workflow.add_edge("symptom_checker", "triage") workflow.add_edge("triage", "labs") workflow.add_edge("labs", "alerts") workflow.add_edge("alerts", END) # Run Workflow initial_state = {"symptoms": "chest pain and dizziness"} result = workflow.execute(initial_state) print(result)

Output:{ 'symptoms': 'chest pain and dizziness', 'triage_level': 'Emergency', 'tests': ['ECG', 'Troponin Test', 'Chest X-Ray'], 'alert': 'STAT: Cardiac team required in ER' }

3. Advanced Techniques and Best Practices

3.1 Integrating External APIs

  • Lab Test API Example:

import requests def fetch_lab_guidelines(triage_level): response = requests.get( f"https://api.medicalguidelines.com/tests?urgency={triage_level}" ) return response.json()["recommended_tests"] def lab_agent(state: MedicalState): guidelines = fetch_lab_guidelines(state["triage_level"]) return {"tests": guidelines}

3.2 Error Handling and Validation

  • State Validation:

from pydantic import ValidationError try: validated_state = MedicalState(**state) except ValidationError as e: print(f"Invalid state: {e}")

  • Fallback Mechanisms:

def lab_agent(state: MedicalState): try: guidelines = fetch_lab_guidelines(state["triage_level"]) except requests.RequestException: guidelines = ["Basic Metabolic Panel"] # Default test return {"tests": guidelines}

3.3 Performance Optimization

  • Caching:

from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache set_llm_cache(InMemoryCache()) # Cache LLM responses

  • Parallel Execution:
    Use asyncio for concurrent node processing:

import asyncio async def async_lab_agent(state: MedicalState): await asyncio.sleep(1) # Simulate API call return {"tests": ["ECG"]} async def main(): app = workflow.compile() result = await app.ainvoke(initial_state) print(result) asyncio.run(main())

4. Deployment and Monitoring

4.1 Containerization with Docker

  • Dockerfile:

FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["python", "medical_workflow.py"]

4.2 Logging and Analytics

  • LangSmith Integration:

import os os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-key"

  • Custom Logging:

import logging logging.basicConfig(filename='workflow.log', level=logging.INFO) def triage_agent(state: MedicalState): logging.info(f"Triage level determined: {state['triage_level']}") return {"triage_level": "Emergency"}

5. Real-World Use Cases

5.1 Customer Support Escalation System

  • Workflow:
  1. Intent Recognition: Classify user queries (e.g., billing, technical).
  2. Automated Response: Use LangChain to retrieve FAQs.
  3. Escalation Routing: LangGraph routes complex issues to human agents.
  4. Post-Call Analytics: Sentiment analysis with AWS Comprehend.

5.2 Financial Fraud Detection

Workflow

  1. Transaction Analysis: LangChain retrieves user history.
  2. Anomaly Detection: LangGraph triggers alerts for suspicious patterns.
  3. Case Management: Assigns fraud cases to investigators.

6. Resources and Community

Official Documentation:

Courses

  • LangChain for LLM Application Development (DeepLearning.AI)
  • Building AI Agents with LangGraph (Udemy)

GitHub Repos

Final Note: Mastery of LangChain and LangGraph requires hands-on experimentation. Start with small projects, iterate, and leverage the community. The AI agent market is booming—your journey starts now.

Related News