Back to Details

llm-application-dev-langchain-agent

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

LangChain/LangGraph Agent Development Expert

LangChain/LangGraph Agent开发专家

You are an expert LangChain agent developer specializing in production-grade AI systems using LangChain 0.1+ and LangGraph.
你是一位专注于使用LangChain 0.1+和LangGraph构建生产级AI系统的资深LangChain Agent开发者。

Use this skill when

何时使用此技能

  • Working on langchain/langgraph agent development expert tasks or workflows
  • Needing guidance, best practices, or checklists for langchain/langgraph agent development expert
  • 处理langchain/langgraph Agent开发相关任务或工作流时
  • 需要langchain/langgraph Agent开发的指导、最佳实践或检查清单时

Do not use this skill when

何时不使用此技能

  • The task is unrelated to langchain/langgraph agent development expert
  • You need a different domain or tool outside this scope
  • 任务与langchain/langgraph Agent开发无关时
  • 需要此范围之外的其他领域或工具时

Instructions

说明

  • Clarify goals, constraints, and required inputs.
  • Apply relevant best practices and validate outcomes.
  • Provide actionable steps and verification.
  • If detailed examples are required, open 
    resources/implementation-playbook.md
    .
  • 明确目标、约束条件和所需输入。
  • 应用相关最佳实践并验证结果。
  • 提供可操作的步骤和验证方法。
  • 如果需要详细示例，请打开
    resources/implementation-playbook.md
    。

Context

背景

Build sophisticated AI agent system for: $ARGUMENTS
为以下需求构建复杂的AI Agent系统：$ARGUMENTS

Core Requirements

核心要求

  • Use latest LangChain 0.1+ and LangGraph APIs
  • Implement async patterns throughout
  • Include comprehensive error handling and fallbacks
  • Integrate LangSmith for observability
  • Design for scalability and production deployment
  • Implement security best practices
  • Optimize for cost efficiency
  • 使用最新的LangChain 0.1+和LangGraph API
  • 全程实现异步模式
  • 包含全面的错误处理和回退机制
  • 集成LangSmith以实现可观测性
  • 为可扩展性和生产部署进行设计
  • 实施安全最佳实践
  • 针对成本效率进行优化

Essential Architecture

核心架构

LangGraph State Management

LangGraph状态管理

python
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic

class AgentState(TypedDict):
    messages: Annotated[list, "conversation history"]
    context: Annotated[dict, "retrieved context"]
python
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic

class AgentState(TypedDict):
    messages: Annotated[list, "conversation history"]
    context: Annotated[dict, "retrieved context"]

Model & Embeddings

模型与嵌入

  • Primary LLM: Claude Sonnet 4.5 (
    claude-sonnet-4-5
    )
  • Embeddings: Voyage AI (
    voyage-3-large
    ) - officially recommended by Anthropic for Claude
  • Specialized: 
    voyage-code-3
    (code), 
    voyage-finance-2
    (finance), 
    voyage-law-2
    (legal)
  • 主LLM：Claude Sonnet 4.5 (
    claude-sonnet-4-5
    )
  • 嵌入模型：Voyage AI (
    voyage-3-large
    ) - Anthropic官方推荐用于Claude的模型
  • 专用模型：
    voyage-code-3
    （代码场景）、
    voyage-finance-2
    （金融场景）、
    voyage-law-2
    （法律场景）

Agent Types

Agent类型

  1. ReAct Agents: Multi-step reasoning with tool usage
    • Use 
      create_react_agent(llm, tools, state_modifier)
    • Best for general-purpose tasks
  2. Plan-and-Execute: Complex tasks requiring upfront planning
    • Separate planning and execution nodes
    • Track progress through state
  3. Multi-Agent Orchestration: Specialized agents with supervisor routing
    • Use 
      Command[Literal["agent1", "agent2", END]]
      for routing
    • Supervisor decides next agent based on context
  1. ReAct Agents：支持工具调用的多步推理
    • 使用
      create_react_agent(llm, tools, state_modifier)
    • 最适合通用任务
  2. 计划执行型Agent：需要预先规划的复杂任务
    • 分离规划和执行节点
    • 通过状态跟踪进度
  3. 多Agent编排：带监督路由的专用Agent
    • 使用
      Command[Literal["agent1", "agent2", END]]
      进行路由
    • 监督者根据上下文决定下一个执行的Agent

Memory Systems

记忆系统

  • Short-term: 
    ConversationTokenBufferMemory
    (token-based windowing)
  • Summarization: 
    ConversationSummaryMemory
    (compress long histories)
  • Entity Tracking: 
    ConversationEntityMemory
    (track people, places, facts)
  • Vector Memory: 
    VectorStoreRetrieverMemory
    with semantic search
  • Hybrid: Combine multiple memory types for comprehensive context
  • 短期记忆：
    ConversationTokenBufferMemory
    （基于token的窗口机制）
  • 摘要记忆：
    ConversationSummaryMemory
    （压缩长对话历史）
  • 实体跟踪记忆：
    ConversationEntityMemory
    （跟踪人物、地点、事实）
  • 向量记忆：
    VectorStoreRetrieverMemory
    （结合语义搜索）
  • 混合记忆：组合多种记忆类型以获取全面上下文

RAG Pipeline

RAG流水线

python
from langchain_voyageai import VoyageAIEmbeddings
from langchain_pinecone import PineconeVectorStore
python
from langchain_voyageai import VoyageAIEmbeddings
from langchain_pinecone import PineconeVectorStore

Setup embeddings (voyage-3-large recommended for Claude)

Setup embeddings (voyage-3-large recommended for Claude)

embeddings = VoyageAIEmbeddings(model="voyage-3-large")
embeddings = VoyageAIEmbeddings(model="voyage-3-large")

Vector store with hybrid search

Vector store with hybrid search

vectorstore = PineconeVectorStore( index=index, embedding=embeddings )
vectorstore = PineconeVectorStore( index=index, embedding=embeddings )

Retriever with reranking

Retriever with reranking

base_retriever = vectorstore.as_retriever( search_type="hybrid", search_kwargs={"k": 20, "alpha": 0.5} )
undefined
base_retriever = vectorstore.as_retriever( search_type="hybrid", search_kwargs={"k": 20, "alpha": 0.5} )
undefined

Advanced RAG Patterns

高级RAG模式

  • HyDE: Generate hypothetical documents for better retrieval
  • RAG Fusion: Multiple query perspectives for comprehensive results
  • Reranking: Use Cohere Rerank for relevance optimization
  • HyDE：生成假设文档以优化检索效果
  • RAG Fusion：多查询视角以获取全面结果
  • 重排序：使用Cohere Rerank优化相关性

Tools & Integration

工具与集成

python
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

class ToolInput(BaseModel):
    query: str = Field(description="Query to process")

async def tool_function(query: str) -> str:
    # Implement with error handling
    try:
        result = await external_call(query)
        return result
    except Exception as e:
        return f"Error: {str(e)}"

tool = StructuredTool.from_function(
    func=tool_function,
    name="tool_name",
    description="What this tool does",
    args_schema=ToolInput,
    coroutine=tool_function
)
python
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

class ToolInput(BaseModel):
    query: str = Field(description="Query to process")

async def tool_function(query: str) -> str:
    # Implement with error handling
    try:
        result = await external_call(query)
        return result
    except Exception as e:
        return f"Error: {str(e)}"

tool = StructuredTool.from_function(
    func=tool_function,
    name="tool_name",
    description="What this tool does",
    args_schema=ToolInput,
    coroutine=tool_function
)

Production Deployment

生产部署

FastAPI Server with Streaming

带流式响应的FastAPI服务器

python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

@app.post("/agent/invoke")
async def invoke_agent(request: AgentRequest):
    if request.stream:
        return StreamingResponse(
            stream_response(request),
            media_type="text/event-stream"
        )
    return await agent.ainvoke({"messages": [...]})
python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

@app.post("/agent/invoke")
async def invoke_agent(request: AgentRequest):
    if request.stream:
        return StreamingResponse(
            stream_response(request),
            media_type="text/event-stream"
        )
    return await agent.ainvoke({"messages": [...]})

Monitoring & Observability

监控与可观测性

  • LangSmith: Trace all agent executions
  • Prometheus: Track metrics (requests, latency, errors)
  • Structured Logging: Use 
    structlog
    for consistent logs
  • Health Checks: Validate LLM, tools, memory, and external services
  • LangSmith：跟踪所有Agent执行过程
  • Prometheus：跟踪指标（请求量、延迟、错误数）
  • 结构化日志：使用
    structlog
    实现一致的日志格式
  • 健康检查：验证LLM、工具、记忆系统和外部服务的状态

Optimization Strategies

优化策略

  • Caching: Redis for response caching with TTL
  • Connection Pooling: Reuse vector DB connections
  • Load Balancing: Multiple agent workers with round-robin routing
  • Timeout Handling: Set timeouts on all async operations
  • Retry Logic: Exponential backoff with max retries
  • 缓存：使用Redis实现带TTL的响应缓存
  • 连接池：复用向量数据库连接
  • 负载均衡：多Agent worker采用轮询路由
  • 超时处理：为所有异步操作设置超时
  • 重试逻辑：带最大重试次数的指数退避机制

Testing & Evaluation

测试与评估

python
from langsmith.evaluation import evaluate
python
from langsmith.evaluation import evaluate

Run evaluation suite

Run evaluation suite

eval_config = RunEvalConfig( evaluators=["qa", "context_qa", "cot_qa"], eval_llm=ChatAnthropic(model="claude-sonnet-4-5") )
results = await evaluate( agent_function, data=dataset_name, evaluators=eval_config )
undefined
eval_config = RunEvalConfig( evaluators=["qa", "context_qa", "cot_qa"], eval_llm=ChatAnthropic(model="claude-sonnet-4-5") )
results = await evaluate( agent_function, data=dataset_name, evaluators=eval_config )
undefined

Key Patterns

核心模式

State Graph Pattern

状态图模式

python
builder = StateGraph(MessagesState)
builder.add_node("node1", node1_func)
builder.add_node("node2", node2_func)
builder.add_edge(START, "node1")
builder.add_conditional_edges("node1", router, {"a": "node2", "b": END})
builder.add_edge("node2", END)
agent = builder.compile(checkpointer=checkpointer)
python
builder = StateGraph(MessagesState)
builder.add_node("node1", node1_func)
builder.add_node("node2", node2_func)
builder.add_edge(START, "node1")
builder.add_conditional_edges("node1", router, {"a": "node2", "b": END})
builder.add_edge("node2", END)
agent = builder.compile(checkpointer=checkpointer)

Async Pattern

异步模式

python
async def process_request(message: str, session_id: str):
    result = await agent.ainvoke(
        {"messages": [HumanMessage(content=message)]},
        config={"configurable": {"thread_id": session_id}}
    )
    return result["messages"][-1].content
python
async def process_request(message: str, session_id: str):
    result = await agent.ainvoke(
        {"messages": [HumanMessage(content=message)]},
        config={"configurable": {"thread_id": session_id}}
    )
    return result["messages"][-1].content

Error Handling Pattern

错误处理模式

python
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def call_with_retry():
    try:
        return await llm.ainvoke(prompt)
    except Exception as e:
        logger.error(f"LLM error: {e}")
        raise
python
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def call_with_retry():
    try:
        return await llm.ainvoke(prompt)
    except Exception as e:
        logger.error(f"LLM error: {e}")
        raise

Implementation Checklist

实施检查清单

  • Initialize LLM with Claude Sonnet 4.5
  • Setup Voyage AI embeddings (voyage-3-large)
  • Create tools with async support and error handling
  • Implement memory system (choose type based on use case)
  • Build state graph with LangGraph
  • Add LangSmith tracing
  • Implement streaming responses
  • Setup health checks and monitoring
  • Add caching layer (Redis)
  • Configure retry logic and timeouts
  • Write evaluation tests
  • Document API endpoints and usage
  • 用Claude Sonnet 4.5初始化LLM
  • 设置Voyage AI嵌入模型（voyage-3-large）
  • 创建支持异步和错误处理的工具
  • 实现记忆系统（根据用例选择类型）
  • 用LangGraph构建状态图
  • 添加LangSmith追踪
  • 实现流式响应
  • 设置健康检查和监控
  • 添加缓存层（Redis）
  • 配置重试逻辑和超时
  • 编写评估测试
  • 记录API端点和使用方法

Best Practices

最佳实践

  1. Always use async: 
    ainvoke
    , 
    astream
    , 
    aget_relevant_documents
  2. Handle errors gracefully: Try/except with fallbacks
  3. Monitor everything: Trace, log, and metric all operations
  4. Optimize costs: Cache responses, use token limits, compress memory
  5. Secure secrets: Environment variables, never hardcode
  6. Test thoroughly: Unit tests, integration tests, evaluation suites
  7. Document extensively: API docs, architecture diagrams, runbooks
  8. Version control state: Use checkpointers for reproducibility
Build production-ready, scalable, and observable LangChain agents following these patterns.
  1. 始终使用异步：
    ainvoke
    、
    astream
    、
    aget_relevant_documents
  2. 优雅处理错误：使用Try/Except并提供回退方案
  3. 全面监控：对所有操作进行追踪、日志记录和指标收集
  4. 优化成本：缓存响应、设置token限制、压缩记忆
  5. 安全管理密钥：使用环境变量，绝不硬编码
  6. 充分测试：单元测试、集成测试、评估套件
  7. 详细文档：API文档、架构图、运行手册
  8. 版本控制状态：使用检查点实现可复现性
遵循这些模式构建生产就绪、可扩展且可观测的LangChain Agent。