langchain について
langchainは、LLMアプリケーションを開発するためのフレームワークです。組み合わせ可能なチェーンとLangGraphのエージェントグラフを備え、RAG、構造化出力、トレースにも対応します。
主な機能
- 01LCELによるチェーン構成
- 02LangGraphでのエージェントグラフ構築
- 03RAGパイプラインの作成
- 04構造化出力の生成
- 05会話履歴の管理とストリーミング
- 06LangSmithによる実行トレース
活用シーン
- 01プロンプトとLLM呼び出しの組み合わせ
- 02ベクトルストアを使った検索フローの構築
- 03ツールを使うエージェントワークフローの定義
- 04チェーン実行のトレース
インストール
コマンドをコピーしてターミナルで実行します。
$
npx skills add https:/ / github.com/ hoodini/ ai-agents-skills --skill langchainインストール後、Claude Code で/langchainを入力して読み込まれたか確認します。
詳細
LangChain & LangGraph
合成可能なチェーンとエージェントグラフを使って、高度なLLMアプリケーションを構築します。
クイックスタート
pip install langchain langchain-openai langchain-anthropic langgraph
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
# Simple chain
llm = ChatAnthropic(model="claude-3-sonnet-20240229")
prompt = ChatPromptTemplate.from_template("Explain {topic} in simple terms.")
chain = prompt | llm
response = chain.invoke({"topic": "quantum computing"})
LCEL(LangChain Expression Language)
パイプ演算子でチェーンを構成します。
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
# Chain with parsing
chain = (
{"topic": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
result = chain.invoke("machine learning")
RAG パイプライン
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
# Create vector store
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(documents, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
# RAG prompt
prompt = ChatPromptTemplate.from_template("""
Answer based on the following context:
{context}
Question: {question}
""")
# RAG chain
rag_chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
answer = rag_chain.invoke("What is the refund policy?")
LangGraph エージェント
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool
from typing import TypedDict, Annotated
import operator
# Define state
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
# Define tools
@tool
def search(query: str) -> str:
"""Search the web."""
return f"Results for: {query}"
@tool
def calculator(expression: str) -> str:
"""Calculate mathematical expression."""
return str(eval(expression))
tools = [search, calculator]
# Create graph
graph = StateGraph(AgentState)
# Add nodes
graph.add_node("agent", call_model)
graph.add_node("tools", ToolNode(tools))
# Add edges
graph.set_entry_point("agent")
graph.add_conditional_edges(
"agent",
should_continue,
{"continue": "tools", "end": END}
)
graph.add_edge("tools", "agent")
# Compile
app = graph.compile()
# Run
result = app.invoke({"messages": [HumanMessage(content="What is 25 * 4?")]})
構造化出力
from langchain_core.pydantic_v1 import BaseModel, Field
class Person(BaseModel):
name: str = Field(description="Person's name")
age: int = Field(description="Person's age")
occupation: str = Field(description="Person's job")
# Structured LLM
structured_llm = llm.with_structured_output(Person)
result = structured_llm.invoke("John is a 30 year old engineer")
# Person(name='John', age=30, occupation='engineer')
メモリ
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
# Message history
store = {}
def get_session_history(session_id: str):
if session_id not in store:
store[session_id] = ChatMessageHistory()
return store[session_id]
# Chain with memory
with_memory = RunnableWithMessageHistory(
chain,
get_session_history,
input_messages_key="input",
history_messages_key="history"
)
# Use with session
response = with_memory.invoke(
{"input": "My name is Alice"},
config={"configurable": {"session_id": "user123"}}
)
ストリーミング
# Stream tokens
async for chunk in chain.astream({"topic": "AI"}):
print(chunk.content, end="", flush=True)
# Stream events (for debugging)
async for event in chain.astream_events({"topic": "AI"}, version="v1"):
print(event)
LangSmith トレーシング
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-project"
# All chains are now traced automatically
chain.invoke({"topic": "AI"})
リソース
- LangChain ドキュメント: https://python.langchain.com/docs/introduction/
- LangGraph ドキュメント: https://langchain-ai.github.io/langgraph/
- LangSmith: https://smith.langchain.com/
- LangChain Hub: https://smith.langchain.com/hub
- LangChain テンプレート: https://github.com/langchain-ai/langchain/tree/master/templates
コメント
Markdown 対応