Primeiros Passos com LangChain em Python

Quando comecei a trabalhar com LLMs em aplicações reais, percebi rapidamente que gerenciar conversas, prompts e chamadas de API na mão vira um pesadelo rápido. Cada projeto reescrevia as mesmas abstrações de formas levemente diferentes. Foi quando o LangChain começou a fazer sentido pra mim — não como "mais um framework", mas como uma linguagem comum para problemas que todo mundo enfrenta.

Vou ser honesto: LangChain tem uma curva de aprendizado e às vezes parece fazer simples complicado. Mas para pipelines complexos com RAG, memória e agentes, o investimento vale. Neste tutorial vou mostrar o caminho que eu recomendaria para mim mesmo se começasse hoje. Depois de dominar o básico aqui, o próximo passo natural é construir um sistema RAG do zero.

LangChain é o framework mais popular para construir aplicações com LLMs (Large Language Models). Ele simplifica a criação de pipelines complexos como chatbots com memória, sistemas RAG e agentes autônomos, fornecendo abstrações que permitem trocar de provedor de IA (OpenAI, Anthropic, Google) sem reescrever a lógica da aplicação.

Sem LangChain, você seria responsável por gerenciar manualmente o histórico de conversas, construir prompts, fazer parsing das respostas e encadear múltiplas chamadas à API. O LangChain padroniza tudo isso em componentes reutilizáveis.

Requisitos

• Python 3.10 ou superior
• Chave de API da OpenAI (ou outro provedor compatível)

Instalação

pip install langchain langchain-openai langchain-community faiss-cpu

Configure sua chave de API:

export OPENAI_API_KEY="sk-..."

Ou usando .env:

pip install python-dotenv

from dotenv import load_dotenv
load_dotenv()  # Carrega variáveis do arquivo .env

1. Primeiro Chat

O componente mais básico do LangChain é o ChatOpenAI, que encapsula a API de chat:

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# Enviar uma mensagem simples
response = llm.invoke([
    SystemMessage(content="Você é um especialista em machine learning."),
    HumanMessage(content="O que é machine learning em uma frase?")
])

print(response.content)
# Machine learning é um campo da IA em que sistemas aprendem padrões
# a partir de dados para fazer previsões ou tomar decisões sem serem
# explicitamente programados para cada situação.

2. Templates de Prompt

Em vez de construir strings de prompt manualmente, o LangChain oferece ChatPromptTemplate — um componente reutilizável com variáveis:

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "Você é um professor de IA que explica conceitos de forma simples."),
    ("human", "Explique {conceito} para um iniciante.")
])

# Usar o operador | para criar chains (pipelines)
chain = prompt | llm

# Invocar a chain com os valores das variáveis
response = chain.invoke({"conceito": "redes neurais"})
print(response.content)

# Você pode reutilizar a mesma chain com conceitos diferentes
for conceito in ["overfitting", "gradient descent", "transformers"]:
    r = chain.invoke({"conceito": conceito})
    print(f"\n{conceito.upper()}:")
    print(r.content[:200] + "...")

O operador | (pipe) é a forma idiomática do LangChain para compor componentes — chamada de LCEL (LangChain Expression Language). Ele funciona como um pipeline Unix: a saída de um componente é passada como entrada para o próximo.

3. Output Parsers

Por padrão, a resposta é um objeto AIMessage. Para extrair tipos específicos, use output parsers:

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from pydantic import BaseModel, Field

# Parser de string simples
chain_str = prompt | llm | StrOutputParser()
texto = chain_str.invoke({"conceito": "backpropagation"})
print(type(texto))  # <class 'str'>

# Parser de JSON estruturado
class ExplicacaoConceito(BaseModel):
    conceito: str = Field(description="Nome do conceito explicado")
    definicao: str = Field(description="Definição em uma frase")
    exemplo: str = Field(description="Exemplo prático")
    nivel: str = Field(description="Nível de dificuldade: iniciante, intermediário, avançado")

json_parser = JsonOutputParser(pydantic_object=ExplicacaoConceito)

prompt_json = ChatPromptTemplate.from_messages([
    ("system", "Explique conceitos de IA em formato JSON estruturado."),
    ("human", "Explique o conceito de {conceito}.\n{format_instructions}"),
]).partial(format_instructions=json_parser.get_format_instructions())

chain_json = prompt_json | llm | json_parser
resultado = chain_json.invoke({"conceito": "dropout"})

print(f"Conceito: {resultado['conceito']}")
print(f"Definição: {resultado['definicao']}")
print(f"Exemplo: {resultado['exemplo']}")

4. Memória de Conversa

Para construir um chatbot que se lembra do contexto, usamos RunnableWithMessageHistory:

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

# Armazenar históricos por session_id
store = {}

def get_session_history(session_id: str) -> ChatMessageHistory:
    """Retorna ou cria o histórico de uma sessão."""
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# Criar chain com memória automática
prompt_com_historico = ChatPromptTemplate.from_messages([
    ("system", "Você é um assistente útil. Responda em português."),
    ("placeholder", "{history}"),
    ("human", "{input}"),
])

chain_com_memoria = RunnableWithMessageHistory(
    prompt_com_historico | llm | StrOutputParser(),
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

# Conversa com memória persistente
config = {"configurable": {"session_id": "usuario_123"}}

r1 = chain_com_memoria.invoke({"input": "Meu nome é João."}, config=config)
print(f"Bot: {r1}")

r2 = chain_com_memoria.invoke({"input": "Qual é meu nome?"}, config=config)
print(f"Bot: {r2}")
# Bot: Seu nome é João, como você me disse anteriormente.

r3 = chain_com_memoria.invoke({"input": "Explique overfitting em 2 frases."}, config=config)
print(f"Bot: {r3}")

5. RAG Básico (Retrieval-Augmented Generation)

RAG é uma técnica que permite ao modelo responder perguntas sobre documentos que não estavam no seu treinamento. O fluxo é: carrega documentos → divide em chunks → gera embeddings → armazena em vetor store → recupera chunks relevantes → gera resposta:

from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.chains import RetrievalQA
from langchain_core.prompts import PromptTemplate

# 1. Carregar documento
# Para este exemplo, vamos criar um documento de teste
with open("conhecimento.txt", "w") as f:
    f.write("""
    FastAPI é um framework web Python moderno, criado por Sebastián Ramírez.
    Foi lançado em 2018 e é baseado em Starlette e Pydantic.
    É amplamente usado para construir APIs REST e microsserviços.
    Suporta operações assíncronas nativamente usando async/await.
    Gera documentação Swagger automaticamente.
    Performance comparável a Node.js e Go para operações de I/O.
    """)

loader = TextLoader("conhecimento.txt", encoding="utf-8")
docs = loader.load()

# 2. Dividir em chunks menores
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", " ", ""]
)
chunks = splitter.split_documents(docs)
print(f"Documento dividido em {len(chunks)} chunks")

# 3. Criar embeddings e armazenar
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(chunks, embeddings)

# Salvar para uso posterior
vectorstore.save_local("meu_indice")

# 4. Criar prompt personalizado para RAG
rag_prompt = PromptTemplate.from_template("""
Use as informações abaixo para responder a pergunta.
Se não souber a resposta baseado no contexto, diga que não sabe.

Contexto:
{context}

Pergunta: {question}

Resposta:""")

# 5. Criar chain de perguntas
qa = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
    chain_type_kwargs={"prompt": rag_prompt},
    return_source_documents=True
)

# Consultar
perguntas = [
    "Quem criou o FastAPI?",
    "O FastAPI suporta operações assíncronas?",
    "Qual é a capital da França?",  # Pergunta fora do contexto
]

for pergunta in perguntas:
    result = qa.invoke({"query": pergunta})
    print(f"\nPergunta: {pergunta}")
    print(f"Resposta: {result['result']}")

Resultado Esperado

Pergunta: Quem criou o FastAPI?
Resposta: O FastAPI foi criado por Sebastián Ramírez.

Pergunta: O FastAPI suporta operações assíncronas?
Resposta: Sim, o FastAPI suporta operações assíncronas nativamente usando async/await.

Pergunta: Qual é a capital da França?
Resposta: Não tenho essa informação no contexto fornecido.

6. Streaming de Respostas

Para aplicações interativas, o streaming evita que o usuário espere a resposta completa:

from langchain_core.callbacks import StreamingStdOutCallbackHandler

llm_stream = ChatOpenAI(
    model="gpt-4o-mini",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

chain_stream = prompt | llm_stream | StrOutputParser()

# A resposta aparece token por token no terminal
print("Resposta (streaming):")
chain_stream.invoke({"conceito": "embeddings"})

Ou de forma programática (para usar em APIs):

async def resposta_streaming(conceito: str):
    """Gera resposta com streaming para uso em FastAPI."""
    async for chunk in chain.astream({"conceito": conceito}):
        yield chunk

# Em uma rota FastAPI:
# from fastapi.responses import StreamingResponse
# return StreamingResponse(resposta_streaming("transformers"), media_type="text/plain")

Boas Práticas

Prefira LCEL para novas aplicações: A LangChain Expression Language (| operator) é a forma moderna e recomendada de compor chains. A API antiga baseada em classes (ConversationChain, LLMChain) ainda funciona mas é considerada legada.

Cache de embeddings: gerar embeddings tem custo. Para documentos que não mudam, salve e recarregue o vetor store:

import os

if os.path.exists("meu_indice"):
    vectorstore = FAISS.load_local("meu_indice", embeddings, allow_dangerous_deserialization=True)
else:
    vectorstore = FAISS.from_documents(chunks, embeddings)
    vectorstore.save_local("meu_indice")

Rate limiting: ao processar muitos documentos, adicione delays para evitar erros de rate limit:

import time

for chunk in chunks:
    embeddings.embed_documents([chunk.page_content])
    time.sleep(0.1)  # 100ms entre chamadas

Próximos Passos

• Agentes com ferramentas: crie agentes que usam ferramentas (busca web, calculadora, APIs) para resolver tarefas que exigem múltiplos passos.
• RAG avançado: implemente re-ranking, HyDE (Hypothetical Document Embeddings) e retrieval com múltiplos vectorstores para melhorar a qualidade das respostas.
• LangSmith: ferramenta de observabilidade oficial do LangChain para debugar, monitorar e avaliar chains em produção.
• Deploy com LangServe: publique suas chains como APIs REST com um único decorador.

No próximo tutorial, vamos construir um chatbot RAG completo que responde perguntas sobre seus documentos com persistência em banco de dados e interface web!