LangChain Enterprise Development Guide

1pip install langchain langchain-openai langchain-community chromadb
2
3import os
4from langchain_openai import ChatOpenAI
5
6llm = ChatOpenAI(
7    model="gpt-4o",
8    temperature=0,
9    api_key=os.environ["OPENAI_API_KEY"],
10)

1from langchain_core.prompts import ChatPromptTemplate
2from langchain_core.output_parsers import StrOutputParser
3
4prompt = ChatPromptTemplate.from_messages([
5    ("system", "You are a {role}. Answer in {language}."),
6    ("human",  "{question}"),
7])
8
9# LCEL (LangChain Expression Language) — composable pipeline
10chain = prompt | llm | StrOutputParser()
11
12result = chain.invoke({
13    "role":     "senior DevOps engineer",
14    "language": "English",
15    "question": "What is the difference between a container and a VM?",
16})
17print(result)

1from langchain_community.document_loaders import DirectoryLoader
2from langchain.text_splitter import RecursiveCharacterTextSplitter
3from langchain_community.vectorstores import Chroma
4from langchain_openai import OpenAIEmbeddings
5from langchain.chains import RetrievalQA
6
7# 1. Load documents
8loader = DirectoryLoader("./docs", glob="**/*.md")
9docs   = loader.load()
10
11# 2. Split into chunks
12splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
13chunks   = splitter.split_documents(docs)
14
15# 3. Embed and store
16vectordb = Chroma.from_documents(
17    documents=chunks,
18    embedding=OpenAIEmbeddings(),
19    persist_directory="./chroma_db",
20)
21
22# 4. Retrieval QA chain
23qa_chain = RetrievalQA.from_chain_type(
24    llm=llm,
25    chain_type="stuff",
26    retriever=vectordb.as_retriever(search_kwargs={"k": 5}),
27    return_source_documents=True,
28)
29
30result = qa_chain.invoke({"query": "How do I deploy with Docker Compose?"})
31print(result["result"])

1from langchain.agents import create_tool_calling_agent, AgentExecutor
2from langchain_core.tools import tool
3
4@tool
5def search_docs(query: str) -> str:
6    """Search the internal documentation for an answer."""
7    results = vectordb.similarity_search(query, k=3)
8    return "
9".join([doc.page_content for doc in results])
10
11@tool
12def get_current_time() -> str:
13    """Get the current UTC time."""
14    from datetime import datetime, timezone
15    return datetime.now(timezone.utc).isoformat()
16
17tools = [search_docs, get_current_time]
18
19agent = create_tool_calling_agent(llm, tools, prompt)
20executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=5)
21
22result = executor.invoke({"input": "What does our docs say about Redis clustering?"})

1from langchain_community.chat_message_histories import RedisChatMessageHistory
2from langchain_core.runnables.history import RunnableWithMessageHistory
3
4# Persistent session memory backed by Redis
5def get_history(session_id: str):
6    return RedisChatMessageHistory(
7        session_id=session_id,
8        url=os.environ["REDIS_URL"],
9        ttl=3600,
10    )
11
12chain_with_memory = RunnableWithMessageHistory(
13    chain,
14    get_history,
15    input_messages_key="input",
16    history_messages_key="history",
17)
18
19response = chain_with_memory.invoke(
20    {"input": "What is Python?"},
21    config={"configurable": {"session_id": "user-123"}},
22)

1# LangSmith — tracing and evaluation
2export LANGCHAIN_TRACING_V2=true
3export LANGCHAIN_API_KEY=ls__...
4export LANGCHAIN_PROJECT=my-rag-app
5
6# All chain invocations are now traced automatically
7
8# Evaluate RAG faithfulness
9from langchain.evaluation import load_evaluator
10
11evaluator = load_evaluator("qa", llm=llm)
12result = evaluator.evaluate_strings(
13    prediction="Redis uses RESP protocol...",
14    reference="Redis uses the RESP (REdis Serialisation Protocol)...",
15    input="What protocol does Redis use?",
16)
17print(result["score"])   # 0.0 - 1.0