# pip install langchain langchain-openai
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
# 1) モデル
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 2) プロンプト
prompt = ChatPromptTemplate.from_messages( and streaming guide.
---
## 80％の時間で使う基本構成要素
### 1) プロンプトと出力解析
- 構造化プロンプトには`ChatPromptTemplate`を使用。
- 型付き応答には`StrOutputParser`やJSONパーサーで出力を解析。
```python
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template(
"""
以下のテキストを3つの箇条書きで要約してください：
---
{text}
"""
)
parser = StrOutputParser
chain = prompt | llm | parser
summary = chain.invoke({"text": "LangChainはRAGやツールを使ったLLMアプリの構築を支援します。"})
print(summary)

# pip install faiss-cpu tiktoken
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
# ドキュメントの準備
texts = .
---
## プロトタイプから本番まで：ステップバイステップの設計図
### ステップ1：ユーザーストーリーを定義
- ユーザーは誰か？どんな仕事を達成しようとしているのか？
- 例：「内部ドキュメントと最近のチケットから製品質問に答えるサポートエージェント」
### ステップ2：ミニマムバイアブルスタックを選択
- モデル：手頃で信頼できるモデルを選ぶ（例：GPT-4o-miniやオープンモデル）。
- データ：今すぐRAGが必要か検討。必要ならFAISSをローカルで開始。
- 入出力：高速イテレーションのためLCELを利用し、カスタム接着コードは避ける。
### ステップ3：クリーンなRAGループを実装
- 文書を適切に分割。
- 埋め込みをインデックス化。
- コンテキストと引用付きでプロンプト作成。
- 関連コンテキストがない場合の幻覚防止策を追加。
```python
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
qa_prompt = ChatPromptTemplate.from_template(
"""
以下のCONTEXT ONLYを使って質問に答えてください。コンテキストに答えがない場合は「わかりません」と答え、引用元のドキュメントIDを含めてください。
CONTEXT:
{context}
QUESTION: {question}
"""
)
parser = StrOutputParser
rag_chain = (RunnableParallel(context=retriever, question=RunnableLambda(lambda x: x.
### ステップ5：型付き出力と検証
- APIレスポンスに構造を与えるために`PydanticOutputParser`やJSONスキーマを使用。
- モデルドリフトを検出するためにフィールドを検証。
### ステップ6：実用的なタスクのためのツールと関数呼び出し
- ツールは必要最低限に導入。
- よく使うツール：電卓、ウェブ検索、SQLクエリ実行、コード実行。
- ツールの機能はドキュメント文字列で明確に記述。
### ステップ7：堅牢化
- レートリミットとリトライ戦略。
- タイムアウトとサーキットブレーカー。
- 安全フィルターとコンテンツチェック。
### ステップ8：評価と継続的改善
- 期待出力とセットされたゴールデンデータセットでテスト。
- 忠実性、回答の完全性、引用精度を評価。
- リトリーバルのヒット率と遅延を測定。
---
## よくあるパターンと注意点
- シンプルに始める：エージェントよりもまずチェーンを。予測可能性と低コストを得やすい。
- チャンク分割が重要：チャンクサイズや重複の調整はモデル差よりもリトリーバル品質に影響大。
- プロンプトリーク：システムプロンプトに過剰な情報を詰め込まず、焦点を絞ること。
- 決定論を：評価や重要な処理では`temperature=0`に設定。
- ストリーミングUX：システムがアセット取得やコンテキストの事前読み込みを行う間にUIへトークンをストリーム配信。
- 構造化出力：パーサーを利用し、下流の統合を容易に。
---
## フルミニプロジェクト：引用つきドキュメントQ&A
この例では、取り込み、RAG、回答生成、ストリーミングをまとめて示します。
```python
# pip install langchain langchain-openai faiss-cpu tiktoken
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnableLambda
# 1) 取り込み
corpus = {
"pricing": "Our Pro plan supports 1M context tokens and includes priority support.",
"limits": "The API rate limit is 60 requests per minute for Pro users.",
"security": "We store logs for 30 days unless logging is disabled by the admin.",
}
splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50)
all_chunks, ids = [], []
for doc_id, text in corpus.items:
for i, chunk in enumerate(splitter.split_text(text)):
all_chunks.append(chunk)
ids.append(f"{doc_id}-{i}")
# 2) インデックス化
db = FAISS.from_texts(all_chunks, OpenAIEmbeddings)
retriever = db.as_retriever(k=4)
# 3) プロンプト
prompt = ChatPromptTemplate.from_template(
"""
あなたはサポートアシスタントです。CONTEXTを使って回答してください。
わからない場合は「わかりません」と答え、ソースIDの引用を含めてください。
CONTEXT:
{context}
QUESTION: {question}
"""
)
# 4) モデルとパーサー
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser
# 5) チェーン合成
rag = (
RunnableParallel(
context=retriever,
question=RunnableLambda(lambda x: x["question"])  # そのまま通す
)
| prompt
| llm
| parser
)
# 6) 質問する
for chunk in rag.stream({"question": "Proのレート制限とログ保持は？"}):
print(chunk, end="", flush=True)