Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Quelldateien prüfen
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
# Core installation
pip install langchain>=0.2.0 langchain-openai>=0.1.0 langchain-core>=0.2.0
# For RAG applications
pip install chromadb>=0.4.0 faiss-cpu>=1.7.0
# For document processing
pip install unstructured>=0.10.0 pypdf>=3.0.0
# For web search and tools
pip install duckduckgo-search wikipedia arxiv
# Optional: Local LLMs
pip install langchain-community ollama
# Environment setupexport OPENAI_API_KEY="sk-..."export ANTHROPIC_API_KEY="sk-ant-..."
Core Capabilities
1. Basic Chain Composition
Simple LLM Chain:
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
defcreate_simple_chain(
model: str = "gpt-4",
temperature: float = 0.7):
"""
Create a simple prompt-model-output chain.
Args:
model: Model name to use
temperature: Sampling temperature
Returns:
Runnable chain that accepts dict input
"""# Define prompt template
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant specializing in {domain}."),
("human", "{question}")
])
# Initialize LLM
llm = ChatOpenAI(model=model, temperature=temperature)
# Create chain with LCEL (LangChain Expression Language)
chain = prompt | llm | StrOutputParser()
return chain
# Usage
chain = create_simple_chain(model="gpt-4", temperature=0.3)
response = chain.invoke({
"domain": "marine engineering",
"question": "What are the key factors in mooring system design?"
})
print(response)
Sequential Chain with Multiple Steps:
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
defcreate_analysis_chain():
"""
Create a multi-step analysis chain:
1. Extract key points
2. Analyze implications
3. Generate recommendations
"""
llm = ChatOpenAI(model="gpt-4", temperature=0.3)
# Step 1: Extract key points
extract_prompt = ChatPromptTemplate.from_template(
"Extract the 5 most important points from this text:\n\n{text}\n\nKey Points:"
)
# Step 2: Analyze implications
analyze_prompt = ChatPromptTemplate.from_template(
"Based on these key points:\n{key_points}\n\n""What are the main implications and potential risks?"
)
# Step 3: Generate recommendations
recommend_prompt = ChatPromptTemplate.from_template(
"Given these key points:\n{key_points}\n\n""And this analysis:\n{analysis}\n\n""Provide 3-5 actionable recommendations."
)
# Build chain
chain = (
{"text": RunnablePassthrough()}
| RunnableParallel(
text=RunnablePassthrough(),
key_points=extract_prompt | llm | StrOutputParser()
)
| RunnableParallel(
key_points=lambda x: x["key_points"],
analysis=analyze_prompt | llm | StrOutputParser()
)
| recommend_prompt
| llm
| StrOutputParser()
)
return chain
# Usage
analysis_chain = create_analysis_chain()
document_text = """
The offshore wind farm project faces several challenges including
supply chain delays, regulatory approval processes, and environmental
impact assessments. Budget overruns of 15% have been reported...
"""
recommendations = analysis_chain.invoke(document_text)
print(recommendations)
2. Agent with Tools
ReAct Agent with Custom Tools:
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import tool
from langchain import hub
from typing importOptionalimport requests
import json
@tooldefcalculate_mooring_tension(
depth: float,
line_length: float,
pretension: float,
offset: float) -> str:
"""
Calculate approximate mooring line tension given parameters.
Args:
depth: Water depth in meters
line_length: Mooring line length in meters
pretension: Initial pretension in kN
offset: Horizontal vessel offset in meters
Returns:
Tension calculation result
"""# Simplified catenary calculationimport math
suspended_length = math.sqrt(line_length**2 - depth**2)
stretch_factor = 1 + (offset / suspended_length) * 0.1
tension = pretension * stretch_factor
return json.dumps({
"horizontal_tension_kN": round(tension, 2),
"vertical_tension_kN": round(tension * (depth / line_length), 2),
"line_angle_deg": round(math.degrees(math.asin(depth / line_length)), 1)
})
@tooldefget_wave_data(location: str, date: Optional[str] = None) -> str:
"""
Get wave condition data for a location.
Args:
location: Location name or coordinates
date: Date in YYYY-MM-DD format (optional)
Returns:
Wave data including Hs, Tp, direction
"""# Simulated data - replace with actual API call
wave_data = {
"location": location,
"significant_wave_height_m": 2.5,
"peak_period_s": 8.5,
"wave_direction_deg": 225,
"data_source": "simulated"
}
return json.dumps(wave_data)
@tooldefsearch_engineering_database(query: str) -> str:
"""
Search the engineering standards database.
Args:
query: Search query for standards/specifications
Returns:
Relevant standards and references
"""# Simulated database - replace with actual search
results = {
"query": query,
"results": [
{"standard": "API RP 2SK", "title": "Design and Analysis of Stationkeeping Systems"},
{"standard": "DNV-OS-E301", "title": "Position Mooring"},
{"standard": "ISO 19901-7", "title": "Stationkeeping systems"}
]
}
return json.dumps(results)
defcreate_engineering_agent():
"""
Create an agent with engineering-specific tools.
"""# Initialize LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)
# Define tools
tools = [
calculate_mooring_tension,
get_wave_data,
search_engineering_database
]
# Get ReAct prompt from hub
prompt = hub.pull("hwchase17/react")
# Create agent
agent = create_react_agent(llm, tools, prompt)
# Create executor with error handling
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
handle_parsing_errors=True,
max_iterations=5
)
return agent_executor
# Usage
agent = create_engineering_agent()
response = agent.invoke({
"input": """
I need to analyze a mooring system in 100m water depth.
The lines are 350m long with 500kN pretension.
What would be the tension if the vessel offset is 15m?
Also, what standards should I reference?
"""
})
print(response["output"])
Tool Agent with Structured Output:
from langchain_openai import ChatOpenAI
from langchain.agents import create_structured_chat_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from pydantic import BaseModel, Field
from typing importListclassAnalysisResult(BaseModel):
"""Structured analysis result."""
summary: str = Field(description="Brief summary of findings")
key_findings: List[str] = Field(description="List of key findings")
recommendations: List[str] = Field(description="List of recommendations")
risk_level: str = Field(description="Risk level: low, medium, or high")
@tooldefanalyze_document(document_path: str) -> str:
"""
Analyze an engineering document and extract key information.
Args:
document_path: Path to the document
Returns:
Extracted information from document
"""# Simulated document analysisreturn"""
Document: Mooring Analysis Report
Key findings:
- Maximum tension: 2500 kN (within limits)
- Safety factor: 1.8 (above minimum 1.67)
- Fatigue life: 45 years (design life: 25 years)
Recommendations:
- Monitor chain condition at fairlead
- Consider dynamic analysis for extreme conditions
"""defcreate_structured_agent():
"""Create agent that returns structured output."""
llm = ChatOpenAI(model="gpt-4", temperature=0)
tools = [analyze_document]
system_prompt = """You are an engineering analysis assistant.
Use the available tools to analyze documents and provide structured insights.
Always provide your final answer in a structured format with:
- summary
- key_findings (list)
- recommendations (list)
- risk_level
{tools}
{tool_names}
"""
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
MessagesPlaceholder("chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder("agent_scratchpad")
])
agent = create_structured_chat_agent(llm, tools, prompt)
return AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
handle_parsing_errors=True
)
# Usage
structured_agent = create_structured_agent()
result = structured_agent.invoke({
"input": "Analyze the mooring report at /data/mooring_report.pdf"
})
3. Conversation Memory
Conversation Buffer Memory:
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
# Store for session histories
store = {}
defget_session_history(session_id: str) -> BaseChatMessageHistory:
"""Get or create message history for a session."""if session_id notin store:
store[session_id] = ChatMessageHistory()
return store[session_id]
defcreate_conversational_chain():
"""
Create a chain with conversation memory.
"""
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
prompt = ChatPromptTemplate.from_messages([
("system", """You are an expert offshore engineering consultant.
You help with mooring design, vessel dynamics, and marine operations.
Maintain context from previous messages in the conversation."""),
MessagesPlaceholder(variable_name="history"),
("human", "{input}")
])
chain = prompt | llm
# Wrap with message history
chain_with_history = RunnableWithMessageHistory(
chain,
get_session_history,
input_messages_key="input",
history_messages_key="history"
)
return chain_with_history
# Usage
conversational_chain = create_conversational_chain()
# First message
response1 = conversational_chain.invoke(
{"input": "I'm designing a spread mooring system for a 100,000 DWT tanker."},
config={"configurable": {"session_id": "project-123"}}
)
print(f"Assistant: {response1.content}")
# Follow-up (remembers context)
response2 = conversational_chain.invoke(
{"input": "What line configuration would you recommend?"},
config={"configurable": {"session_id": "project-123"}}
)
print(f"Assistant: {response2.content}")
# Check history
history = get_session_history("project-123")
print(f"\nConversation has {len(history.messages)} messages")
Summary Memory for Long Conversations:
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationSummaryBufferMemory
from langchain.chains import ConversationChain
defcreate_summary_memory_chain():
"""
Create chain with summary memory for long conversations.
Keeps recent messages verbatim, summarizes older ones.
"""
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
# Summary buffer keeps last 1000 tokens verbatim
memory = ConversationSummaryBufferMemory(
llm=llm,
max_token_limit=1000,
return_messages=True
)
chain = ConversationChain(
llm=llm,
memory=memory,
verbose=True
)
return chain, memory
# Usage
chain, memory = create_summary_memory_chain()
# Simulate long conversation
responses = []
questions = [
"What are the main types of mooring systems?",
"Tell me about spread moorings in detail.",
"What about single point moorings?",
"How do turret moorings work?",
"Compare the maintenance requirements.",
"What are the cost implications?"
]
for q in questions:
response = chain.predict(input=q)
responses.append(response)
print(f"Q: {q}")
print(f"A: {response[:200]}...")
print()
# Check memory stateprint("Memory Summary:")
print(memory.moving_summary_buffer)
4. RAG (Retrieval Augmented Generation)
Complete RAG Pipeline:
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from pathlib import Path
from typing importListdefcreate_rag_pipeline(
documents_dir: str,
collection_name: str = "engineering_docs",
chunk_size: int = 1000,
chunk_overlap: int = 200):
"""
Create a complete RAG pipeline.
Args:
documents_dir: Directory containing documents
collection_name: Name for vector store collection
chunk_size: Size of text chunks
chunk_overlap: Overlap between chunks
Returns:
RAG chain for question answering
"""# 1. Load documents
loader = DirectoryLoader(
documents_dir,
glob="**/*.pdf",
loader_cls=PyPDFLoader,
show_progress=True
)
documents = loader.load()
print(f"Loaded {len(documents)} document pages")
# 2. Split documents into chunks
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len,
separators=["\n\n", "\n", " ", ""]
)
chunks = text_splitter.split_documents(documents)
print(f"Created {len(chunks)} chunks")
# 3. Create embeddings and vector store
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
collection_name=collection_name,
persist_directory="./chroma_db"
)
# 4. Create retriever
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 5}
)
# 5. Create RAG prompt
rag_prompt = ChatPromptTemplate.from_template("""
You are an expert assistant answering questions based on the provided context.
Use only the information from the context to answer.
If the context doesn't contain the answer, say "I don't have enough information."
Context:
{context}
Question: {question}
Answer:
""")
# 6. Create LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)
# 7. Build RAG chaindefformat_docs(docs):
return"\n\n---\n\n".join(
f"Source: {doc.metadata.get('source', 'Unknown')}\n{doc.page_content}"for doc in docs
)
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| rag_prompt
| llm
| StrOutputParser()
)
return rag_chain, retriever
# Usage
rag_chain, retriever = create_rag_pipeline(
documents_dir="./engineering_docs",
collection_name="offshore_standards"
)
# Query
answer = rag_chain.invoke(
"What are the safety factor requirements for mooring lines?"
)
print(answer)
# Get source documents
docs = retriever.get_relevant_documents(
"mooring line safety factors"
)
for doc in docs:
print(f"Source: {doc.metadata['source']}")
print(f"Content: {doc.page_content[:200]}...")
print()
RAG with Reranking:
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
defcreate_reranked_rag_pipeline(
vectorstore: Chroma,
top_k_initial: int = 20,
top_k_final: int = 5):
"""
Create RAG pipeline with reranking for better relevance.
Args:
vectorstore: Existing vector store
top_k_initial: Number of docs to retrieve initially
top_k_final: Number of docs after reranking
"""# Base retriever - get more docs initially
base_retriever = vectorstore.as_retriever(
search_kwargs={"k": top_k_initial}
)
# Reranker using cross-encoder
reranker_model = HuggingFaceCrossEncoder(
model_name="cross-encoder/ms-marco-MiniLM-L-6-v2"
)
compressor = CrossEncoderReranker(
model=reranker_model,
top_n=top_k_final
)
# Compression retriever with reranking
retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=base_retriever
)
# Build chain
llm = ChatOpenAI(model="gpt-4", temperature=0)
prompt = ChatPromptTemplate.from_template("""
Answer the question based on the context below.
Cite your sources by mentioning which document the information came from.
Context:
{context}
Question: {question}
Answer with citations:
""")
defformat_docs_with_citations(docs):
formatted = []
for i, doc inenumerate(docs, 1):
source = doc.metadata.get("source", "Unknown")
formatted.append(f"[{i}] {source}:\n{doc.page_content}")
return"\n\n".join(formatted)
chain = (
{"context": retriever | format_docs_with_citations, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
return chain
# Usage# Assuming vectorstore already exists
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
persist_directory="./chroma_db",
embedding_function=embeddings
)
reranked_chain = create_reranked_rag_pipeline(vectorstore)
answer = reranked_chain.invoke(
"What are the fatigue analysis requirements for mooring chains?"
)
print(answer)
5. Document Processing
Multi-Format Document Loader:
from langchain_community.document_loaders import (
DirectoryLoader,
PyPDFLoader,
Docx2txtLoader,
UnstructuredExcelLoader,
TextLoader,
CSVLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from pathlib import Path
from typing importList, Dictfrom dataclasses import dataclass
@dataclassclassLoaderConfig:
"""Configuration for document loader."""
glob_pattern: str
loader_cls: type
loader_kwargs: dict = None
LOADER_CONFIGS = {
".pdf": LoaderConfig("**/*.pdf", PyPDFLoader),
".docx": LoaderConfig("**/*.docx", Docx2txtLoader),
".xlsx": LoaderConfig("**/*.xlsx", UnstructuredExcelLoader),
".csv": LoaderConfig("**/*.csv", CSVLoader),
".txt": LoaderConfig("**/*.txt", TextLoader),
".md": LoaderConfig("**/*.md", TextLoader),
}
defload_documents_multi_format(
directory: str,
extensions: List[str] = None,
chunk_size: int = 1000,
chunk_overlap: int = 200) -> List:
"""
Load documents from multiple formats.
Args:
directory: Base directory for documents
extensions: List of extensions to load (None = all)
chunk_size: Size of text chunks
chunk_overlap: Overlap between chunks
Returns:
List of chunked documents
"""if extensions isNone:
extensions = list(LOADER_CONFIGS.keys())
all_documents = []
for ext in extensions:
if ext notin LOADER_CONFIGS:
print(f"Warning: No loader for {ext}")
continue
config = LOADER_CONFIGS[ext]
try:
loader = DirectoryLoader(
directory,
glob=config.glob_pattern,
loader_cls=config.loader_cls,
loader_kwargs=config.loader_kwargs or {},
show_progress=True
)
docs = loader.load()
print(f"Loaded {len(docs)} documents with extension {ext}")
all_documents.extend(docs)
except Exception as e:
print(f"Error loading {ext} files: {e}")
# Split into chunks
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap
)
chunks = text_splitter.split_documents(all_documents)
print(f"Total: {len(all_documents)} documents -> {len(chunks)} chunks")
return chunks
defadd_metadata_to_chunks(
chunks: List,
additional_metadata: Dict = None) -> List:
"""
Enrich document chunks with additional metadata.
"""for chunk in chunks:
# Extract filename without path
source = chunk.metadata.get("source", "")
chunk.metadata["filename"] = Path(source).name
chunk.metadata["extension"] = Path(source).suffix
# Add custom metadataif additional_metadata:
chunk.metadata.update(additional_metadata)
# Add chunk stats
chunk.metadata["chunk_length"] = len(chunk.page_content)
chunk.metadata["word_count"] = len(chunk.page_content.split())
return chunks
# Usage
chunks = load_documents_multi_format(
directory="./project_docs",
extensions=[".pdf", ".docx", ".md"],
chunk_size=1500,
chunk_overlap=300
)
enriched_chunks = add_metadata_to_chunks(
chunks,
additional_metadata={
"project": "Offshore Platform Analysis",
"version": "2.0"
}
)
print(f"First chunk metadata: {enriched_chunks[0].metadata}")
6. Streaming Responses
Streaming Chain:
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import asyncio
defcreate_streaming_chain():
"""Create chain that supports streaming output."""
llm = ChatOpenAI(
model="gpt-4",
temperature=0.7,
streaming=True
)
prompt = ChatPromptTemplate.from_template(
"You are an expert engineer. Explain in detail: {topic}"
)
chain = prompt | llm | StrOutputParser()
return chain
# Synchronous streamingdefstream_response(chain, topic: str):
"""Stream response token by token."""print("Response: ", end="", flush=True)
for chunk in chain.stream({"topic": topic}):
print(chunk, end="", flush=True)
print("\n")
# Async streamingasyncdefastream_response(chain, topic: str):
"""Async stream response."""print("Response: ", end="", flush=True)
asyncfor chunk in chain.astream({"topic": topic}):
print(chunk, end="", flush=True)
print("\n")
# Usage
chain = create_streaming_chain()
# Sync streaming
stream_response(chain, "mooring line catenary equations")
# Async streaming
asyncio.run(astream_response(chain, "wave-structure interaction"))
Streaming with Callbacks:
from langchain_openai import ChatOpenAI
from langchain_core.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from langchain_core.callbacks.base import BaseCallbackHandler
from typing importAny, Dict, Listimport json
classCustomStreamHandler(BaseCallbackHandler):
"""Custom callback handler for streaming."""def__init__(self):
self.tokens = []
self.final_response = ""defon_llm_new_token(self, token: str, **kwargs) -> None:
"""Called when a new token is generated."""self.tokens.append(token)
self.final_response += token
# Could send to WebSocket, write to file, etc.print(token, end="", flush=True)
defon_llm_end(self, response: Any, **kwargs) -> None:
"""Called when LLM finishes."""print(f"\n\n[Generation complete: {len(self.tokens)} tokens]")
defon_llm_error(self, error: Exception, **kwargs) -> None:
"""Called on error."""print(f"\n[Error: {error}]")
defstream_with_callbacks(prompt: str):
"""Stream with custom callbacks."""
handler = CustomStreamHandler()
llm = ChatOpenAI(
model="gpt-4",
temperature=0.7,
streaming=True,
callbacks=[handler]
)
response = llm.invoke(prompt)
return handler.final_response, handler.tokens
# Usage
response, tokens = stream_with_callbacks(
"Explain the design considerations for a turret mooring system"
)
print(f"\nTotal tokens: {len(tokens)}")
Complete Examples
Example 1: Engineering Documentation Assistant
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from pathlib import Path
classEngineeringAssistant:
"""
Complete engineering documentation assistant with:
- RAG for document retrieval
- Conversation memory
- Streaming responses
"""def__init__(
self,
docs_directory: str,
persist_directory: str = "./chroma_db"):
self.docs_directory = docs_directory
self.persist_directory = persist_directory
self.session_store = {}
# Initialize componentsself.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
self.llm = ChatOpenAI(model="gpt-4", temperature=0.3, streaming=True)
# Setup RAGself.vectorstore = self._setup_vectorstore()
self.retriever = self.vectorstore.as_retriever(search_kwargs={"k": 5})
# Setup conversational chainself.chain = self._create_chain()
def_setup_vectorstore(self):
"""Load documents and create vector store."""
persist_path = Path(self.persist_directory)
# Check if vector store existsif persist_path.exists():
print("Loading existing vector store...")
return Chroma(
persist_directory=self.persist_directory,
embedding_function=self.embeddings
)
# Load and process documentsprint("Creating new vector store...")
loader = DirectoryLoader(
self.docs_directory,
glob="**/*.pdf",
loader_cls=PyPDFLoader
)
documents = loader.load()
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
chunks = text_splitter.split_documents(documents)
return Chroma.from_documents(
documents=chunks,
embedding=self.embeddings,
persist_directory=self.persist_directory
)
def_get_session_history(self, session_id: str):
"""Get or create session history."""if session_id notinself.session_store:
self.session_store[session_id] = ChatMessageHistory()
returnself.session_store[session_id]
def_create_chain(self):
"""Create the conversational RAG chain."""
prompt = ChatPromptTemplate.from_messages([
("system", """You are an expert engineering assistant.
Use the following context from engineering documents to answer questions.
Always cite your sources by mentioning the document name.
If you cannot find the answer in the context, say so.
Context:
{context}
"""),
MessagesPlaceholder(variable_name="history"),
("human", "{question}")
])
defformat_docs(docs):
return"\n\n---\n\n".join(
f"[{doc.metadata.get('source', 'Unknown')}]\n{doc.page_content}"for doc in docs
)
chain = (
RunnablePassthrough.assign(
context=lambda x: format_docs(self.retriever.get_relevant_documents(x["question"]))
)
| prompt
| self.llm
)
return RunnableWithMessageHistory(
chain,
self._get_session_history,
input_messages_key="question",
history_messages_key="history"
)
defask(self, question: str, session_id: str = "default"):
"""Ask a question with conversation context."""
response = self.chain.invoke(
{"question": question},
config={"configurable": {"session_id": session_id}}
)
return response.content
defstream_ask(self, question: str, session_id: str = "default"):
"""Ask a question with streaming response."""for chunk inself.chain.stream(
{"question": question},
config={"configurable": {"session_id": session_id}}
):
ifhasattr(chunk, 'content'):
yield chunk.content
defget_sources(self, question: str, k: int = 5):
"""Get source documents for a question."""
docs = self.retriever.get_relevant_documents(question)[:k]
return [
{
"source": doc.metadata.get("source"),
"content": doc.page_content[:500] + "..."
}
for doc in docs
]
# Usage
assistant = EngineeringAssistant(
docs_directory="./engineering_standards"
)
# Regular query
answer = assistant.ask(
"What is the minimum safety factor for permanent mooring lines?",
session_id="project-001"
)
print(answer)
# Follow-up (uses conversation history)
follow_up = assistant.ask(
"What about temporary moorings?",
session_id="project-001"
)
print(follow_up)
# Streaming queryprint("\nStreaming response:")
for chunk in assistant.stream_ask("Explain fatigue analysis methodology"):
print(chunk, end="", flush=True)
print()
# Get sources
sources = assistant.get_sources("mooring safety factors")
for source in sources:
print(f"\nSource: {source['source']}")
print(f"Content: {source['content']}")
Example 2: Multi-Tool Research Agent
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from langchain_community.tools import DuckDuckGoSearchRun, WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
from pydantic import BaseModel, Field
from typing importList, Optionalimport json
classResearchResult(BaseModel):
"""Structured research result."""
topic: str
summary: str
key_findings: List[str]
sources: List[str]
confidence: float@tooldefsave_research_note(
topic: str,
content: str,
tags: List[str]
) -> str:
"""
Save a research note for later reference.
Args:
topic: Research topic
content: Note content
tags: List of relevant tags
Returns:
Confirmation message
"""
note_id = hash(topic) % 10000# In production, save to databasereturnf"Note saved with ID {note_id}. Topic: {topic}, Tags: {', '.join(tags)}"@tooldefcalculate_statistics(
values: List[float],
operation: str = "summary") -> str:
"""
Calculate statistics on a list of values.
Args:
values: List of numeric values
operation: Type of calculation (summary, mean, std, range)
Returns:
Calculation results
"""import statistics
if operation == "summary":
result = {
"count": len(values),
"mean": statistics.mean(values),
"std": statistics.stdev(values) iflen(values) > 1else0,
"min": min(values),
"max": max(values),
"median": statistics.median(values)
}
elif operation == "mean":
result = {"mean": statistics.mean(values)}
elif operation == "std":
result = {"std": statistics.stdev(values) iflen(values) > 1else0}
elif operation == "range":
result = {"min": min(values), "max": max(values), "range": max(values) - min(values)}
else:
result = {"error": f"Unknown operation: {operation}"}
return json.dumps(result, indent=2)
defcreate_research_agent():
"""Create a multi-tool research agent."""
llm = ChatOpenAI(model="gpt-4", temperature=0)
# Initialize tools
search = DuckDuckGoSearchRun()
wikipedia = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
tools = [
search,
wikipedia,
save_research_note,
calculate_statistics
]
prompt = ChatPromptTemplate.from_messages([
("system", """You are a research assistant that helps gather and analyze information.
You have access to:
- Web search for current information
- Wikipedia for background knowledge
- Note-taking for saving important findings
- Statistics calculator for numerical analysis
When researching:
1. Search for relevant information
2. Verify facts from multiple sources
3. Save important findings as notes
4. Provide a comprehensive summary
Always cite your sources and be clear about uncertainty."""),
MessagesPlaceholder(variable_name="chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
agent = create_openai_tools_agent(llm, tools, prompt)
return AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=10,
handle_parsing_errors=True
)
# Usage
research_agent = create_research_agent()
result = research_agent.invoke({
"input": """
Research the current state of floating offshore wind technology.
I need to know:
1. What are the main platform types?
2. What are typical water depths for deployment?
3. What are the largest projects currently operating?
Save the key findings as research notes.
"""
})
print(result["output"])
Integration Patterns
LangServe Deployment
from fastapi import FastAPI
from langserve import add_routes
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
# Create app
app = FastAPI(
title="Engineering Assistant API",
version="1.0"
)
# Create chain
prompt = ChatPromptTemplate.from_template(
"You are an engineering expert. Answer: {question}"
)
llm = ChatOpenAI(model="gpt-4")
chain = prompt | llm | StrOutputParser()
# Add routes
add_routes(app, chain, path="/assistant")
# Run with: uvicorn server:app --reload
LangSmith Tracing
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
# Enable LangSmith tracing
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "engineering-assistant"# All chain invocations are now traced
chain = ChatPromptTemplate.from_template("{input}") | ChatOpenAI()
response = chain.invoke({"input": "Hello"})
Best Practices
1. Error Handling
from langchain_core.runnables import RunnableConfig
from langchain_core.callbacks import CallbackManager
import logging
logger = logging.getLogger(__name__)
defsafe_invoke(chain, input_data, max_retries=3):
"""Invoke chain with retry logic."""for attempt inrange(max_retries):
try:
return chain.invoke(input_data)
except Exception as e:
logger.warning(f"Attempt {attempt + 1} failed: {e}")
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt) # Exponential backoff
# Process documents in batchesdefbatch_process_documents(documents, batch_size=100):
for i inrange(0, len(documents), batch_size):
batch = documents[i:i + batch_size]
yield process_batch(batch)
Vector Store Performance
# Use FAISS for larger collectionsfrom langchain_community.vectorstores import FAISS
vectorstore = FAISS.from_documents(
documents,
embeddings,
distance_strategy="COSINE"
)
# Add index for faster retrieval
vectorstore.save_local("faiss_index")
