AIã®æªæ¥ã¯åŒ·åãªã¢ãã«ãæã€ããšã ãã§ã¯ãªãã ããããã€ã³ããªãžã§ã³ãã«ãªãŒã±ã¹ãã¬ãŒã·ã§ã³ããããš ã§ããOpenAIãClaudeãGoogle Geminiã«ãŸãããæ°çŸã®ãšãŒãžã§ã³ãå®è£ ãæãããçµéšãããç§ã¯1ã€ã®éèŠãªçå®ãåŠã³ãŸããããããã¿ã€ããšãŒãžã§ã³ããšæ¬çªç°å¢å¯Ÿå¿ã·ã¹ãã ã®éã®ã®ã£ããã¯ãã³ãŒãå質ã§ã¯ãªã ä¿¡é Œæ§ã¢ãŒããã¯ãã£ ã§æž¬ããããšããããšã§ãã
仿¥ã¯ãæ¬çªç°å¢ã®AIãšãŒãžã§ã³ãéçºã®èå°è£ãã玹ä»ããŸãããšãŒãžã§ã³ãã1æéã«æ°åã®ãªã¯ãšã¹ããåŠçãããŠãŒã¶ãŒã5ç§æªæºã®ã¬ã¹ãã³ã¹ãæåŸ ãã1ã€ã®ããŒã«åŒã³åºãã®å€±æãã·ã¹ãã å šäœã®æ··ä¹±ã«ã€ãªããå¯èœæ§ãããå Žåã«ãå®éã«æ©èœããLangChainãªãŒã±ã¹ãã¬ãŒã·ã§ã³ãã¿ãŒã³ãæ·±ãæãäžããŸãã
ããã¯çè«ã§ã¯ãããŸãããAIãšã³ãžãã¢ãªã³ã°ã®æåç·ããã®å®èšŒæžã¿ã®ç¥èã§ãã
æ¬çªç°å¢ã®çŸå®: ãªãã»ãšãã©ã®AIãšãŒãžã§ã³ãã¯å€±æããã®ã
è¡æçãªçµ±èšããå§ããŸãããã ã¯ãŒã¯ãããŒå ã®åAIãšãŒãžã§ã³ãã®ä¿¡é Œæ§ã95%ã®å Žåãããã3ã€ã®ãšãŒãžã§ã³ããé£éãããã ãã§å šäœã®æåçã¯çŽ86%ã«äœäžããŸã ãããã«ã¹ãããã远å ãããšãä¿¡é Œæ§ã¯ææ°é¢æ°çã«æ¥èœããŸãã
ç§ã¯ãåªç§ãªãšã³ãžãã¢ãéçºç°å¢ã§ã¯å®ç§ã«åäœããæŽç·Žããããã«ããšãŒãžã§ã³ãã·ã¹ãã ãæ§ç¯ãããã®ã®ãæ¬çªç°å¢ã®è² è·ã§åŽ©å£ããã®ãèŠãŠããŸãããåé¡ã¯äœã§ãããã? 圌ãã¯ä¿¡é Œæ§ã®ä»£ããã«èœåãæé©åããŠããŸãã圌ãã¯ãç¹å®ã®å¶åŸ¡ããã倿ã«LLMãæŽ»çšãã åªãããšã³ãžãã¢ãªã³ã°ã®ãœãããŠã§ã¢ã·ã¹ãã ãæ§ç¯ãã¹ããšãã«ãããšãŒãžã§ã³ãçãªãã·ã¹ãã ãæ§ç¯ããŠããã®ã§ãã
2025幎ã®çŸåšèµ·ãã£ãŠãããã©ãã€ã ã·ããã¯æ¬¡ã®ãšããã§ãã èªåŸåãšãŒãžã§ã³ãã«åãçµãAIéçºè ã®60%ãLangChainãäž»èŠãªãªãŒã±ã¹ãã¬ãŒã·ã§ã³ã¬ã€ã€ãŒãšããŠäœ¿çšããŠããŸã ããããŠãLinkedInãUberãKlarnaãªã©ã®äŒæ¥ã¯æ¬çªç°å¢ã®ãããã€ã«LangGraphã«è³ããŠããŸãããªãã§ãããã? LangChainããããã¿ã€ãã³ã°ãã¬ãŒã ã¯ãŒã¯ããæ¬çªç°å¢å¯Ÿå¿ã®ãªãŒã±ã¹ãã¬ãŒã·ã§ã³ãã©ãããã©ãŒã ã«é²åããããã§ãã
åã«åäœããã ãã§ãªãã ã¹ã±ãŒã«ãã ãšãŒãžã§ã³ãã®æ§ç¯æ¹æ³ãæ¢ããŸãããã
ã¢ãŒããã¯ãã£ç¬¬äž: LangGraphã®åºç€
2025å¹Žã«æ¬çªç°å¢ã®AIãšãŒãžã§ã³ããæ§ç¯ããŠããŠãLangGraphã䜿çšããŠããªãå ŽåãçæãåŸãã«çžãããŠæŠã£ãŠãããããªãã®ã§ããLangGraphã¯ãé·å¹Žã®LangChainãã£ãŒãããã¯ããçãŸããæ¬çªç°å¢ã«ãããŠãšãŒãžã§ã³ããã¬ãŒã ã¯ãŒã¯ãã©ã®ããã«æ©èœãã¹ãããæ ¹æ¬çã«åèããŸããã
ãªãçã®LangChainã§ã¯ãªãLangGraphãªã®ã?
LangGraphã¯ã次ã®ãããªæ©èœãæäŸãã äœã¬ãã«ã®ãšãŒãžã§ã³ããªãŒã±ã¹ãã¬ãŒã·ã§ã³ãã¬ãŒã ã¯ãŒã¯ ã§ãã
- æ°žç¶çãªå®è¡ - ãšãŒãžã§ã³ãã®ç¶æ ã¯ã¯ã©ãã·ã¥ãåèµ·åãè¶ããŠæç¶ããŸã
- ãã现ããå¶åŸ¡ - åžæãšç¥ãã®ã«ãŒãã§ã¯ãªããããŒããšãšããžãšããŠã¢ããªã±ãŒã·ã§ã³ãããŒã衚çŸããŸã
- èªåã§ç°¡åã«æ§ç¯ã§ããªã æ¬çªç°å¢ã«äžå¯æ¬ ãªæ©èœ :
- äœæ¥ã倱ããã«äººéã«ããã«ãŒãå²ã蟌ã¿
- ãšãŒãžã§ã³ãã«ãŒããšè»è·¡ãžã®å®å šãªãã¬ãŒã·ã³ã°å¯èŠæ§
- ããŒã¿ç«¶åãåé¿ããçã®äžŠåå
- èªèã¬ã€ãã³ã·ãŒãåæžããã¹ããªãŒãã³ã°
ãããç§ã«ãšã£ãŠãã¹ãŠãå€ããã¢ãŒããã¯ãã£ã§ã:
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage
# ãªãã¥ãŒãµãŒé¢æ°ã䜿çšããç¶æ
管ç - ä¿¡é Œæ§ã®ããã¯ããŒã³
class AgentState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
current_intent: str | None
tool_results: dict
error_count: int
resolved: bool
# æ¬çªç°å¢å¯Ÿå¿ã®ã«ã¹ã¿ããŒãµãŒãã¹ã°ã©ã
class ProductionAgentGraph:
def __init__ (self):
self.graph = StateGraph(AgentState)
# ããŒããå®çŸ© - ãããããç¹åãã颿°
self.graph.add_node("classify_intent", self.classify_intent)
self.graph.add_node("execute_tools", self.execute_tools)
self.graph.add_node("validate_response", self.validate_response)
self.graph.add_node("error_handler", self.error_handler)
# ãšããžãå®çŸ© - ä¿¡é Œæ§ãå·Šå³ããå¶åŸ¡ãããŒ
self.graph.add_edge("classify_intent", "execute_tools")
self.graph.add_conditional_edges(
"execute_tools",
self.should_validate_or_retry,
{
"validate": "validate_response",
"retry": "execute_tools",
"error": "error_handler"
}
)
self.graph.add_edge("validate_response", END)
# ãšã³ããªãŒãã€ã³ããèšå®
self.graph.set_entry_point("classify_intent")
self.compiled_graph = self.graph.compile()
async def classify_intent(self, state: AgentState) -> AgentState:
"""ãã©ã³ããŒãšãŒãžã§ã³ã - ã·ã¹ãã ã®æŠç¥çé è³"""
# ãšã©ãŒå¢çãæã€å®è£
pass
def should_validate_or_retry(self, state: AgentState) -> str:
"""ã«ãŒãã£ã³ã°ããžã㯠- ãªãŒã±ã¹ãã¬ãŒã·ã§ã³ã®ã€ã³ããªãžã§ã³ã¹"""
if state["error_count"] > 3:
return "error"
if state["tool_results"].get("status") == "success":
return "validate"
return "retry"
ããã§äœãèµ·ãã£ãŠãããæ³šæããŠãã ãã : LLMã«ãããŒå¶åŸ¡ã決å®ãããŠããŸãããæ¡ä»¶ä»ããšããžãšæç€ºçãªã«ãŒãã£ã³ã°ããžãã¯ã䜿çšããŠããŸãããããããã¢ã§ãéæ³ã®ããã«æããããšãŒãžã§ã³ããš æ¬çªç°å¢ã§ç¢ºå®ã«å®è¡ããã ãšãŒãžã§ã³ãã®éãã§ãã
ãã«ããšãŒãžã§ã³ãã¢ãŒããã¯ãã£ãã¿ãŒã³
LangChainã®2025幎ã¢ãŒããã¯ãã£ã¯ããšãŒãžã§ã³ããå°éåããã¢ãžã¥ã©ãŒãªéå±€åãããã·ã¹ãã ã«é²åããŸããããããè€éãªã¯ãŒã¯ãããŒã«äœ¿çšãããã¿ãŒã³ã§ã:
- ãã©ã³ããŒãšãŒãžã§ã³ã - ãŠãŒã¶ãŒã®æå³ããµãã¿ã¹ã¯ã«åè§£ããæŠç¥çé è³
- ãšã°ãŒãã¥ãŒã¿ãŒãšãŒãžã§ã³ã - ç¹å®ã®ãµãã¿ã¹ã¯(ããŒã¿ããŒã¹ã¯ãšãªãAPIåŒã³åºããããŒã¿å€æ)ãåŠçããå°éã®ã¯ãŒã«ãŒ
- ã³ãã¥ãã±ãŒã¿ãŒãšãŒãžã§ã³ã - ãšãŒãžã§ã³ãéã®ã¹ã ãŒãºãªåãæž¡ããä¿èšŒããäžæµã®æ¶è²»ã®ããã«åºåãåãã©ãŒããã
- ããªããŒã¿ãŒãšãŒãžã§ã³ã - ãŠãŒã¶ãŒã«å°éããåã«å¹»èŠããšã©ãŒããã£ããããå質ã²ãŒã
ããã¯ææå°æ©ãªæœè±¡åã§ã¯ãããŸãããã·ã¹ãã ãæ°åã®å€æ§ãªãªã¯ãšã¹ããåŠçããå¿ èŠããããšãã® æ¬è³ªçãªè€éæ§ç®¡ç ã§ãã
ãã«ãã¢ãã«ãªãŒã±ã¹ãã¬ãŒã·ã§ã³: æŠç¥çåªäœæ§
ãããããé¢çœããªããŸãã2025å¹Žã®æã匷åãªAIã·ã¹ãã ã¯ãåäžã®ã¢ãã«ã«äŸåããŠããŸããã ãããããæãåŸæãªããšãåŠçããè€æ°ã®ã¢ãã«ãçµã¿åãããŠããŸã ã
ã¢ãã«éžææŠç¥
åºç¯ãªæ¬çªç°å¢ãã¹ãã«åºã¥ããŠãç§ã®ã¢ãã«ã«ãŒãã£ã³ã°å²åŠã¯æ¬¡ã®ãšããã§ã:
ãªãŒã±ã¹ãã¬ãŒã·ã§ã³ã¬ã€ã€ãŒ:
- GPT-4o - æåªå ãåªããããã©ãŒãã³ã¹ãè²»çšå¯Ÿå¹æãå®å®æ§ãæç€ºã«æ£ç¢ºã«åŸããŸãã
- ãªãClaudeã§ã¯ãªãã®ã? Claudeã¯å€§å±çãªæšè«ã«åªããŠããŸãããè¶ ç²Ÿå¯ãªãªãŒã±ã¹ãã¬ãŒã·ã§ã³äœæ¥ã«ã¯èŠåŽããŸãã
å°éã¿ã¹ã¯:
- Claude 4 (Anthropic APIçµç±) - è€éãªæšè«ãå®å šæ§ãéèŠãªæ±ºå®ããã¥ã¢ã³ã¹ã®ããã³ã³ãã³ãçæ
- GPT-5 - ã¿ã¹ã¯ã®è€éãã«åºã¥ããŠé«é/æèã¢ãŒãéã®ã€ã³ããªãžã§ã³ããªã«ãŒãã£ã³ã°ãå èµ
- Haikuã¢ãã« - åé¡ãšåçŽãªå€æã®ããã®è¶ é«é
ããŒã«åŒã³åºã:
- GPT-4.1 - ããŒã«å©çšã®åºç¯ãªãã¬ãŒãã³ã°ãåããŸãããAPIããŒã¹ãããããŒã«èšè¿°ã¯ãæåã¹ããŒãæ¿å ¥ãããSWE-bench Verifiedã§2%äžåããŸãã
åçã¢ãã«ã«ãŒãã£ã³ã°ãã¿ãŒã³
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from typing import Literal
class MultiModelOrchestrator:
def __init__ (self):
# æé©ãªèšå®ã§ã¢ãã«ãåæå
self.orchestrator = ChatOpenAI(
model="gpt-4o",
temperature=0 # ã«ãŒãã£ã³ã°æ±ºå®ã®ããã®æ±ºå®è«ç
)
self.reasoning_engine = ChatAnthropic(
model="claude-4-opus-20250514",
temperature=0.3
)
self.fast_classifier = ChatOpenAI(
model="gpt-4o-mini",
temperature=0
)
async def route_request(
self,
task: str,
complexity_score: float
) -> Literal["fast", "reasoning", "orchestrator"]:
"""
ã€ã³ããªãžã§ã³ãã«ãŒãã£ã³ã° - ã€ã³ããªãžã§ã³ã¹ã®ããŒããã©ã³ãµãŒ
åçŽãªã¯ãšãª â é«éã§å®äŸ¡ãªã¢ãã«
è€éãªæšè« â 匷åãªã¢ãã«
"""
if complexity_score < 0.3:
return "fast"
elif complexity_score < 0.7:
return "orchestrator"
else:
return "reasoning"
async def execute_with_routing(self, user_query: str):
# ãžã£ããžãšãŒãžã§ã³ããã¿ã¹ã¯ã®è€éããåé¡
classification = await self.fast_classifier.ainvoke([
{"role": "system", "content": "Classify task complexity (0-1)"},
{"role": "user", "content": user_query}
])
complexity = float(classification.content)
route = await self.route_request(user_query, complexity)
# é©åãªã¢ãã«ã«ã«ãŒãã£ã³ã°
model_map = {
"fast": self.fast_classifier,
"reasoning": self.reasoning_engine,
"orchestrator": self.orchestrator
}
selected_model = model_map[route]
return await selected_model.ainvoke([
{"role": "user", "content": user_query}
])
ãã®ãã¿ãŒã³ã¯ãOpenAIã®GPT-5ãå éšã§è¡ã£ãŠããããšãåæ ããŠããŸãã ã€ã³ããªãžã§ã³ã¹ã®ããŒããã©ã³ãµãŒã®ããã«æ¯ãèããŸã ã ããããèªåã§å®è£ ããããšã§ãã³ã¹ããã¬ã€ãã³ã·ãŒãã¢ãã«åºæã®åŒ·ã¿ãå¶åŸ¡ã§ããŸãã
ããã³ãããšã³ãžãã¢ãªã³ã°: æ¬çªç°å¢å¯Ÿå¿ãã¿ãŒã³
ã¢ããã¥ã¢ãšãšãã¹ããŒãã®ããã³ãããšã³ãžãã¢ãªã³ã°ã®éãã¯æž¬å®ã§ããæ¬çªç°å¢ã§ã¯ããã¹ãŠã®ããã³ããã¯ãã¹ããããŒãžã§ã³ç®¡çãç£èŠãå¿ èŠãª APIã³ã³ãã©ã¯ã ã§ãã
3å±€ããã³ããæŠç¥
å±€1: ã·ã¹ãã ããã³ãã(åºç€)
ORCHESTRATOR_SYSTEM_PROMPT = """ããªãã¯ãŠãŒã¶ãŒãªã¯ãšã¹ããå®è¡å¯èœãªãµãã¿ã¹ã¯ã«åè§£ãã責任ãæã€AIãªãŒã±ã¹ãã¬ãŒã·ã§ã³ãšãŒãžã§ã³ãã§ãã
éèŠãªã«ãŒã«:
1. åžžã«TaskPlanã¹ããŒãã«äžèŽããæå¹ãªJSONãåºåããŠãã ãã
2. ããŒã«åãå¹»èŠããªãã§ãã ãã - æäŸããããªã¹ãã®ããŒã«ã®ã¿ã䜿çšããŠãã ãã
3. äžç¢ºããªå Žåã¯ããneeds_clarificationããšããŠåé¡ããå
·äœçãªè³ªåãããŠãã ãã
å©çšå¯èœãªããŒã«:
{tool_descriptions}
åºåãã©ãŒããã:
{
"tasks": [{"tool": "tool_name", "params": {...}, "depends_on": []}],
"reasoning": "ç°¡åãªèª¬æ",
"estimated_complexity": 0.0-1.0
}
枩床ã¬ã€ãã³ã¹: 決å®è«çåäœã®ããã«temperature=0ã§å®è¡ããŠããŸãã"""
ãªããããæ©èœããã®ã: æç¢ºãªå¶çŽãæç€ºçãªåºåãã©ãŒããããããŒã«ã®å¯èŠæ§ã枩床ã®èªèã
å±€2: Few-Shotã®äŸ(æåž«)
æ¬çªç°å¢ã®AIã§æãæŽ»çšãããŠããªãæè¡ãOpenAIã®ç ç©¶ã¯ãFew-ShotåŠç¿ãããŒã«åŒã³åºãã®ç²ŸåºŠãåçã«åäžãããããšã瀺ããŠããŸã:
FEW_SHOT_EXAMPLES = [
{
"user": "What's the weather in Tokyo and what's 15% of 2847?",
"assistant": {
"tasks": [
{"tool": "weather_api", "params": {"location": "Tokyo"}, "depends_on": []},
{"tool": "calculator", "params": {"expression": "2847 * 0.15"}, "depends_on": []}
],
"reasoning": "Two independent tasks - can parallelize",
"estimated_complexity": 0.2
}
}
]
å±€3: åçã³ã³ããã¹ãæ³šå ¥(ãªããã£ãã€ã¶ãŒ)
Anthropicã®ããã³ãããã£ãã·ã³ã°ã䜿çšããŠãã¬ã€ãã³ã·ãŒãšã³ã¹ããåçã«åæžããŸã:
from anthropic import Anthropic
client = Anthropic()
# 倧ããªéçã³ã³ããã¹ãããã£ãã·ã¥
cached_context = """
[å€§èŠæš¡ãªããŒã«ããã¥ã¡ã³ããAPIã¹ããŒããäŸ - 50,000ããŒã¯ã³]
"""
response = client.messages.create(
model="claude-4-opus-20250514",
max_tokens=1024,
system=[
{
"type": "text",
"text": "You are a helpful assistant.",
},
{
"type": "text",
"text": cached_context,
"cache_control": {"type": "ephemeral"} # ããããã£ãã·ã¥!
}
],
messages=[{"role": "user", "content": user_query}]
)
å®äžçãžã®åœ±é¿: Nationwide Building Societyã¯ãã€ã³ã¡ã¢ãªãã£ãã·ã³ã°ã䜿çšããŠAIå¿çæéã10ç§ãã1ç§æªæºã«ççž®ããŸããã ããã¯æ®µéçãªæ¹åã§ã¯ãããŸãããå€é©ã§ãã
ããã³ãããšã³ãžãã¢ãªã³ã°ã®ãã¹ããã©ã¯ãã£ã¹(2025幎ç)
OpenAIãšAnthropicã®å ¬åŒã¬ã€ãã³ã¹ã«åºã¥ããŠããŸã:
- 決å®è«çã¿ã¹ã¯ã«ã¯temperature=0ãäœ¿çš (ããŒã¿æœåºãåé¡ãããŒã«åŒã³åºã)
- ããŒã«ã«æç¢ºãªååãä»ãã - GPT-4.1ã¯ãæåæ¿å ¥ãšæ¯èŒããŠAPIããŒã¹ãããããŒã«èšè¿°ã§2%åªããããã©ãŒãã³ã¹ãçºæ®
- äœç³»çã«å埩 - ã·ã³ãã«ã«å§ããããã©ãŒãã³ã¹ã枬å®ããå¿ èŠãªå Žåã«ã®ã¿è€éãã远å
- æ§é åãããåºåãæŽ»çš - JSONã¹ããŒãæ€èšŒã䜿çšããŠäžæ£ãªåœ¢åŒã®å¿çãé²ã
- ãšãŒãžã§ã³ãçãªãã€ã³ããŒãå«ãã - GPT-4.1ã®å Žåããã¹ãŠã®ãšãŒãžã§ã³ãããã³ããã«3ã€ã®äž»èŠãªã¿ã€ãã®ãªãã€ã³ããŒãå«ããŠæå 端ã®ããã©ãŒãã³ã¹ãå®çŸ
ããŒã«äœ¿çš: ãªãŒã±ã¹ãã¬ãŒã·ã§ã³ã®ããã¯ããŒã³
ããŒã«ã¯ããšãŒãžã§ã³ããæçšã«ãªãå Žæã§ããããããããŒã«åŒã³åºãã¯ãã»ãšãã©ã®æ¬çªç°å¢ã·ã¹ãã ã倱æããå Žæã§ããããŸãã
æ¬çªç°å¢ããŒã«ãã¿ãŒã³
from langchain_core.tools import tool
from typing import Optional
from pydantic import BaseModel, Field
class DatabaseQueryInput(BaseModel):
"""ããŒã¿ããŒã¹ã¯ãšãªã®å
¥åã¹ããŒã - æç€ºçã«!"""
query: str = Field(description="SQL query to execute")
timeout_seconds: int = Field(
default=30,
description="Query timeout in seconds"
)
dry_run: bool = Field(
default=True,
description="If true, validate but don't execute"
)
@tool(args_schema=DatabaseQueryInput)
async def query_database(
query: str,
timeout_seconds: int = 30,
dry_run: bool = True
) -> dict:
"""
æ¬çªç°å¢ã®ã»ãŒãã¬ãŒããåããããŒã¿ããŒã¹ã¯ãšãªãå®è¡ããŸãã
å®å
šæ©èœ:
- å®è¡åã«SQLæ§æãæ€èšŒ
- ã¿ã€ã ã¢ãŠãå¶éã匷å¶
- å®å
šæ§ãã¹ãã®ããã®ãã©ã€ã©ã³ã¢ãŒã
- æ§é åããããšã©ãŒæ
å ±ãè¿ã
æ»ãå€:
{
"status": "success" | "error",
"data": [...] | null,
"error": null | {"type": str, "message": str},
"execution_time_ms": float
}
"""
import asyncio
import time
start_time = time.time()
try:
# æ€èšŒã¬ã€ã€ãŒ
if not is_valid_sql(query):
return {
"status": "error",
"data": None,
"error": {
"type": "ValidationError",
"message": "Invalid SQL syntax"
},
"execution_time_ms": (time.time() - start_time) * 1000
}
# ãã©ã€ã©ã³ã¢ãŒã - å®è¡ããã«æ€èšŒ
if dry_run:
return {
"status": "success",
"data": None,
"error": None,
"execution_time_ms": (time.time() - start_time) * 1000,
"dry_run": True
}
# ã¿ã€ã ã¢ãŠãä»ãã§å®è¡
result = await asyncio.wait_for(
execute_query(query),
timeout=timeout_seconds
)
return {
"status": "success",
"data": result,
"error": None,
"execution_time_ms": (time.time() - start_time) * 1000
}
except asyncio.TimeoutError:
return {
"status": "error",
"data": None,
"error": {
"type": "TimeoutError",
"message": f"Query exceeded {timeout_seconds}s timeout"
},
"execution_time_ms": (time.time() - start_time) * 1000
}
except Exception as e:
return {
"status": "error",
"data": None,
"error": {
"type": type(e). __name__ ,
"message": str(e)
},
"execution_time_ms": (time.time() - start_time) * 1000
}
ããŒã«èšèšã®äž»èŠåå
LangChainå ¬åŒããã¥ã¡ã³ããã:
- ã·ã³ãã«ã§çãã¹ã³ãŒããããããŒã« ã¯ãè€éãªããŒã«ãããã¢ãã«ã䜿ãããã
- ããéžã°ããååãšèª¬æ ã¯ãã¢ãã«ã®ããã©ãŒãã³ã¹ãå€§å¹ ã«åäžããã
-
@toolãã³ã¬ãŒã¿ãŒãäœ¿çš - ååã説æãåŒæ°ãèªåçã«æšæž¬ - æ§é åãããããŒã¿ãè¿ã - åžžã«statusãdataãerrorãã£ãŒã«ããå«ãã
- ã¿ã€ã ã¢ãŠããšãªãã©ã€ãå®è£ - æ¬çªç°å¢ã·ã¹ãã ã¯å埩åãå¿ èŠ
åæå®è¡ã®ããã®LangGraph ToolNode
LangGraphã®ãã©ãŒæ©èœã®1ã€: ããã©ã«ãã§ãšã©ãŒãåŠçããªããè€æ°ã®ããŒã«ãåæå®è¡ :
from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage
# ããŒã«ãå®çŸ©
tools = [query_database, call_external_api, process_document]
# ToolNodeãäœæ - èªåçã«äžŠè¡æ§ãåŠç
tool_node = ToolNode(tools)
# ã°ã©ãå
graph.add_node("tools", tool_node)
# éæ³: LangGraphã¯ãäºãã«äŸåããªãè€æ°ã®ããŒã«åŒã³åºãã
# 䞊åã§å®è¡ããã¬ã€ãã³ã·ãŒãåçã«åæž
ããã¯ãèªåã§æ£ããæ§ç¯ããã«ã¯æ°é±éããã ã€ã³ãã©ã¹ãã©ã¯ãã£ã¬ãã«ã®æé©å ã§ãã
ãšã©ãŒåŠç: ä¿¡é Œæ§ã®å
ãããæ®é ·ãªçå®ã§ã: æ¬çªç°å¢ã§ã¯ããšãŒãžã§ã³ãã¯å€±æããŸããåé¡ã¯ããããåªé ã«å€±æãããã壿» çã«å€±æãããã§ãã
æ¬çªç°å¢ã®ä¿¡é Œæ§ã¿ãŒã²ãã
AIãšãŒãžã§ã³ãã®ä¿¡é Œæ§ã«é¢ããæ¥çç ç©¶ã«ãããš:
- ããŒã«åŒã³åºããšã©ãŒç: 3%æªæºãäžæ£ãªãã©ã¡ãŒã¿ãŒã«ãããã®ã¯1%æªæº
- P95ã¬ã€ãã³ã·ãŒ: ã·ã³ã°ã«ã¿ãŒã³ã§5ç§æªæº
- ã«ãŒãå°ã蟌ãç: 99%以äž(ç¡éã«ãŒããé²ã)
- ã°ã¬ãŒã¹ãã«ãã°ã©ããŒã·ã§ã³: ã·ã¹ãã ã¯ã¯ã©ãã·ã¥ã§ã¯ãªãããã¯ã¢ããã«ç§»è¡ãã¹ã
ãšã©ãŒåŠçã¢ãŒããã¯ãã£
from enum import Enum
from typing import Optional, Callable, TypeVar
import asyncio
from functools import wraps
T = TypeVar('T')
class ErrorSeverity(Enum):
RECOVERABLE = "recoverable" # ããã¯ãªãã§å詊è¡
DEGRADABLE = "degradable" # ããã·ã³ãã«ãªã¢ãã«ã«ãã©ãŒã«ããã¯
FATAL = "fatal" # é«é倱æã人éã«èŠå
class ProductionErrorHandler:
"""
ãªãã©ã€ãããã¯ãªããã°ã¬ãŒã¹ãã«ãã°ã©ããŒã·ã§ã³ãåããæ¬çªç°å¢å¯Ÿå¿ã®ãšã©ãŒåŠçã
æ¬çªç°å¢ã®AIã·ã¹ãã ã®60%ãä¿¡é Œæ§ã®ããã«äœ¿çšããŠããŸãã
"""
def __init__ (
self,
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0
):
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
async def with_retry(
self,
func: Callable[..., T],
*args,
severity: ErrorSeverity = ErrorSeverity.RECOVERABLE,
**kwargs
) -> T:
"""ææ°ããã¯ãªããªãã©ã€ããžãã¯ã§é¢æ°ãå®è¡ããŸãã"""
last_exception = None
for attempt in range(self.max_retries):
try:
return await func(*args, **kwargs)
except Exception as e:
last_exception = e
# èŽåœçãªãšã©ãŒã¯å詊è¡ãããŸãã
if severity == ErrorSeverity.FATAL:
raise
# ææ°ããã¯ãªããèšç®
delay = min(
self.base_delay * (2 ** attempt),
self.max_delay
)
# 芳枬æ§ã®ããã®ãã°
self._log_retry(attempt, delay, e)
# å詊è¡åã«åŸ
æ©
await asyncio.sleep(delay)
# ãã¹ãŠã®å詊è¡ã䜿ãæããããŸãã
if severity == ErrorSeverity.DEGRADABLE:
return await self._graceful_degradation(*args, **kwargs)
raise last_exception
async def _graceful_degradation(self, *args, **kwargs):
"""
ããã·ã³ãã«ã§ä¿¡é Œæ§ã®é«ãã¢ãããŒãã«ãã©ãŒã«ããã¯ã
äŸ: Claude 4 Opusã倱æããå ŽåãSonnetã«ãã©ãŒã«ããã¯ã
"""
# ãŠãŒã¹ã±ãŒã¹ã«åºæã®å®è£
pass
def _log_retry(self, attempt: int, delay: float, error: Exception):
"""ç£èŠãšãããã°ã®ããã«å詊è¡ãèšé²ããŸãã"""
print(f"Retry {attempt + 1}/{self.max_retries} after {delay}s: {error}")
# æ¬çªç°å¢ã§ã®äœ¿çš
error_handler = ProductionErrorHandler(max_retries=3)
async def production_agent_call(query: str):
try:
result = await error_handler.with_retry(
agent.ainvoke,
query,
severity=ErrorSeverity.DEGRADABLE
)
return result
except Exception as e:
# ãã¹ãŠã®å埩詊è¡ã倱æ - 人éã«èŠå
await send_alert(f"Agent failure: {e}")
raise
Microsoftã®Agent Frameworkãã¿ãŒã³
Microsoftã®Agent Framework(2025幎çºè¡š)ã¯ãå€§èŠæš¡ãªä¿¡é Œæ§ãåäžãããããã«ãçµã¿èŸŒã¿ã®ãšã©ãŒåŠçããªãã©ã€ãå埩ãæäŸããŸãã éèŠãªæŽå¯: ä¿¡é Œæ§ã¯ã¢ããªã±ãŒã·ã§ã³ã³ãŒãã§ã¯ãªããã€ã³ãã©ã¹ãã©ã¯ãã£ã§ãªããã°ãªããŸãã ã
圌ãã®ã¢ãããŒã:
- ææ°ããã¯ãªãã䜿çšãã èªåãªãã©ã€ããžãã¯
- ã«ã¹ã±ãŒãé害ãé²ã ãµãŒããããã¬ãŒã«ãŒ
- 倱æãããšãŒãžã§ã³ããäžæåæ¢ãã ãã«ã¹ãã§ãã¯
- 芳枬æ§ã®ããã®OpenTelemetryãšã® ãã¬ã¡ããªçµ±å
ç£èŠãšèŠ³æž¬æ§: æ¬çªç°å¢ã®å¿ é äºé
枬å®ããªããã®ã¯æ¹åã§ããŸãããæ¬çªç°å¢ã®AIã·ã¹ãã ã§ã¯ãç£èŠã¯ãªãã·ã§ã³ã§ã¯ãããŸãããåç¶ã«é¢ãããŸãã
éèŠãªã¡ããªã¯ã¹
æ¬çªç°å¢ãšãŒãžã§ã³ãç ç©¶ã«åºã¥ããŠããŸã:
from dataclasses import dataclass
from datetime import datetime
from typing import Dict, List
@dataclass
class AgentMetrics:
"""ãã¹ãŠã®AIãšãŒãžã§ã³ãã远跡ãã¹ãæ¬çªç°å¢ã¡ããªã¯ã¹ã"""
# ã¬ã€ãã³ã·ãŒã¡ããªã¯ã¹
p50_latency_ms: float
p95_latency_ms: float
p99_latency_ms: float
# ä¿¡é Œæ§ã¡ããªã¯ã¹
success_rate: float
tool_call_error_rate: float
loop_containment_rate: float
# ããŒã¯ã³äœ¿çšé(ã³ã¹ã远跡)
total_input_tokens: int
total_output_tokens: int
estimated_cost_usd: float
# ãšã©ãŒãã¿ãŒã³
error_types: Dict[str, int]
failed_tools: Dict[str, int]
# ããã©ãŒãã³ã¹
avg_tools_per_request: float
cache_hit_rate: float
timestamp: datetime = datetime.now()
OpenTelemetryçµ±å
LangChainã¯ãOpenTelemetryã®è²¢ç®ã«ãããæšæºåããããã¬ãŒã·ã³ã°ãšãã¬ã¡ããªãæäŸãããã«ããšãŒãžã§ã³ãã®èŠ³æž¬æ§ã匷åããŸãã:
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
# OpenTelemetryãã»ããã¢ãã
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer( __name__ )
# ãšã¯ã¹ããŒã¿ãŒãèšå®(DatadogãNew Relicãªã©)
otlp_exporter = OTLPSpanExporter(endpoint="your-telemetry-endpoint")
span_processor = BatchSpanProcessor(otlp_exporter)
trace.get_tracer_provider().add_span_processor(span_processor)
# ãšãŒãžã§ã³ããèšæž¬
@tracer.start_as_current_span("agent_execution")
async def instrumented_agent_call(query: str):
span = trace.get_current_span()
span.set_attribute("query_length", len(query))
try:
result = await agent.ainvoke(query)
span.set_attribute("success", True)
span.set_attribute("tool_calls", len(result.tool_calls))
return result
except Exception as e:
span.set_attribute("success", False)
span.set_attribute("error", str(e))
raise
ããã«ããã ãšãŒãžã§ã³ãã®åäœãã¿ãŒã³ãçºå±ããã«ã€ããŠå³åº§ã«æŽå¯ ãåŸãããŸããæ¬çªç°å¢ã€ã³ã·ãã³ãããããã°ããæ°é±éåŸã§ã¯ãªãã
æ¬çªç°å¢ãããã€ã¯ãŒã¯ãããŒ
Claude(ãã¹ãŠã®æ¬çªç°å¢AIã«é©çšå¯èœ)ã®Anthropicã®æšå¥šãããã€ããã»ã¹:
- çµ±åèšèš - ã¬ã€ãã³ã·ãŒ/ã³ã¹ã/å質ã®ãã¬ãŒããªãã«åºã¥ããŠã¢ãã«ãšæ©èœãéžæ
- ããŒã¿æºå - ãã¬ããžããŒã¹ãããŒã¿ããŒã¹ãããŒã«ã¹ããŒããã¯ãªãŒã³ã¢ããããŠæ§é å
- ããã³ããéçº - Anthropic WorkbenchãŸãã¯åæ§ã®ããŒã«ã䜿çšããŠè©äŸ¡ã§å埩
- å®è£ - ã·ã¹ãã ãšçµ±åãã人éã«ããã«ãŒãèŠä»¶ãå®çŸ©
- ãã¹ããšã¬ããããŒãã³ã° - æµå¯Ÿçå ¥åãéç¶ãšããããŒã¿ãäžå®å®ãªããŒã«ãã·ãã¥ã¬ãŒã
- A/Bãã¹ã - æ¢åã®ã·ã¹ãã ãšäžŠè¡ããŠãããã€ããæ¹åãæž¬å®
- æ¬çªç°å¢ããã〠- å®å šãªç£èŠãšã¢ã©ãŒãã§ãããã€
éèŠãªæŽå¯: ãšãŒãžã§ã³ãã¯æ¬çªç°å¢ã®åã«æµå¯Ÿçãã¹ãã«åæ Œããå¿ èŠããããŸã ãéç¶ãšããå ¥åãææ§ãªãªã¯ãšã¹ããã·ãã¥ã¬ãŒãããã倱æã§ãã¹ãããŸãã
ããžã¥ã¢ã«ã¢ãŒããã¯ãã£ã®äŸ
ãããã®æŠå¿µãèŠèŠåããããã«ãæ¬çªç°å¢ã®AIãšãŒãžã§ã³ãã·ã¹ãã ã瀺ãäž»èŠãªã¢ãŒããã¯ãã£å³ãããã€ã瀺ããŸã:
ãã«ããšãŒãžã§ã³ãã·ã¹ãã ã¢ãŒããã¯ãã£
æ¬çªç°å¢ã®AIãšãŒãžã§ã³ãã·ã¹ãã ã¯ãäžç·ã«åäœããå°éã³ã³ããŒãã³ããæã€æç¢ºãªã¢ãŒããã¯ãã£ãã¿ãŒã³ã«åŸããŸã:
ãã®é¢å¿ã®åé¢ã«ãããåã³ã³ããŒãã³ããåå¥ã«ãã¹ããç£èŠãæé©åã§ããŸãã
ã¢ãã«ã«ãŒãã£ã³ã°æ±ºå®ãããŒ
ãªã¯ãšã¹ããã·ã¹ãã ã«å ¥ããšãã«ãŒãã£ã³ã°ããžãã¯ã¯æ¬¡ãè©äŸ¡ããŸã:
ãã®ã€ã³ããªãžã§ã³ããªã«ãŒãã£ã³ã°ã¯ãå質ãç¶æããªãããå¿çæéãšéçšã³ã¹ãã®äž¡æ¹ãæé©åããŸãã
ãšã©ãŒåŠçãšã°ã¬ãŒã¹ãã«ãã°ã©ããŒã·ã§ã³
æ¬çªç°å¢ã®ãšã©ãŒåŠçã¯ããŠã©ãŒã¿ãŒãã©ãŒã«ãã¿ãŒã³ã«åŸããŸã:
åã¹ãããã¯ãæåçãã¬ã€ãã³ã·ãŒããšã©ãŒã¿ã€ãã远跡ããã¡ããªã¯ã¹ã§èšæž¬ãããŠããŸãã
åé²ãžã®é: ä¿¡é Œæ§ã®é«ãAIã·ã¹ãã ã®æ§ç¯
AIãšãŒãžã§ã³ãã®é©åœã¯ãããããããããšãŒãžã§ã³ãçãã«ããããšã§ã¯ãªãã ããä¿¡é Œæ§ã®é«ã ãã®ã«ããããšã§ãããã®åéã®åè ã¯ãé©åãªãšã©ãŒåŠçãç£èŠããã¹ãããã©ãŒã«ããã¯ã¡ã«ããºã ãåããçå£ãªãœãããŠã§ã¢ãšã³ãžãã¢ãªã³ã°ãããžã§ã¯ããšããŠAIãšãŒãžã§ã³ããæ±ãããŒã ã§ãã
LangChainãšLangGraphã¯ããŒã«ãæäŸããŸãããã«ãã¢ãã«ãªãŒã±ã¹ãã¬ãŒã·ã§ã³ã¯æè»æ§ãæäŸããŸããæ¬çªç°å¢å¯Ÿå¿ã®ããã³ãããšã³ãžãã¢ãªã³ã°ã¯å¶åŸ¡ãæäŸããŸãããšã©ãŒåŠçã¯å埩åãæäŸããŸãã
ããããæçµçã« ä¿¡é Œæ§ã¯éžæã§ã ãéçºãé ãããã«ããããããããªãã©ã€ãå®è£ ããããšãéžæããããšã§ããè€éãã远å ããã«ããããããããã¬ã¡ããªã远å ããããšãéžæããããšã§ããäžå¿«ã§ããã«ãããããããæµå¯Ÿçå ¥åã§ãã¹ãããããšãéžæããããšã§ãã
æªæ¥ã¯ãå€§èŠæš¡ã«ç¢ºå®ã«åäœããAIã·ã¹ãã ã«å±ããŠããŸããäžç·ã«æ§ç¯ããŸãããã
éèŠãªãã€ã³ã
- æ¬çªç°å¢ã«ã¯ çã®LangChainããLangGraph - æ°žç¶çãªå®è¡ãšãã现ããå¶åŸ¡ãéèŠ
- ãã«ãã¢ãã«ã«ãŒãã£ã³ã° ã¯æŠç¥çåªäœæ§ - åã¿ã¹ã¯ã«é©åãªã¢ãã«ã䜿çš
- ããã³ãããšã³ãžãã¢ãªã³ã°ã¯APIã³ã³ãã©ã¯ã - ãã¹ãŠã®ããã³ããããã¹ããããŒãžã§ã³ç®¡çãç£èŠ
- ããŒã«åŒã³åºãã«ã¯æ¬çªç°å¢ãã¿ãŒã³ãå¿ èŠ - ã¿ã€ã ã¢ãŠãããªãã©ã€ãæ§é åãããåºåããšã©ãŒåŠç
- ãšã©ãŒåŠçã¯ãªãã·ã§ã³ã§ã¯ãªã - 3%æªæºã®ããŒã«ãšã©ãŒçãš5ç§æªæºã®P95ã¬ã€ãã³ã·ãŒãç®æã
- 芳枬æ§ã¯åç¶ã«é¢ãã - 忥ããOpenTelemetryãå®è£
- ä¿¡é Œæ§ã¿ãŒã²ãã ã¯æç€ºçã§ç¶ç¶çã«æž¬å®ãããå¿ èŠããã
åèæç®ãšãããªãèªæž
: [1] Galileo AI. (2025). "A Guide to AI Agent Reliability for Mission Critical Systems." https://galileo.ai/blog/ai-agent-reliability-strategies
: [2] Beam AI. (2025). "Production-Ready AI Agents: The Design Principles That Actually Work." https://beam.ai/agentic-insights/production-ready-ai-agents-the-design-principles-that-actually-work
: [3] LangChain Blog. (2025). "LangChain & Multi-Agent AI in 2025: Framework, Tools & Use Cases." https://blogs.infoservices.com/artificial-intelligence/langchain-multi-agent-ai-framework-2025/
: [4] LangChain Blog. (2025). "Building LangGraph: Designing an Agent Runtime from first principles." https://blog.langchain.com/building-langgraph/
: [5] LangChain Documentation. (2025). "Agents - Conceptual Guide." https://python.langchain.com/docs/concepts/agents/
: [6] LangChain Blog. (2025). "LangGraph: Multi-Agent Workflows." https://blog.langchain.com/langgraph-multi-agent-workflows/
: [7] Waveloom. (2025). "Building Multi-Model AI Agents: Combining GPT, Claude, and RAG." https://www.waveloom.dev/blog/building-multi-model-ai-agents-combining-gpt-claude-and-rag
: [8] Medium - Devansh. (2025). "GPT vs Claude vs Gemini for Agent Orchestration." https://machine-learning-made-simple.medium.com/gpt-vs-claude-vs-gemini-for-agent-orchestration-b3fbc584f0f7
: [9] Bind AI IDE. (2025). "OpenAI GPT-5 vs Claude 4 Feature Comparison." https://blog.getbind.co/2025/08/04/openai-gpt-5-vs-claude-4-feature-comparison/
: [10] OpenAI Cookbook. (2025). "GPT-4.1 Prompting Guide." https://cookbook.openai.com/examples/gpt4-1_prompting_guide
: [11] Langflow. (2025). "Build Your Own GPT-5: Smart Model Routing with Langflow." https://www.langflow.org/blog/how-to-build-your-own-gpt-5
: [12] OpenAI Platform. (2025). "Prompt Engineering - Best Practices." https://platform.openai.com/docs/guides/prompt-engineering
: [13] Anthropic. (2025). "Get to production faster with the upgraded Anthropic Console." https://www.anthropic.com/news/upgraded-anthropic-console
: [14] Anthropic. (2025). "Claude API Usage and Best Practices." https://support.anthropic.com/en/collections/9811458-api-usage-and-best-practices
: [15] OpenAI Help Center. (2025). "Best practices for prompt engineering with the OpenAI API." https://help.openai.com/en/articles/6654000-best-practices-for-prompt-engineering-with-the-openai-api
: [16] Anthropic Documentation. (2025). "Home - Claude Docs." https://docs.anthropic.com/en/home
: [17] OpenAI Cookbook. (2025). "GPT-5 Prompting Guide." https://cookbook.openai.com/examples/gpt-5/gpt-5_prompting_guide
: [18] LangChain Documentation. (2025). "Tool Calling - Concepts." https://python.langchain.com/docs/concepts/tool_calling/
: [19] LangGraph Documentation. (2025). "Call tools - How-to Guide." https://langchain-ai.github.io/langgraph/how-tos/tool-calling/
: [20] Microsoft Azure Blog. (2025). "Introducing Microsoft Agent Framework." https://azure.microsoft.com/en-us/blog/introducing-microsoft-agent-framework/
: [21] Galileo AI. (2025). "AI Agent Reliability: The Playbook for Production-Ready Systems." https://www.getmaxim.ai/articles/ai-agent-reliability-the-long-term-playbook-for-production-ready-systems/
: [22] DEV Community. (2025). "The 12-Factor Agent: A Practical Framework for Building Production AI Systems." https://dev.to/bredmond1019/the-12-factor-agent-a-practical-framework-for-building-production-ai-systems-3oo8
: [23] Medium - Data Science Collective. (2025). "How to Build Production Ready AI Agents in 5 Steps." https://medium.com/data-science-collective/why-most-ai-agents-fail-in-production-and-how-to-build-ones-that-dont-f6f604bcd075
: [24] Anthropic. (2025). "Anthropic Academy: Claude API Development Guide." https://www.anthropic.com/learn/build-with-claude
: [25] Anthropic. (2025). "Building Effective AI Agents." https://www.anthropic.com/research/building-effective-agents
æ¬çªç°å¢ã®AIãã¿ãŒã³ã«ã€ããŠè°è«ãããããªãŒã±ã¹ãã¬ãŒã·ã§ã³ã®èª²é¡ãå ±æãããã§ãã? Kanaeru AIããŒã ãšã€ãªãããŸããããç§ãã¡ã¯ãããã®ããšãæ¥ã å®è·µããŠããŸãã
Originally published at kanaeru.ai



Top comments (0)