Zum Inhalt

LangChain Framework Cheat Sheet

Überblick

LangChain ist ein umfassender Rahmen für die Entwicklung von Anwendungen, die von großen Sprachmodellen (LLMs) betrieben werden, um jede Phase des LLM Anwendungslebenszyklus von der Entwicklung bis zum Einsatz zu vereinfachen. Um die Komplexität der produktionsbereiten KI-Anwendungen zu bewältigen, bietet LangChain eine einheitliche Schnittstelle für die Zusammenarbeit mit verschiedenen LLMs und bietet leistungsstarke Abstraktionen für Kettenoperationen, die Verwaltung von Kontexten und die Integration mit externen Systemen.

Was LangChain besonders leistungsstark macht, ist seine modulare Architektur, mit der Entwickler komplexe KI-Workflows mit einfachen, wiederverwendbaren Komponenten komponieren können. Der Rahmen zeichnet sich durch die Erstellung von Kontext-Aware-Anwendungen aus, die LLMs mit privaten Datenquellen, externen APIs und spezialisierten Tools verbinden können. Mit seiner LangChain Expression Language (LCEL) können Entwickler anspruchsvolle Ketten aufbauen, die alles von einfachen Frage-Anbietern bis hin zu komplexen mehrstufigen Denkens und autonomen Verhaltensweisen handhaben.

LangChain ist zum De-facto-Standard für die LLM-Anwendungsentwicklung geworden, unterstützt die Integration mit nahezu jedem großen LLM-Anbieter und bietet durch seine Begleitplattform LangSmith umfangreiche Werkzeuge für die Produktion, Überwachung und Auswertung.

Installation und Inbetriebnahme

Einfache Installation

# Install core LangChain
pip install langchain

# Install with specific integrations
pip install langchain-openai
pip install langchain-anthropic
pip install langchain-google-genai
pip install langchain-community

# Install additional components
pip install langchain-experimental
pip install langsmith  # For monitoring and evaluation
pip install langgraph  # For advanced agent workflows
```_

### Umweltkonfiguration
```python
import os
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# Set up API keys
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"
os.environ["ANTHROPIC_API_KEY"] = "your-anthropic-api-key"
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"

# Initialize LLMs
openai_llm = ChatOpenAI(model="gpt-4", temperature=0.7)
anthropic_llm = ChatAnthropic(model="claude-3-sonnet-20240229")
```_

### Projektstruktur

langchain_project/ ├── chains/ │ ├── init.py │ ├── qa_chain.py │ └── summarization_chain.py ├── agents/ │ ├── init.py │ ├── research_agent.py │ └── analysis_agent.py ├── tools/ │ ├── init.py │ ├── search_tools.py │ └── database_tools.py ├── prompts/ │ ├── init.py │ └── templates.py ├── memory/ │ ├── init.py │ └── custom_memory.py ├── config/ │ ├── init.py │ └── settings.py └── main.py ```_

LangChain Expression Language (LCEL)

Basic Chain Konstruktion

```python from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI

Create a simple chain using LCEL

prompt = ChatPromptTemplate.from_template("Tell me a joke about \\{topic\\}") llm = ChatOpenAI(model="gpt-3.5-turbo") output_parser = StrOutputParser()

Chain components using the pipe operator

| chain = prompt | llm | output_parser |

Execute the chain

result = chain.invoke(\\{"topic": "programming"\\}) print(result) ```_

Komplexe Kettenzusammensetzung

```python from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnableParallel, RunnablePassthrough from langchain_openai import ChatOpenAI from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings

RAG (Retrieval Augmented Generation) chain

vectorstore = FAISS.from_texts( ["LangChain is a framework for LLM applications", "LCEL is the expression language for LangChain"], embedding=OpenAIEmbeddings() ) retriever = vectorstore.as_retriever()

Complex chain with parallel processing

rag_chain = ( RunnableParallel(\\{ "context": retriever, "question": RunnablePassthrough() \\}) |ChatPromptTemplate.from_template(""" Answer the question based on the context: Context: \\{context\\} Question: \\{question\\} Answer: """) |ChatOpenAI() |StrOutputParser() )

result = rag_chain.invoke("What is LangChain?") ```_

Bedingte Logik in Ketten

```python from langchain_core.runnables import RunnableBranch from langchain_core.prompts import ChatPromptTemplate

Conditional chain execution

def route_question(info): if "math" in info["question"].lower(): return "math" elif "history" in info["question"].lower(): return "history" else: return "general"

Different prompts for different question types

math_prompt = ChatPromptTemplate.from_template( "You are a math expert. Solve this problem: \\{question\\}" ) history_prompt = ChatPromptTemplate.from_template( "You are a history expert. Answer this question: \\{question\\}" ) general_prompt = ChatPromptTemplate.from_template( "Answer this general question: \\{question\\}" )

Branching chain

branching_chain = RunnableBranch( (lambda x: route_question(x) == "math", math_prompt|llm), (lambda x: route_question(x) == "history", history_prompt|llm), general_prompt|llm # Default branch )

result = branching_chain.invoke(\\{"question": "What is 2+2?"\\}) ```_

Streaming und Async-Unterstützung

```python import asyncio from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-3.5-turbo") prompt = ChatPromptTemplate.from_template("Write a story about \\{topic\\}") chain = prompt|llm

Streaming execution

for chunk in chain.stream(\\{"topic": "space exploration"\\}): print(chunk.content, end="", flush=True)

Async execution

async def async_chain_execution(): result = await chain.ainvoke(\\{"topic": "artificial intelligence"\\}) return result

Batch processing

batch_inputs = [ \\{"topic": "robots"\\}, \\{"topic": "future"\\}, \\{"topic": "technology"\\} ] batch_results = chain.batch(batch_inputs) ```_

Vorteile und Vorlagen

Basic Prompt Vorlagen

```python from langchain_core.prompts import PromptTemplate, ChatPromptTemplate from langchain_core.prompts import HumanMessagePromptTemplate, SystemMessagePromptTemplate

Simple prompt template

simple_prompt = PromptTemplate( input_variables=["product"], template="What is a good name for a company that makes \\{product\\}?" )

Chat prompt template

chat_prompt = ChatPromptTemplate.from_messages([ SystemMessagePromptTemplate.from_template( "You are a helpful assistant that translates \\{input_language\\} to \\{output_language\\}." ), HumanMessagePromptTemplate.from_template("\\{text\\}") ])

Using the templates

formatted_prompt = simple_prompt.format(product="colorful socks") chat_messages = chat_prompt.format_messages( input_language="English", output_language="French", text="I love programming" ) ```_

Advanced Prompt Engineering

```python from langchain_core.prompts import ChatPromptTemplate, FewShotPromptTemplate from langchain_core.example_selectors import SemanticSimilarityExampleSelector from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS

Few-shot prompting

examples = [ \\{"input": "happy", "output": "sad"\\}, \\{"input": "tall", "output": "short"\\}, \\{"input": "energetic", "output": "lethargic"\\} ]

example_prompt = PromptTemplate( input_variables=["input", "output"], template="Input: \\{input\\}\nOutput: \\{output\\}" )

Semantic similarity example selector

example_selector = SemanticSimilarityExampleSelector.from_examples( examples, OpenAIEmbeddings(), FAISS, k=2 )

few_shot_prompt = FewShotPromptTemplate( example_selector=example_selector, example_prompt=example_prompt, prefix="Give the antonym of every input", suffix="Input: \\{adjective\\}\nOutput:", input_variables=["adjective"] )

Dynamic prompt selection

def select_prompt_based_on_input(input_text): if len(input_text.split()) > 50: return long_text_prompt else: return short_text_prompt

dynamic_chain = ( RunnablePassthrough.assign( prompt=lambda x: select_prompt_based_on_input(x["text"]) ) |(lambda x: x["prompt"].format(**x)) |llm ) ```_

Prompt Zusammensetzung und Wiederverwendung

```python from langchain_core.prompts import ChatPromptTemplate

Reusable prompt components

system_context = """You are an expert \\{domain\\} consultant with over 10 years of experience. You provide detailed, actionable advice based on industry best practices."""

analysis_template = """Analyze the following \\{item_type\\}:

\\{content\\}

Please provide: 1. Key strengths and weaknesses 2. Recommendations for improvement 3. Industry benchmarks and comparisons 4. Next steps and action items"""

Compose prompts dynamically

def create_analysis_prompt(domain, item_type): return ChatPromptTemplate.from_messages([ ("system", system_context.format(domain=domain)), ("human", analysis_template.format(item_type=item_type, content="\\{content\\}")) ])

Create specialized prompts

marketing_prompt = create_analysis_prompt("marketing", "campaign") financial_prompt = create_analysis_prompt("finance", "budget") technical_prompt = create_analysis_prompt("technology", "architecture") ```_

Speicherverwaltung

Grundlegende Speichertypen

```python from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory from langchain.memory import ConversationBufferWindowMemory, ConversationTokenBufferMemory from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-3.5-turbo")

Buffer memory - stores all conversation history

buffer_memory = ConversationBufferMemory( memory_key="chat_history", return_messages=True )

Window memory - keeps only last k interactions

window_memory = ConversationBufferWindowMemory( k=5, memory_key="chat_history", return_messages=True )

Summary memory - summarizes old conversations

summary_memory = ConversationSummaryMemory( llm=llm, memory_key="chat_history", return_messages=True )

Token buffer memory - limits by token count

token_memory = ConversationTokenBufferMemory( llm=llm, max_token_limit=1000, memory_key="chat_history", return_messages=True ) ```_

Custom Memory Implementierung

```python from langchain.memory.chat_memory import BaseChatMemory from langchain_core.messages import BaseMessage from typing import List, Dict, Any import json

class DatabaseChatMemory(BaseChatMemory): """Custom memory that persists to database"""

def __init__(self, session_id: str, connection_string: str):
    super().__init__()
    self.session_id = session_id
    self.connection_string = connection_string
    self.setup_database()

def setup_database(self):
    # Initialize database connection and tables
    pass

def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
    # Save conversation to database
    conversation_data = \\\\{
        "session_id": self.session_id,
        "inputs": inputs,
        "outputs": outputs,
        "timestamp": datetime.now()
    \\\\}
    self.save_to_database(conversation_data)

def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
    # Load conversation history from database
    history = self.load_from_database(self.session_id)
    return \\\\{"chat_history": history\\\\}

def clear(self) -> None:
    # Clear conversation history
    self.clear_database_session(self.session_id)

Usage

custom_memory = DatabaseChatMemory( session_id="user_123", connection_string="postgresql://localhost/chatbot" ) ```_

Speicher mit LCEL Ketten

```python from langchain_core.runnables import RunnablePassthrough, RunnableLambda from langchain.memory import ConversationBufferMemory from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

Memory integration with LCEL

memory = ConversationBufferMemory( memory_key="chat_history", return_messages=True )

prompt = ChatPromptTemplate.from_messages([ ("system", "You are a helpful assistant"), MessagesPlaceholder(variable_name="chat_history"), ("human", "\\{input\\}") ])

def load_memory(input_dict): return memory.load_memory_variables(\\{\\})

def save_memory(input_dict, output): memory.save_context( \\{"input": input_dict["input"]\\}, \\{"output": output.content\\} ) return output

Chain with memory

memory_chain = ( RunnablePassthrough.assign( chat_history=RunnableLambda(load_memory) ) |prompt |llm |RunnableLambda(lambda output: save_memory(\\{"input": "current_input"\\}, output)) ) ```_

Persistenter Speicher

```python import pickle import os from langchain.memory import ConversationBufferMemory

class PersistentMemory: def init(self, file_path: str): self.file_path = file_path self.memory = self.load_memory()

def load_memory(self):
    if os.path.exists(self.file_path):
        with open(self.file_path, 'rb') as f:
            return pickle.load(f)
    else:
        return ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True
        )

def save_memory(self):
    with open(self.file_path, 'wb') as f:
        pickle.dump(self.memory, f)

def add_message(self, inputs, outputs):
    self.memory.save_context(inputs, outputs)
    self.save_memory()

def get_memory_variables(self):
    return self.memory.load_memory_variables(\\\\{\\\\})

def clear_memory(self):
    self.memory.clear()
    self.save_memory()

Usage

persistent_memory = PersistentMemory("./chat_memory.pkl") ```_

Werkzeuge und Funktion Anrufen

Eingebaute Werkzeuge

```python from langchain_community.tools import DuckDuckGoSearchRun from langchain_community.tools import WikipediaQueryRun from langchain_community.utilities import WikipediaAPIWrapper from langchain_community.tools import ShellTool from langchain_community.tools.file_management import WriteFileTool, ReadFileTool

Search tools

search_tool = DuckDuckGoSearchRun() wikipedia_tool = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())

File management tools

write_tool = WriteFileTool() read_tool = ReadFileTool()

System tools

shell_tool = ShellTool()

Using tools directly

search_result = search_tool.run("LangChain framework") wiki_result = wikipedia_tool.run("artificial intelligence") ```_

Personalentwicklung

```python from langchain.tools import BaseTool from typing import Optional, Type from pydantic import BaseModel, Field import requests

class WeatherInput(BaseModel): location: str = Field(description="Location to get weather for")

class WeatherTool(BaseTool): name = "weather_tool" description = "Get current weather information for a location" args_schema: Type[BaseModel] = WeatherInput

def _run(self, location: str) -> str:
    # Implement weather API call
    try:
        api_key = os.getenv("WEATHER_API_KEY")
        url = f"http://api.openweathermap.org/data/2.5/weather?q=\\\\{location\\\\}&appid;=\\\\{api_key\\\\}"
        response = requests.get(url)
        data = response.json()

        if response.status_code == 200:
            temp = data['main']['temp'] - 273.15  # Convert from Kelvin
            description = data['weather'][0]['description']
            return f"Weather in \\\\{location\\\\}: \\\\{temp:.1f\\\\}°C, \\\\{description\\\\}"
        else:
            return f"Could not get weather for \\\\{location\\\\}"
    except Exception as e:
        return f"Error getting weather: \\\\{str(e)\\\\}"

async def _arun(self, location: str) -> str:
    # Async implementation
    return self._run(location)

Database query tool

class DatabaseTool(BaseTool): name = "database_query" description = "Execute SQL queries against the database"

def _run(self, query: str) -> str:
    # Implement database query logic
    try:
        # Connect to database and execute query
        result = execute_sql_query(query)
        return f"Query result: \\\\{result\\\\}"
    except Exception as e:
        return f"Database error: \\\\{str(e)\\\\}"

API integration tool

class APITool(BaseTool): name = "api_call" description = "Make HTTP requests to external APIs"

def _run(self, endpoint: str, method: str = "GET", data: dict = None) -> str:
    try:
        if method.upper() == "GET":
            response = requests.get(endpoint)
        elif method.upper() == "POST":
            response = requests.post(endpoint, json=data)

        return response.json()
    except Exception as e:
        return f"API error: \\\\{str(e)\\\\}"

```_

Werkzeugintegration mit Ketten

```python from langchain.tools import Tool from langchain_core.prompts import ChatPromptTemplate from langchain.agents import create_tool_calling_agent, AgentExecutor

Create tools list

tools = [ WeatherTool(), DatabaseTool(), APITool(), search_tool, wikipedia_tool ]

Tool-calling chain

prompt = ChatPromptTemplate.from_messages([ ("system", "You are a helpful assistant with access to various tools."), ("human", "\\{input\\}"), ("placeholder", "\\{agent_scratchpad\\}") ])

Create agent with tools

agent = create_tool_calling_agent(llm, tools, prompt) agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

Execute with tools

result = agent_executor.invoke(\\{ "input": "What's the weather in New York and find information about climate change?" \\}) ```_

Agenten und Autonomes Verhalten

Grundlegende Agententypen

```python from langchain.agents import create_react_agent, create_openai_functions_agent from langchain.agents import AgentExecutor from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

ReAct agent (Reasoning and Acting)

react_prompt = ChatPromptTemplate.from_messages([ ("system", "You are a helpful assistant"), ("human", "\\{input\\}"), MessagesPlaceholder(variable_name="agent_scratchpad") ])

react_agent = create_react_agent(llm, tools, react_prompt) react_executor = AgentExecutor( agent=react_agent, tools=tools, verbose=True, max_iterations=5 )

OpenAI Functions agent

functions_prompt = ChatPromptTemplate.from_messages([ ("system", "You are a helpful assistant"), ("human", "\\{input\\}"), MessagesPlaceholder(variable_name="agent_scratchpad") ])

functions_agent = create_openai_functions_agent(llm, tools, functions_prompt) functions_executor = AgentExecutor( agent=functions_agent, tools=tools, verbose=True ) ```_

Implementierung von Zollagern

```python from langchain.agents import BaseSingleActionAgent from langchain_core.agents import AgentAction, AgentFinish from typing import List, Union, Any, Dict

class CustomAgent(BaseSingleActionAgent): """Custom agent with specialized logic"""

def __init__(self, llm, tools, prompt):
    self.llm = llm
    self.tools = tools
    self.prompt = prompt

@property
def input_keys(self):
    return ["input"]

def plan(
    self,
    intermediate_steps: List[tuple],
    callbacks=None,
    **kwargs: Any
) -> Union[AgentAction, AgentFinish]:
    # Custom planning logic
    full_inputs = self.get_full_inputs(intermediate_steps, **kwargs)
    full_output = self.llm.predict(full_inputs)

    # Parse output to determine action
    if "Final Answer:" in full_output:
        return AgentFinish(
            return_values=\\\\{"output": full_output.split("Final Answer:")[-1].strip()\\\\},
            log=full_output
        )
    else:
        # Extract action and action input
        action = self.extract_action(full_output)
        return AgentAction(
            tool=action["tool"],
            tool_input=action["input"],
            log=full_output
        )

def get_full_inputs(self, intermediate_steps, **kwargs):
    # Build the full input string
    thoughts = ""
    for action, observation in intermediate_steps:
        thoughts += f"Action: \\\\{action.tool\\\\}\n"
        thoughts += f"Action Input: \\\\{action.tool_input\\\\}\n"
        thoughts += f"Observation: \\\\{observation\\\\}\n"

    return self.prompt.format(
        input=kwargs["input"],
        agent_scratchpad=thoughts
    )

def extract_action(self, text):
    # Parse the LLM output to extract action
    # Implementation depends on your prompt format
    pass

Use custom agent

custom_executor = AgentExecutor( agent=CustomAgent(llm, tools, custom_prompt), tools=tools, verbose=True ) ```_

Multi-Agent Systeme

```python from langchain.agents import AgentExecutor, create_openai_functions_agent from langchain_core.prompts import ChatPromptTemplate

Research agent

research_prompt = ChatPromptTemplate.from_messages([ ("system", "You are a research specialist. Focus on gathering comprehensive information."), ("human", "\\{input\\}"), MessagesPlaceholder(variable_name="agent_scratchpad") ])

research_agent = create_openai_functions_agent(llm, [search_tool, wikipedia_tool], research_prompt) research_executor = AgentExecutor(agent=research_agent, tools=[search_tool, wikipedia_tool])

Analysis agent

analysis_prompt = ChatPromptTemplate.from_messages([ ("system", "You are an analysis specialist. Focus on interpreting and analyzing information."), ("human", "\\{input\\}"), MessagesPlaceholder(variable_name="agent_scratchpad") ])

analysis_agent = create_openai_functions_agent(llm, [DatabaseTool()], analysis_prompt) analysis_executor = AgentExecutor(agent=analysis_agent, tools=[DatabaseTool()])

Coordinator agent

def multi_agent_workflow(query): # Step 1: Research research_result = research_executor.invoke(\\{"input": f"Research: \\{query\\}"\\})

# Step 2: Analysis
analysis_input = f"Analyze this research: \\\\{research_result['output']\\\\}"
analysis_result = analysis_executor.invoke(\\\\{"input": analysis_input\\\\})

# Step 3: Synthesis
synthesis_prompt = f"""
Based on the research and analysis below, provide a comprehensive answer:

Research: \\\\{research_result['output']\\\\}
Analysis: \\\\{analysis_result['output']\\\\}

Original Question: \\\\{query\\\\}
"""

final_result = llm.invoke(synthesis_prompt)
return final_result.content

Execute multi-agent workflow

result = multi_agent_workflow("What are the latest trends in AI?") ```_

Retrieval Augmented Generation (RAG)

Implementierung der RAG

```python from langchain_community.document_loaders import TextLoader, PDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough

Load and process documents

loader = TextLoader("documents.txt") documents = loader.load()

Split documents

text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200 ) splits = text_splitter.split_documents(documents)

Create embeddings and vector store

embeddings = OpenAIEmbeddings() vectorstore = FAISS.from_documents(splits, embeddings) retriever = vectorstore.as_retriever(search_kwargs=\\{"k": 3\\})

RAG prompt

rag_prompt = ChatPromptTemplate.from_template(""" Answer the question based only on the following context:

\\{context\\}

Question: \\{question\\}

Answer: """)

RAG chain

def format_docs(docs): return "\n\n".join(doc.page_content for doc in docs)

rag_chain = ( \\{"context": retriever|format_docs, "question": RunnablePassthrough()\\} |rag_prompt |llm |StrOutputParser() )

Query the RAG system

result = rag_chain.invoke("What is the main topic of the documents?") ```_

Erweiterte RAG mit Reranking

```python from langchain_community.document_transformers import EmbeddingsRedundantFilter from langchain.retrievers.document_compressors import EmbeddingsFilter from langchain.retrievers import ContextualCompressionRetriever

Create compression retriever with reranking

embeddings_filter = EmbeddingsFilter( embeddings=embeddings, similarity_threshold=0.76 )

compression_retriever = ContextualCompressionRetriever( base_compressor=embeddings_filter, base_retriever=retriever )

Advanced RAG with multiple retrieval strategies

def multi_retrieval_rag(question): # Similarity search similarity_docs = vectorstore.similarity_search(question, k=3)

# MMR (Maximum Marginal Relevance) search
mmr_docs = vectorstore.max_marginal_relevance_search(question, k=3)

# Combine and deduplicate
all_docs = similarity_docs + mmr_docs
unique_docs = list(\\\\{doc.page_content: doc for doc in all_docs\\\\}.values())

# Format context
context = format_docs(unique_docs)

# Generate answer
prompt = rag_prompt.format(context=context, question=question)
response = llm.invoke(prompt)

return response.content

result = multi_retrieval_rag("What are the key benefits mentioned?") ```_

RAG mit Metadatenfilterung

```python from langchain_core.documents import Document

Documents with metadata

documents_with_metadata = [ Document( page_content="LangChain is a framework for LLM applications", metadata=\\{"source": "documentation", "category": "framework", "date": "2024"\\} ), Document( page_content="RAG combines retrieval with generation", metadata=\\{"source": "tutorial", "category": "technique", "date": "2024"\\} ) ]

Create vector store with metadata

vectorstore_with_metadata = FAISS.from_documents( documents_with_metadata, embeddings )

Filtered retrieval

def filtered_rag(question, filter_criteria): # Retrieve with metadata filtering docs = vectorstore_with_metadata.similarity_search( question, k=5, filter=filter_criteria )

context = format_docs(docs)
prompt = rag_prompt.format(context=context, question=question)
response = llm.invoke(prompt)

return response.content

Query with filtering

result = filtered_rag( "What is LangChain?", \\{"category": "framework"\\} ) ```_

Produktionsentwicklung und -überwachung

LangSmith Integration

```python import os from langsmith import Client from langchain.callbacks import LangChainTracer

Set up LangSmith

os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key" os.environ["LANGCHAIN_PROJECT"] = "your-project-name"

Initialize client

client = Client()

Chain with tracing

traced_chain = ( prompt |llm.with_config(\\{"tags": ["production", "v1.0"]\\}) |StrOutputParser() )

Execute with automatic tracing

result = traced_chain.invoke( \\{"input": "test query"\\}, config=\\{"metadata": \\{"user_id": "user123", "session_id": "session456"\\}\\} ) ```_

Kundenspezifische Rückrufe und Überwachung

```python from langchain.callbacks.base import BaseCallbackHandler from typing import Dict, Any, List import logging import time

class CustomCallbackHandler(BaseCallbackHandler): """Custom callback for monitoring and logging"""

def __init__(self):
    self.start_time = None
    self.logger = logging.getLogger(__name__)

def on_chain_start(
    self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any
) -> Any:
    self.start_time = time.time()
    self.logger.info(f"Chain started with inputs: \\\\{inputs\\\\}")

def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> Any:
    duration = time.time() - self.start_time
    self.logger.info(f"Chain completed in \\\\{duration:.2f\\\\}s")
    self.logger.info(f"Outputs: \\\\{outputs\\\\}")

def on_chain_error(self, error: Exception, **kwargs: Any) -> Any:
    self.logger.error(f"Chain error: \\\\{error\\\\}")

def on_llm_start(
    self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
) -> Any:
    self.logger.info(f"LLM called with prompts: \\\\{prompts\\\\}")

def on_llm_end(self, response, **kwargs: Any) -> Any:
    self.logger.info(f"LLM response: \\\\{response\\\\}")

Use custom callback

custom_callback = CustomCallbackHandler() monitored_chain = chain.with_config(\\{"callbacks": [custom_callback]\\}) ```_

Fehler Handling und Widerstandsfähigkeit

```python from langchain_core.runnables import RunnableRetry from langchain.schema import OutputParserException import time

Retry configuration

retry_chain = RunnableRetry( bound=chain, max_attempt_number=3, wait_exponential_jitter=True )

Custom error handling

def handle_chain_errors(func): def wrapper(args, kwargs): max_retries = 3 for attempt in range(max_retries): try: return func(args, **kwargs) except OutputParserException as e: print(f"Parsing error on attempt \\{attempt + 1\\}: \\{e\\}") if attempt == max_retries - 1: raise time.sleep(2 ** attempt) except Exception as e: print(f"Unexpected error: \\{e\\}") if attempt == max_retries - 1: raise time.sleep(2 ** attempt) return wrapper

@handle_chain_errors def robust_chain_execution(input_data): return chain.invoke(input_data) ```_

Caching und Leistungsoptimierung

```python from langchain.cache import InMemoryCache, SQLiteCache from langchain.globals import set_llm_cache import sqlite3

In-memory caching

set_llm_cache(InMemoryCache())

SQLite caching for persistence

set_llm_cache(SQLiteCache(database_path=".langchain.db"))

Custom caching implementation

class RedisCache: def init(self, redis_client): self.redis_client = redis_client

def lookup(self, prompt, llm_string):
    key = f"\\\\{llm_string\\\\}:\\\\{hash(prompt)\\\\}"
    return self.redis_client.get(key)

def update(self, prompt, llm_string, return_val):
    key = f"\\\\{llm_string\\\\}:\\\\{hash(prompt)\\\\}"
    self.redis_client.set(key, return_val, ex=3600)  # 1 hour expiry

Performance monitoring

class PerformanceMonitor: def init(self): self.metrics = \\{\\}

def track_execution(self, chain_name, execution_time, token_count):
    if chain_name not in self.metrics:
        self.metrics[chain_name] = []

    self.metrics[chain_name].append(\\\\{
        "execution_time": execution_time,
        "token_count": token_count,
        "timestamp": time.time()
    \\\\})

def get_average_metrics(self, chain_name):
    if chain_name not in self.metrics:
        return None

    data = self.metrics[chain_name]
    avg_time = sum(d["execution_time"] for d in data) / len(data)
    avg_tokens = sum(d["token_count"] for d in data) / len(data)

    return \\\\{"avg_time": avg_time, "avg_tokens": avg_tokens\\\\}

monitor = PerformanceMonitor() ```_

Best Practices und Muster

Kettendesignprinzipien

  • ** Modularität**: Wiederverwendbare Kettenkomponenten bauen
  • Error Handling: Implementieren Sie robuste Fehlerbehandlung und Retries
  • Monitoring: Fügen Sie umfassende Protokollierung und Überwachung hinzu
  • Testing: Einheitstests für Kettenkomponenten erstellen
  • ** Aussprache**: Dokumentkettenverhalten und erwartete Ein-/Ausgänge

Leistungsoptimierung

  • Caching: Umsetzung geeigneter Caching-Strategien
  • Batching: Batch-Verarbeitung für mehrere Anfragen verwenden
  • Streaming: Implementieren Sie Streaming für Echtzeit-Reaktionen
  • Async: Verwenden Sie Async-Operationen für bessere Konkurrenz
  • Resource Management: Überwachen und verwalten API-Ratengrenzen

Sicherheitsüberlegungen

  • *Input Validation: Gültig und gesund
  • API Schlüsselmanagement: Sichere API-Schlüsselspeicherung und Rotation
  • Datenschutz*: Umsetzung von Datenschutz- und Aufbewahrungsrichtlinien
  • Access Control: Durchführung einer ordnungsgemäßen Authentifizierung und Autorisierung
  • *Audit Logging: Vollständige Auditprotokolle halten

Teststrategien

```python import pytest from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI

class TestLangChainComponents: def setup_method(self): self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) self.prompt = ChatPromptTemplate.from_template("Test prompt: \\{input\\}") self.chain = self.prompt|self.llm

def test_chain_basic_functionality(self):
    result = self.chain.invoke(\\\\{"input": "hello"\\\\})
    assert result is not None
    assert len(result.content) > 0

def test_chain_with_invalid_input(self):
    with pytest.raises(KeyError):
        self.chain.invoke(\\\\{"wrong_key": "value"\\\\})

def test_chain_performance(self):
    import time
    start_time = time.time()
    result = self.chain.invoke(\\\\{"input": "performance test"\\\\})
    execution_time = time.time() - start_time
    assert execution_time ``< 10  # Should complete within 10 seconds

@pytest.mark.asyncio
async def test_async_chain(self):
    result = await self.chain.ainvoke(\\\{"input": "async test"\\\})
    assert result is not None

```_

Probleme bei der Fehlerbehebung

API Rate Limits

```python from langchain.llms.base import LLM import time import random

class RateLimitedLLM(LLM): def init(self, base_llm, max_requests_per_minute=60): self.base_llm = base_llm self.max_requests_per_minute = max_requests_per_minute self.request_times = []

def _call(self, prompt, stop=None, run_manager=None):
    self._wait_if_needed()
    return self.base_llm._call(prompt, stop, run_manager)

def _wait_if_needed(self):
    now = time.time()
    # Remove requests older than 1 minute
    self.request_times = [t for t in self.request_times if now - t < 60]

    if len(self.request_times) >``= self.max_requests_per_minute:
        sleep_time = 60 - (now - self.request_times[0]) + random.uniform(1, 3)
        time.sleep(sleep_time)

    self.request_times.append(now)

@property
def _llm_type(self):
    return "rate_limited"

```_

Speicherprobleme

```python

Memory cleanup utilities

def cleanup_memory(memory_object): """Clean up memory to prevent memory leaks""" if hasattr(memory_object, 'clear'): memory_object.clear()

# Force garbage collection
import gc
gc.collect()

Memory usage monitoring

import psutil import os

def monitor_memory_usage(): process = psutil.Process(os.getpid()) memory_info = process.memory_info() return \\{ "rss": memory_info.rss / 1024 / 1024, # MB "vms": memory_info.vms / 1024 / 1024 # MB \\} ```_

Debugging Chain Execution

```python from langchain.globals import set_debug, set_verbose

Enable debugging

set_debug(True) set_verbose(True)

Custom debug callback

class DebugCallback(BaseCallbackHandler): def on_chain_start(self, serialized, inputs, **kwargs): print(f"🔗 Chain started: \\{serialized.get('name', 'Unknown')\\}") print(f"📥 Inputs: \\{inputs\\}")

def on_llm_start(self, serialized, prompts, **kwargs):
    print(f"🤖 LLM called with prompts:")
    for i, prompt in enumerate(prompts):
        print(f"  Prompt \\\\{i\\\\}: \\\\{prompt[:100]\\\\}...")

def on_tool_start(self, serialized, input_str, **kwargs):
    print(f"🔧 Tool called: \\\\{serialized.get('name', 'Unknown')\\\\}")
    print(f"📥 Input: \\\\{input_str\\\\}")

debug_chain = chain.with_config(\\{"callbacks": [DebugCallback()]\\}) ```_

--

*Dieses umfassende LangChain-Cheatsheet umfasst alles vom Grundaufbau bis hin zu fortschrittlichen Produktionsmustern. Verwenden Sie diese Beispiele und Best Practices, um robuste, skalierbare LLM-Anwendungen mit LangChains leistungsfähigem Rahmen aufzubauen. *