ImapRetriever
本指南将帮助您开始使用 IMAP 检索器。 ImapRetriever 支持从 IMAP 服务器搜索和检索电子邮件,作为 LangChain Document objects.
集成详情
| 检索器 | 来源 | 包 |
|---|---|---|
ImapRetriever | IMAP 电子邮件服务器 | langchain-imap |
设置
安装
该 ImapRetriever 位于 langchain-imap package:
pip install -U langchain-imap
使用 docling 进行完整文档处理(DOCX、PPTX 等)(未测试):
pip install "langchain-imap[docling]"
测试环境设置(可选)
为了测试目的,您可以使用 GreenMail 设置本地 IMAP 服务器:
from pathlib import Path
preload_dir = Path(os.getcwd()).parent / "tests" / "fixtures" / "preload"
log_path = Path(os.getcwd()).parent / "tests" / "container.log"
# GreenMail configuration
env_vars = {
"GREENMAIL_OPTS": " ".join([
"-Dgreenmail.setup.test.all",
"-Dgreenmail.users=test:test123@localhost",
"-Dgreenmail.users.login=local_part",
"-Dgreenmail.preload.dir=/preload",
"-Dgreenmail.verbose",
"-Dgreenmail.hostname=0.0.0.0"
])
}
# Start GreenMail container
container_name = "langchain-imap-test"
cmd = [
"podman", "run", "--rm", "-d",
"--name", container_name,
"-e", f"GREENMAIL_OPTS={env_vars['GREENMAIL_OPTS']}",
"-v", f"{preload_dir}:/preload:ro,Z",
"-p", "3143:3143",
"-p", "3993:3993",
"-p", "8080:8080",
"--log-driver", "k8s-file",
"--log-opt", f"path={log_path.absolute()}",
"docker.io/greenmail/standalone:2.1.5",
]
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
实例化
要使用 ImapRetriever,您需要使用以下方式配置您的 IMAP 服务器详细信息 ImapConfig:
from langchain_imap import ImapConfig, ImapRetriever
config = ImapConfig(
host="imap.gmail.com",
port=993,
user="your-email@gmail.com",
password="your-app-password", # Use app password for Gmail
ssl_mode="ssl",
)
retriever = ImapRetriever(config=config, k=10)
对于测试环境:
from langchain_imap import ImapRetriever, ImapConfig
config = ImapConfig(
host="localhost",
port=3143,
user="test",
password="test123",
ssl_mode="plain",
verify_cert=False,
)
retriever = ImapRetriever(
config=config,
k=50
)
配置选项
- auth_method:认证方法(默认:"login")
- ssl_mode:SSL 模式 - "ssl"(默认)、"starttls" 或 "plain"
- verify_cert:设置为
False用于自签名证书(不推荐用于生产环境) - k:要检索的文档数量
用法
基本搜索
使用 IMAP 语法搜索电子邮件:
# Search all emails
query = 'ALL'
docs = retriever.invoke(query)
# Search by subject
query = 'SUBJECT "URGENT"'
docs = retriever.invoke(query)
# Search by sender
docs = retriever.invoke('FROM "john@example.com"')
# Search by date
docs = retriever.invoke('SENTSINCE "01-Oct-2024"')
# Combine criteria
docs = retriever.invoke('FROM "boss@company.com" SUBJECT "urgent"')
for doc in docs:
print(doc.page_content) # Formatted email content
附件处理
检索器支持三种处理电子邮件附件的模式:
- -
"names_only"(默认):仅列出附件名称 - -
"text_extract":从 PDF 和纯文本附件中提取文本 - -
"full_content":使用 docling 从 office 文档中完整提取(需要[docling]额外)
retriever = ImapRetriever(
config=config,
k=10,
attachment_mode="text_extract"
)
在链中使用
与其他检索器一样, ImapRetriever 可以通过链合并到 LLM 应用程序中。以下是一个完整示例,使用 LLM 生成 IMAP 查询并根据电子邮件内容回答问题:
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_imap import ImapRetriever, ImapConfig
# Setup LLM (example using OpenRouter)
llm = ChatOpenAI(
model="google/gemini-2.5-flash",
temperature=0,
openai_api_key=os.getenv("OPENAI_API_KEY"),
openai_api_base="https://openrouter.ai/api/v1"
)
# IMAP query generation prompt
query_prompt = ChatPromptTemplate.from_template(
"""Convert the following user question into an IMAP search query.
IMAP query syntax examples:
- 'FROM "john@example.com"' - emails from specific sender
- 'SUBJECT "project update"' - emails with specific subject
- 'SENTSINCE "01-Oct-2024"' - emails since specific date
- 'BODY "meeting"' - emails containing specific word in body
- 'FROM "boss@company.com" SUBJECT "urgent"' - combine criteria
IMPORTANT: Include only VALID imap command in output.
IMPORTANT: Do not include any other text in output.
User Question: {question}
IMAP Query:"""
)
# Answer generation prompt
answer_prompt = ChatPromptTemplate.from_template(
"""Answer the question based only on the context provided from emails.
Context:
{context}
Question: {question}
Answer:"""
)
# IMAP retriever configuration
config = ImapConfig(
host="localhost",
port=3993,
user="test",
password="test123",
ssl_mode="ssl",
auth_method="login",
verify_cert=False,
)
retriever = ImapRetriever(
config=config,
k=5,
attachment_mode="names_only"
)
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
# Create the chain
query_chain = query_prompt | llm | StrOutputParser()
def generate_imap_query(question):
return query_chain.invoke({"question": question})
def search_emails(query):
return retriever.invoke(query)
full_chain = (
{
"question": lambda x: x,
"imap_query": lambda x: generate_imap_query(x)
}
| RunnablePassthrough.assign(
context=lambda x: format_docs(search_emails(x["imap_query"]))
)
| answer_prompt
| llm
| StrOutputParser()
)
# Use the chain
TODO = full_chain.invoke("Please make a TODO based on the e-mails having URGENT in subject")
print(TODO)
清理测试环境
如果您使用的是 GreenMail 测试容器,请在测试后进行清理:
cmd = ["podman", "rm", "--force", "langchain-imap-test"]
result = subprocess.run(cmd, capture_output=True, text=True, check=True)