LangChain 個人知識ベースQ&Aシステム
この章では、Markdownファイル、PDFドキュメントを読み込み、これらの内容に基づいてQ&Aを行う個人知識ベースシステムを構築します。
システム設計
- ドキュメント読み込み：Markdown、TXT、PDFの複数形式に対応
- ベクトル検索：Chroma 永続化ストレージ、増分更新に対応
- 引用元：回答に元ドキュメントと断片の位置を添付
- ストリーミング出力：回答をトークンごとに表示
完全なコード
実行前に.envファイルに設定する必要がありますDEEPSEEK_API_KEY（Chatモデル）とDASHSCOPE_API_KEY（Alibaba Cloud百炼、知識ベースのテキストベクトル化に使用）、具体的な申請方法とよくある問題の調査は『LangChain スマートカスタマーサービスボット』の節を参照してください。
サンプル
# ファイルパス：knowledge_qa.py
# pip install langchain langchain-deepseek langchain-openai langchain-community langchain-chroma chromadb pypdf
from dotenv import load_dotenv
load_dotenv()
import os
from pathlib import Path
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader, PyPDFLoader
class KnowledgeBase:
"""個人知識ベースマネージャー"""
def __init__(self, persist_dir: str = "./my_knowledge_db"):
self.persist_dir = persist_dir
# 阿里云百炼のEmbedding APIはOpenAI仕様と互換性があるため、langchain-openaiを使う
# のOpenAIEmbeddingsを呼び出すだけでよい。langchain-community内の
# メンテナンス停止済みのDashScopeEmbeddingsは不要です。
# check_embedding_ctx_length=False：tiktokenの事前トークン化をオフにし、元のテキストを直接送信する
# （百炼APIはtoken idの配列を受け付けない）
# chunk_size=10：百炼のEmbedding APIは1回のリクエストで最大10件のテキストを受け付ける。
self.embeddings = OpenAIEmbeddings(
model="text-embedding-v4",
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
check_embedding_ctx_length=False,
chunk_size=10,
)
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500, chunk_overlap=50,
separators=["\n\n", "\n", "。", "！", "？", ". ", "! ", "? ", " "],
)
self.vector_store = None
self._load_or_create()
def _load_or_create(self):
"""既存のベクトルライブラリを読み込むか新規作成する（Chromaは同じパラメータでよい、
ディレクトリの存在有無はプロンプトメッセージに影響するだけで、作成ロジックには影響しない）"""
is_existing = os.path.exists(self.persist_dir) and os.listdir(self.persist_dir)
self.vector_store = Chroma(
embedding_function=self.embeddings,
persist_directory=self.persist_dir,
)
if is_existing:
print(f"ベクトルライブラリを読み込みました：{self.vector_store._collection.count()} 個のドキュメントチャンク")
else:
print("新しいベクトルライブラリを作成しました")
def add_file(self, file_path: str) -> int:
"""ファイルを知識ベースに追加し、追加されたドキュメントチャンク数を返す
ファイル拡張子に応じて適切なローダーを自動選択する：
- .pdf は PyPDFLoader（pypdf ライブラリでPDFを解析し、ページごとに複数のDocumentを返す）
- その他（.md、.txt などのプレーンテキストファイル）は TextLoader で元のテキストとして読み取る
"""
suffix = Path(file_path).suffix.lower()
if suffix == ".pdf":
loader = PyPDFLoader(file_path)
else:
loader = TextLoader(file_path, encoding="utf-8")
docs = loader.load()
# ファイル名の出所を統一的に補足する；PyPDFLoaderで読み込んだ各Documentには自動的に
# pageフィールド（何ページ目）が付与され、検索時に断片の位置を特定するために一緒に使える
for doc in docs:
doc.metadata["source"] = Path(file_path).name
chunks = self.text_splitter.split_documents(docs)
self.vector_store.add_documents(chunks)
print(f"{Path(file_path).name} を追加しました：{len(chunks)} 個のドキュメントチャンク")
return len(chunks)
def add_text(self, text: str, source: str = "手動追加") -> int:
"""テキストを直接知識ベースに追加する"""
chunks = self.text_splitter.create_documents(
[text], metadatas=[{"source": source}]
)
self.vector_store.add_documents(chunks)
return len(chunks)
def search(self, query: str, k: int = 3) -> list:
"""知識ベースを検索する"""
return self.vector_store.similarity_search(query, k=k)
def get_retriever(self):
"""レトリーバーを取得する"""
return self.vector_store.as_retriever(search_kwargs={"k": 3})
# pip install langchain langchain-deepseek langchain-openai langchain-community langchain-chroma chromadb pypdf
from dotenv import load_dotenv
load_dotenv()
import os
from pathlib import Path
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader, PyPDFLoader
class KnowledgeBase:
"""個人知識ベースマネージャー"""
def __init__(self, persist_dir: str = "./my_knowledge_db"):
self.persist_dir = persist_dir
# 阿里云百炼のEmbedding APIはOpenAI仕様と互換性があるため、langchain-openaiを使う
# のOpenAIEmbeddingsを呼び出すだけでよい。langchain-community内の
# メンテナンス停止済みのDashScopeEmbeddingsは不要です。
# check_embedding_ctx_length=False：tiktokenの事前トークン化をオフにし、元のテキストを直接送信する
# （百炼APIはtoken idの配列を受け付けない）
# chunk_size=10：百炼のEmbedding APIは1回のリクエストで最大10件のテキストを受け付ける。
self.embeddings = OpenAIEmbeddings(
model="text-embedding-v4",
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
check_embedding_ctx_length=False,
chunk_size=10,
)
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500, chunk_overlap=50,
separators=["\n\n", "\n", "。", "！", "？", ". ", "! ", "? ", " "],
)
self.vector_store = None
self._load_or_create()
def _load_or_create(self):
"""既存のベクトルライブラリを読み込むか新規作成する（Chromaは同じパラメータでよい、
ディレクトリの存在有無はプロンプトメッセージに影響するだけで、作成ロジックには影響しない）"""
is_existing = os.path.exists(self.persist_dir) and os.listdir(self.persist_dir)
self.vector_store = Chroma(
embedding_function=self.embeddings,
persist_directory=self.persist_dir,
)
if is_existing:
print(f"ベクトルライブラリを読み込みました：{self.vector_store._collection.count()} 個のドキュメントチャンク")
else:
print("新しいベクトルライブラリを作成しました")
def add_file(self, file_path: str) -> int:
"""ファイルを知識ベースに追加し、追加されたドキュメントチャンク数を返す
ファイル拡張子に応じて適切なローダーを自動選択する：
- .pdf は PyPDFLoader（pypdf ライブラリでPDFを解析し、ページごとに複数のDocumentを返す）
- その他（.md、.txt などのプレーンテキストファイル）は TextLoader で元のテキストとして読み取る
"""
suffix = Path(file_path).suffix.lower()
if suffix == ".pdf":
loader = PyPDFLoader(file_path)
else:
loader = TextLoader(file_path, encoding="utf-8")
docs = loader.load()
# ファイル名の出所を統一的に補足する；PyPDFLoaderで読み込んだ各Documentには自動的に
# pageフィールド（何ページ目）が付与され、検索時に断片の位置を特定するために一緒に使える
for doc in docs:
doc.metadata["source"] = Path(file_path).name
chunks = self.text_splitter.split_documents(docs)
self.vector_store.add_documents(chunks)
print(f"{Path(file_path).name} を追加しました：{len(chunks)} 個のドキュメントチャンク")
return len(chunks)
def add_text(self, text: str, source: str = "手動追加") -> int:
"""テキストを直接知識ベースに追加する"""
chunks = self.text_splitter.create_documents(
[text], metadatas=[{"source": source}]
)
self.vector_store.add_documents(chunks)
return len(chunks)
def search(self, query: str, k: int = 3) -> list:
"""知識ベースを検索する"""
return self.vector_store.similarity_search(query, k=k)
def get_retriever(self):
"""レトリーバーを取得する"""
return self.vector_store.as_retriever(search_kwargs={"k": 3})
現在
TextLoader、PyPDFLoaderこの種のローカルファイルローダーには独立したメンテナンスパッケージがまだなく、現時点ではlangchain_community.document_loadersからインポートするしかなく、実行時にlangchain-community全体のDeprecationWarningが表示される場合があります。この警告は現時点では無視して問題ありません。公式が独立したドキュメントローダーパッケージをリリースしたら移行すればよいです。前述のembeddingの「既に代替手段があるのに切り替えていない」ケースとは異なります。
Agentパートを続ける：
サンプル
# ========== 知識ベースを作成し、サンプルデータを追加 ==========
kb = KnowledgeBase("./my_knowledge_db")
# サンプル知識を追加
kb.add_text(
"サンプル の Python3 基礎チュートリアルには以下の章が含まれます："
"1. Python入門と環境構築 2. 基本データ型 3. 演算子と式 "
"4. 条件分岐 if-else 5. ループ for/while 6. 関数定義と呼び出し "
"7. モジュールとパッケージ 8. ファイル操作 9. 例外処理 10. オブジェクト指向プログラミング",
source="Python3 チュートリアルの概要"
)
kb.add_text(
"優れたPython開発者になるには、以下のルートで学習することをお勧めします："
"ステップ1：Pythonの基本構文を習得する（1〜2週間）；"
"ステップ2：データ構造とアルゴリズムの基礎を学ぶ（2〜3週間）；"
"ステップ3：一つの方向を選んで深く学ぶ（Web開発/データ分析/AI）；"
"ステップ4：実践プロジェクトを2〜3個行い、知識を定着させます。",
source="Python 学習ロードマップ"
)
kb.add_text(
"サンプルのオンラインプログラミング環境は、Python、JavaScript、Java、C++など複数の言語に対応しています。"
"ユーザーはソフトウェアのインストール不要で、ブラウザを開くだけでコードの作成と実行ができます。"
"オンライン環境はコードハイライト、自動補完、エラー表示機能もサポートしています。",
source="オンラインプログラミング環境の説明"
)
# ローカルファイルも読み込めます。PDFとMarkdown/TXTは自動的に認識されます：
# kb.add_file("./docs/产品手册.pdf")
# kb.add_file("./docs/常见问题.md")
# ========== RAG Agent を作成 ==========
@tool
def search_knowledge(query: str) -> str:
"""個人知識ベースで関連情報を検索します。検索する際は完全な質問またはキーフレーズを使用します。
Args:
query: 検索する質問またはキーフレーズ
"""
docs = kb.search(query, k=3)
if not docs:
return "知識ベースに関連情報が見つかりませんでした。"
results = []
for i, doc in enumerate(docs, 1):
source = doc.metadata.get("source", "不明な出所")
page = doc.metadata.get("page")
location = f"{source}" + (f" {page + 1} ページ目" if page is not None else "")
content = doc.page_content[:200]
results.append(f"[{i}] 出所：{location}\n{content}")
return "\n\n---\n\n".join(results)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_knowledge],
system_prompt="""あなたは個人知識ベースアシスタントです。
## ルール
1. すべての質問は最初にsearch_knowledgeツールで知識ベースを検索すること
2. 回答時には情報の出所（ドキュメント名、PDFの場合はページ番号も）を明記すること
3. 知識ベースに関連内容がない場合は、正直に伝えること
4. 回答は構造化し、番号付きリストまたは段落を使用すること""",
)
# ========== テスト：非ストリーミング、検索された内容も同時に確認 ==========
def ask(question: str):
"""質問し、検索された元の断片と最終回答を同時に表示して、デバッグを容易にする"""
print(f"\n{'='*60}")
print(f"Q: {question}")
print(f"{'='*60}")
result = agent.invoke({
"messages": [HumanMessage(content=question)]
})
# 検索された内容を表示
for msg in result["messages"]:
if msg.type == "tool":
print(f"\n[検索された内容]")
print(msg.content[:300])
print(f"\n[回答]")
print(result["messages"][-1].content)
# ========== テスト：ストリーミング、システム設計の「ストリーミング出力」に対応 ==========
def ask_stream(question: str):
"""質問し、回答をトークンごとにストリーミング表示して、タイプライター効果を実現する"""
print(f"\n{'='*60}")
print(f"Q: {question}")
print(f"{'='*60}")
print("\n[回答] ", end="", flush=True)
for chunk, metadata in agent.stream(
{"messages": [HumanMessage(content=question)]},
stream_mode="messages",
):
# metadata["langgraph_node"] == "model" はこのチャンクがモデル生成によるものであることを示す
# 最終回答のノード。ツール呼び出しなど他のタイプのチャンクを除外する
if metadata.get("langgraph_node") == "model" and chunk.content:
print(chunk.content, end="", flush=True)
print()
ask("Python3 基礎チュートリアルにはどのような章が含まれますか？")
ask("Pythonの学習ロードマップをどのように計画しますか？")
ask_stream("サンプルのオンラインプログラミング環境はどのような機能をサポートしていますか？")
kb = KnowledgeBase("./my_knowledge_db")
# サンプル知識を追加
kb.add_text(
"サンプル の Python3 基礎チュートリアルには以下の章が含まれます："
"1. Python入門と環境構築 2. 基本データ型 3. 演算子と式 "
"4. 条件分岐 if-else 5. ループ for/while 6. 関数定義と呼び出し "
"7. モジュールとパッケージ 8. ファイル操作 9. 例外処理 10. オブジェクト指向プログラミング",
source="Python3 チュートリアルの概要"
)
kb.add_text(
"優れたPython開発者になるには、以下のルートで学習することをお勧めします："
"ステップ1：Pythonの基本構文を習得する（1〜2週間）；"
"ステップ2：データ構造とアルゴリズムの基礎を学ぶ（2〜3週間）；"
"ステップ3：一つの方向を選んで深く学ぶ（Web開発/データ分析/AI）；"
"ステップ4：実践プロジェクトを2〜3個行い、知識を定着させます。",
source="Python 学習ロードマップ"
)
kb.add_text(
"サンプルのオンラインプログラミング環境は、Python、JavaScript、Java、C++など複数の言語に対応しています。"
"ユーザーはソフトウェアのインストール不要で、ブラウザを開くだけでコードの作成と実行ができます。"
"オンライン環境はコードハイライト、自動補完、エラー表示機能もサポートしています。",
source="オンラインプログラミング環境の説明"
)
# ローカルファイルも読み込めます。PDFとMarkdown/TXTは自動的に認識されます：
# kb.add_file("./docs/产品手册.pdf")
# kb.add_file("./docs/常见问题.md")
# ========== RAG Agent を作成 ==========
@tool
def search_knowledge(query: str) -> str:
"""個人知識ベースで関連情報を検索します。検索する際は完全な質問またはキーフレーズを使用します。
Args:
query: 検索する質問またはキーフレーズ
"""
docs = kb.search(query, k=3)
if not docs:
return "知識ベースに関連情報が見つかりませんでした。"
results = []
for i, doc in enumerate(docs, 1):
source = doc.metadata.get("source", "不明な出所")
page = doc.metadata.get("page")
location = f"{source}" + (f" {page + 1} ページ目" if page is not None else "")
content = doc.page_content[:200]
results.append(f"[{i}] 出所：{location}\n{content}")
return "\n\n---\n\n".join(results)
model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
model=model,
tools=[search_knowledge],
system_prompt="""あなたは個人知識ベースアシスタントです。
## ルール
1. すべての質問は最初にsearch_knowledgeツールで知識ベースを検索すること
2. 回答時には情報の出所（ドキュメント名、PDFの場合はページ番号も）を明記すること
3. 知識ベースに関連内容がない場合は、正直に伝えること
4. 回答は構造化し、番号付きリストまたは段落を使用すること""",
)
# ========== テスト：非ストリーミング、検索された内容も同時に確認 ==========
def ask(question: str):
"""質問し、検索された元の断片と最終回答を同時に表示して、デバッグを容易にする"""
print(f"\n{'='*60}")
print(f"Q: {question}")
print(f"{'='*60}")
result = agent.invoke({
"messages": [HumanMessage(content=question)]
})
# 検索された内容を表示
for msg in result["messages"]:
if msg.type == "tool":
print(f"\n[検索された内容]")
print(msg.content[:300])
print(f"\n[回答]")
print(result["messages"][-1].content)
# ========== テスト：ストリーミング、システム設計の「ストリーミング出力」に対応 ==========
def ask_stream(question: str):
"""質問し、回答をトークンごとにストリーミング表示して、タイプライター効果を実現する"""
print(f"\n{'='*60}")
print(f"Q: {question}")
print(f"{'='*60}")
print("\n[回答] ", end="", flush=True)
for chunk, metadata in agent.stream(
{"messages": [HumanMessage(content=question)]},
stream_mode="messages",
):
# metadata["langgraph_node"] == "model" はこのチャンクがモデル生成によるものであることを示す
# 最終回答のノード。ツール呼び出しなど他のタイプのチャンクを除外する
if metadata.get("langgraph_node") == "model" and chunk.content:
print(chunk.content, end="", flush=True)
print()
ask("Python3 基礎チュートリアルにはどのような章が含まれますか？")
ask("Pythonの学習ロードマップをどのように計画しますか？")
ask_stream("サンプルのオンラインプログラミング環境はどのような機能をサポートしていますか？")
実行結果：
============================================================ Q: Python3 基础教程包含哪些章节？ ============================================================ [检索到的内容] [1] 来源：Python3 教程大纲 サンプル 的 Python3 基础教程包含以下章节：... [回答] Python3 基础教程包含以下章节（来源：Python3 教程大纲）： 1. Python 简介与环境搭建 2. 基本数据类型 3. 运算符与表达式 ... ============================================================ Q: 如何规划 Python 学习路线？ ============================================================ [回答] 根据知识库中的 Python 学习路线建议（来源：Python 学习路线）： 第一步：掌握基础语法（1-2 周） 第二步：学习数据结构和算法（2-3 周） 第三步：选择方向深入学习（Web/数据分析/AI） 第四步：做 2-3 个实战项目巩固 ============================================================ Q: サンプル的在线编程环境支持哪些功能？ ============================================================ [回答] 根据知识库中的说明（来源：在线编程环境说明），サンプル的在线编程环境： 1. 支持 Python、JavaScript、Java、C++ 等多种语言 2. 无需安装任何软件，打开浏览器即可编写和运行代码 3. 支持代码高亮、自动补全和错误提示功能
その他の拡張最後の質問は
ask_stream()、実際に実行すると"[回答]"の後の文字が1文字ずつ（または数文字ずつ）次々に出力され、タイプライター効果のようになります。上記は表示の便宜上、出力完了後の最終テキストを直接貼り付けています。