Hey folks! I’m struggling to integrate LangSmith monitoring into my FastAPI project and could really use some guidance. My setup doesn’t use any LLMs - I’m working with a third-party API from RapidAPI instead. The workflow involves uploading documents through FastAPI endpoints.
I’m trying to add tracking capabilities but can’t figure out how to properly implement LangSmith with my current architecture. The main confusion is around monitoring the document processing pipeline and API calls.
Here’s my current implementation. My app processes PDF and DOCX files, extracts content using LangChain loaders, then sends that content to an external service for analysis.
Any suggestions would be amazing - I’m really stuck on this one!
from fastapi import FastAPI, UploadFile, File
import requests
import tempfile
import os
server = FastAPI()
# Configuration
RAPID_API_TOKEN = <my_token>
API_ENDPOINT = <service_url>
# Function to analyze content
def analyze_content(content_text):
endpoint = <analysis_url>
request_data = {
"content": content_text,
"sensitivity": 15
}
request_headers = {
"x-rapidapi-key": RAPID_API_TOKEN,
"x-rapidapi-host": API_ENDPOINT,
"Content-Type": "application/json"
}
api_response = requests.post(endpoint, json=request_data, headers=request_headers)
return api_response.json()
@server.get('/status')
async def get_status():
return {"status": "Document Analysis Service Running"}
@server.post("/process-document/")
async def handle_file_upload(uploaded_file: UploadFile = File(...)):
file_extension = os.path.splitext(uploaded_file.filename)[1].lower()
temp_path = tempfile.mktemp(suffix=file_extension)
with open(temp_path, 'wb') as temp_file:
temp_file.write(await uploaded_file.read())
try:
if file_extension == '.pdf':
pdf_loader = PyPDFLoader(temp_path)
pages = pdf_loader.load_and_split()
extracted_text = " ".join([str(page) for page in pages])
elif file_extension == '.docx':
doc_loader = UnstructuredFileLoader(temp_path)
content = doc_loader.load()
extracted_text = str(content)
else:
return {"error": "File format not supported"}
analysis_result = analyze_content(extracted_text)
return analysis_result
except Exception as error:
return {"error": str(error)}
finally:
os.remove(temp_path)
if __name__ == "__main__":
import uvicorn
uvicorn.run(server, host="0.0.0.0", port=8080)
langsmith setup’s easy once you figure it out. install the langsmith package and set your project name in the env variables. the @traceable decorator works great, but context managers give you more control over tracking. i’ve used it for api monitoring - catches errors well too.
Had the same issue recently - needed to monitor document processing without LLMs. Skip the decorators and use LangSmith’s Client directly for custom traces. Way better control.
Create a parent trace for the whole document process, then separate child traces for extraction and API calls. Start a trace at your endpoint, then log file size, processing time, and API response status. This setup helped me catch that certain doc types were timing out with external APIs.
LangSmith’s filtering is clutch when you’re handling tons of documents. Filter by file type, processing time, or errors to spot patterns fast.
Client approach needs more setup than decorators, but you’ll get much better visibility into third-party API performance - which sounds like exactly what you need.
I ran into the same thing building a document processing pipeline at work. Here’s the thing with LangSmith - you don’t need LLMs to use it for monitoring.
Just wrap your main processing functions with LangSmith’s traceable decorators. This worked for me:
from langsmith import traceable
import os
# Set your LangSmith API key
os.environ["LANGSMITH_API_KEY"] = "your_api_key"
os.environ["LANGSMITH_PROJECT"] = "document-processing"
@traceable(name="document_extraction")
def extract_document_content(file_path, file_extension):
if file_extension == '.pdf':
pdf_loader = PyPDFLoader(file_path)
pages = pdf_loader.load_and_split()
return " ".join([str(page) for page in pages])
elif file_extension == '.docx':
doc_loader = UnstructuredFileLoader(file_path)
content = doc_loader.load()
return str(content)
@traceable(name="rapidapi_analysis")
def analyze_content(content_text):
# your existing analyze_content function
# LangSmith will automatically track the execution time and errors
endpoint = <analysis_url>
request_data = {
"content": content_text,
"sensitivity": 15
}
# rest of your code
In your FastAPI endpoint, just call these wrapped functions normally. LangSmith tracks everything automatically - timing, inputs, outputs, and errors.
Biggest win for me was seeing which documents took forever to process and spotting bottlenecks in the RapidAPI calls.
Look, LangSmith works for monitoring, but you’re overcomplicating things. I’ve dealt with similar document processing workflows - manual tracing gets messy fast.
You need proper automation for this pipeline. Don’t bolt monitoring onto your existing FastAPI setup. Build it as an automated workflow from day one.
I’d approach this differently - create a workflow that handles file uploads, processes documents, calls your RapidAPI endpoint, and tracks everything automatically. No decorators, no manual tracing setup.
Workflows give you built-in monitoring, error handling, and retry logic. You can easily add steps like file validation, content preprocessing, or result formatting without touching your core logic.
Document processing becomes way more reliable when you design it as connected steps instead of individual functions you’re monitoring after the fact.
For this automation, Latenode handles the orchestration and gives you monitoring without setup headaches.