LangChain · LangGraph · LangSmith 面经

📖 "Agent = Model + Harness. LangChain provides create_agent: a minimal, highly configurable harness. The harness is everything around the model loop: the prompt, the tools, and any middleware that shapes behavior." — LangChain overview

📖 "Start with the primitives and compose exactly what your use case needs." — LangChain overview

from langchain.agents import create_agent

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    return f"It's always sunny in {city}!"

agent = create_agent(
    model="claude-sonnet-4-6",
    tools=[get_weather],
    system_prompt="You are a helpful assistant",
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "What's the weather in San Francisco?"}]}
)
print(result["messages"][-1].content_blocks)

📖 "An agent is a model calling tools in a loop until a given task is complete." — Agents

📖 "The job of a harness: get the model the right context at the right time for the given task." — Agents

📖 "A harness is everything around that loop: the model, its prompt, its tools, and any middleware that shapes its behavior." — Agents

flowchart LR
    U[User message] --> M[Model]
    M -->|tool_calls| T[Tools]
    T -->|results| M
    M -->|task complete| A[Final answer]

📖 "Pass a model identifier string ("provider:model") or an initialized model instance to select the model for your agent." — Agents

📖 "To provide the agent with tools, pass any Python callable, LangChain tool, or tool dict." — Agents

📖 "Return a validated schema from the agent using response_format=." — Agents

from pydantic import BaseModel
from langchain.agents import create_agent


class Answer(BaseModel):
    summary: str
    confidence: float


agent = create_agent(model="anthropic:claude-sonnet-4-6", tools=tools, response_format=Answer)
result = agent.invoke({"messages": [{"role": "user", "content": "Summarize AI trends"}]})
result["structured_response"]  # Answer(summary=..., confidence=...)

📖 "invoke returns the final response at the end of a run. If an agent executes multiple tool calls, users often need progress updates before completion. Use streaming to surface intermediate messages and tool activity as they happen." — Agents

📖 "thread_id scopes the *conversation* (message history, checkpoints), while context carries *per-run* data your tools and middleware read at invocation time. Both are commonly passed together." — Agents

from langchain.agents import create_agent
from langchain_core.utils.uuid import uuid7
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": str(uuid7())}}

result = agent.invoke(
    {"messages": [{"role": "user", "content": "What's the weather in San Francisco?"}]},
    config=config,
)

# A follow-up turn on the same conversation: reuse the same thread_id to keep history
result = agent.invoke(
    {"messages": [{"role": "user", "content": "What about tomorrow?"}]},
    config=config,
)

📖 "Runtime context provides dependency injection for your tools and middleware." — Runtime

📖 "static information like user id, db connections, or other dependencies for an agent invocation" — Runtime

📖 "server_info is None when not running on LangGraph Server (e.g., during local development)." — Runtime

from dataclasses import dataclass
from langchain.tools import tool, ToolRuntime

@dataclass
class Context:
    user_id: str

@tool
def fetch_user_email_preferences(runtime: ToolRuntime[Context]) -> str:
    """Fetch the user's email preferences from the store."""
    user_id = runtime.context.user_id

    preferences: str = "The user prefers you to write a brief and polite email."
    if runtime.store:
        if memory := runtime.store.get(("users",), user_id):
            preferences = memory.value["preferences"]

    return preferences

📖 "create_agent is highly extensible. Middleware is the primitive for customization: each piece handles one concern, hooks into the agent loop at the right moment, and composes freely with any other." — Agents

📖 "Every model call has a fixed context window." — Agents

📖 "Some policies can't live in a prompt—they need to be enforced deterministically regardless of what the model does." — Agents

from langchain.agents import create_agent
from langchain.agents.middleware import ModelRetryMiddleware, ToolRetryMiddleware

agent = create_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[search],
    middleware=[
        ModelRetryMiddleware(max_retries=3),
        ToolRetryMiddleware(max_retries=2),
    ],
)

📖 "Start with Deep Agents for a "batteries-included" agent with features like automatic context compression, a virtual filesystem, and subagent-spawning." — LangChain overview

📖 "Use LangGraph, our low-level orchestration framework, for advanced needs combining deterministic and agentic workflows." — LangChain overview

📖 "LangChain's agents are built on top of LangGraph. This allows us to take advantage of LangGraph's durable execution, human-in-the-loop support, persistence, and more." — LangChain overview

flowchart TD
    Q{需求?} -->|开箱即用、长任务| DA[Deep Agents]
    Q -->|可裁剪 harness、贴合用例| LC[LangChain create_agent]
    Q -->|低层编排、确定性+agentic| LG[LangGraph]
    DA -.built on.-> LC
    LC -.built on.-> LG

📖 "Each provider package implements the same standard interface, so you can swap providers without rewriting application logic." — Models

📖 "Retries use exponential backoff with jitter. Network errors, rate limits (429), and server errors (5xx) are retried automatically. Client errors such as 401 (unauthorized) or 404 are not retried." — Models

from langchain.chat_models import init_chat_model

model = init_chat_model(
    "google_genai:gemini-3.5-flash",
    max_retries=10,  # Increase for unreliable networks (default: 6)
    timeout=120,  # Seconds; increase for slow connections
)

📖 "Models can request to call tools that perform tasks such as fetching data from a database, searching the web, or running code." — Models

📖 "Each ToolMessage returned by the tool includes a tool_call_id that matches the original tool call, helping the model correlate results with requests." — Models

model_with_tools = model.bind_tools([get_weather])

# Step 1: Model generates tool calls
messages = [{"role": "user", "content": "What's the weather in Boston?"}]
ai_msg = model_with_tools.invoke(messages)
messages.append(ai_msg)

# Step 2: Execute tools and collect results
for tool_call in ai_msg.tool_calls:
    tool_result = get_weather.invoke(tool_call)
    messages.append(tool_result)

# Step 3: Pass results back to model for final response
final_response = model_with_tools.invoke(messages)
print(final_response.text)

sequenceDiagram
    participant U as User
    participant M as Model
    participant T as Tools
    U->>M: 提问
    M->>M: 决定需要哪些工具
    M->>T: tool_calls (含 id)
    T-->>M: ToolMessage (tool_call_id 匹配)
    M->>U: 基于结果生成最终回答

📖 "Messages are the fundamental unit of context for models in LangChain. They represent the input and output of models, carrying both the content and metadata needed to represent the state of a conversation when interacting with an LLM." — Messages

📖 "Tool messages are used to pass the results of a single tool execution back to the model." — Messages

📖 "Content blocks were introduced as a new property on messages in LangChain v1 to standardize content formats across providers while maintaining backward compatibility with existing code." — Messages

from langchain.messages import AIMessage, HumanMessage, ToolMessage

ai_message = AIMessage(
    content=[],
    tool_calls=[{"name": "get_weather", "args": {"location": "San Francisco"}, "id": "call_123"}]
)
tool_message = ToolMessage(content="Sunny, 72°F", tool_call_id="call_123")  # Must match the call ID

messages = [
    HumanMessage("What's the weather in San Francisco?"),
    ai_message,   # Model's tool call
    tool_message, # Tool execution result
]
response = model.invoke(messages)

📖 "A chat model must be invoked to generate an output. There are three primary invocation methods, each suited to different use cases." — Models

📖 "Most models can stream their output content while it is being generated. By displaying output progressively, streaming significantly improves user experience, particularly for longer responses." — Models

📖 "Batching a collection of independent requests to a model can significantly improve performance and reduce costs, as the processing can be done in parallel:" — Models

full = None  # None | AIMessageChunk
for chunk in model.stream("What color is the sky?"):
    full = chunk if full is None else full + chunk
    print(full.text)

print(full.content_blocks)
# [{"type": "text", "text": "The sky is typically blue..."}]

📖 "Models can be requested to provide their response in a format matching a given schema. This is useful for ensuring the output can be easily parsed and used in subsequent processing." — Models

📖 "'json_mode': A precursor to 'json_schema' offered by some providers. Generates valid JSON, but the schema must be described in the prompt." — Models

📖 "Pydantic models provide automatic validation. TypedDict and JSON Schema require manual validation." — Models

from pydantic import BaseModel, Field

class Movie(BaseModel):
    """A movie with details."""
    title: str = Field(description="The title of the movie")
    year: int = Field(description="The year the movie was released")
    director: str = Field(description="The director of the movie")
    rating: float = Field(description="The movie's rating out of 10")

model_with_structure = model.with_structured_output(Movie)
response = model_with_structure.invoke("Provide details about the movie Inception")

📖 "Type hints are required as they define the tool's input schema. The docstring should be informative and concise to help the model understand the tool's purpose." — Tools

📖 "Return a string when the tool should provide plain text for the model to read and use in its next response." — Tools

from langchain.tools import tool

@tool
def search_database(query: str, limit: int = 10) -> str:
    """Search the customer database for records matching the query.

    Args:
        query: Search terms to look for
        limit: Maximum number of results to return
    """
    return f"Found {limit} results for '{query}'"

📖 "Add runtime: ToolRuntime to your tool signature to access state. This parameter is automatically injected and hidden from the LLM - it won't appear in the tool's schema." — Tools

📖 "Set return direct on a tool to short-circuit the agent loop: the agent returns the tool's output to the caller immediately, without sending it back through the model for further processing." — Tools

📖 "Because the model does not process the tool's output, return_direct=True is not suitable for tools whose results require further reasoning, summarization, or chaining with other tool calls." — Tools

from langchain.messages import ToolMessage
from langchain.tools import ToolRuntime, tool
from langgraph.types import Command

@tool
def set_language(language: str, runtime: ToolRuntime) -> Command:
    """Set the preferred response language."""
    return Command(
        update={
            "preferred_language": language,
            "messages": [
                ToolMessage(
                    content=f"Language set to {language}.",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
        }
    )

📖 "Structured output allows agents to return data in a specific, predictable format." — Structured output

📖 "Provider-native structured output provides high reliability and strict validation because the model provider enforces the schema. Use it when available." — Structured output

from pydantic import BaseModel, Field
from langchain.agents import create_agent


class ContactInfo(BaseModel):
    """Contact information for a person."""
    name: str = Field(description="The name of the person")
    email: str = Field(description="The email address of the person")
    phone: str = Field(description="The phone number of the person")

agent = create_agent(
    model="gpt-5.5",
    response_format=ContactInfo  # Auto-selects ProviderStrategy
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "Extract contact info from: John Doe, john@example.com, (555) 123-4567"}]
})

print(result["structured_response"])
# ContactInfo(name='John Doe', email='john@example.com', phone='(555) 123-4567')

📖 "Models can make mistakes when generating structured output via tool calling. LangChain provides intelligent retry mechanisms to handle these errors automatically." — Structured output

📖 "Error handling strategy for structured output validation failures. Defaults to True." — Structured output

from langchain.agents.structured_output import StructuredOutputValidationError
from langchain.agents.structured_output import MultipleStructuredOutputsError

def custom_error_handler(error: Exception) -> str:
    if isinstance(error, StructuredOutputValidationError):
        return "There was an issue with the format. Try again."
    elif isinstance(error, MultipleStructuredOutputsError):
        return "Multiple structured outputs were returned. Pick the most relevant one."
    else:
        return f"Error: {str(error)}"

flowchart LR
    U[User input] --> B[before hooks]
    B --> M[Model call]
    M --> T{Tools?}
    T -- yes --> X[Tool execution]
    X --> M
    T -- no --> A[after hooks]
    A --> O[Output]

📖 "Middleware provides a way to more tightly control what happens inside the agent." — Overview

📖 "The core agent loop involves calling a model, letting it choose tools to execute, and then finishing when it calls no more tools:" — Overview

📖 "Middleware is not a separate runtime: hooks run inside the compiled" — Overview

from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.graph import START, StateGraph

# Assumes read_email, send_email, classify_node, and route are defined elsewhere.
email_agent = create_agent(
    model="claude-sonnet-4-6",
    tools=[read_email, send_email],
    middleware=[HumanInTheLoopMiddleware(interrupt_on={"send_email": True})],
)

graph = (
    StateGraph(AgentState)
    .add_node("classify", classify_node)
    .add_node("email_agent", email_agent)
    .add_edge(START, "classify")
    .add_conditional_edges("classify", route)
    .compile()
)

📖 "Guardrails help you build safe, compliant AI applications by validating and filtering content at key points in your agent's execution." — Guardrails

📖 "Use rule-based logic like regex patterns, keyword matching, or explicit checks. Fast, predictable, and cost-effective, but may miss nuanced violations." — Guardrails

📖 "Use LLMs or classifiers to evaluate content with semantic understanding. Catch subtle issues that rules miss, but are slower and more expensive." — Guardrails

from typing import Any

from langchain.agents.middleware import before_agent, AgentState, hook_config
from langgraph.runtime import Runtime

banned_keywords = ["hack", "exploit", "malware"]

@before_agent(can_jump_to=["end"])
def content_filter(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """Deterministic guardrail: Block requests containing banned keywords."""
    if not state["messages"]:
        return None

    first_message = state["messages"][0]
    if first_message.type != "human":
        return None

    content = first_message.content.lower()

    for keyword in banned_keywords:
        if keyword in content:
            return {
                "messages": [{
                    "role": "assistant",
                    "content": "I cannot process requests containing inappropriate content. Please rephrase your request."
                }],
                "jump_to": "end"
            }

    return None

📖 "LangChain provides built-in middleware for detecting and handling Personally Identifiable Information (PII) in conversations. This middleware can detect common PII types like emails, credit cards, IP addresses, and more." — Guardrails

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware


agent = create_agent(
    model="gpt-5.5",
    tools=[customer_service_tool, email_tool],
    middleware=[
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
        PIIMiddleware("api_key", detector=r"sk-[a-zA-Z0-9]{32}", strategy="block", apply_to_input=True),
    ],
)

📖 "LangChain provides built-in middleware for requiring human approval before executing sensitive operations. This is one of the most effective guardrails for high-stakes decisions." — Guardrails

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command


agent = create_agent(
    model="gpt-5.5",
    tools=[search_tool, send_email_tool, delete_database_tool],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "send_email": True,
                "delete_database": True,
                "search": False,
            }
        ),
    ],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "some_id"}}
result = agent.invoke(
    {"messages": [{"role": "user", "content": "Send an email to the team"}]},
    config=config
)
result = agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config=config
)

📖 "Short term memory lets your application remember previous interactions within a single thread or conversation." — Short-term memory

📖 "Unlike short-term memory, which is scoped to a single thread, long-term memory persists across threads and can be recalled at any time." — Long-term memory

flowchart TD
    subgraph 短期记忆
      T1["thread_id=1
state.messages"] --> CP[(checkpointer)]
    end
    subgraph 长期记忆
      S[(store
namespace/key → JSON)]
    end
    A[agent] --> T1
    A -.跨会话读写.-> S
    T2["thread_id=2 全新会话"] -.仍可召回.-> S

📖 "To add short-term memory (thread-level persistence) to an agent, you need to specify a checkpointer when creating an agent." — Short-term memory

📖 "State is persisted to a database (or memory) using a checkpointer so the thread can be resumed at any time." — Short-term memory

from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver

def get_user_info() -> str:
    """Look up information about the current user."""
    return "No user profile on file."

DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup() # auto create tables in PostgreSQL
    agent = create_agent(
        "gpt-5.5",
        tools=[get_user_info],
        checkpointer=checkpointer,
    )

📖 "Even if your model supports the full context length, most LLMs still perform poorly over long contexts." — Short-term memory

📖 "The problem with trimming or removing messages, as shown above, is that you may lose information from culling of the message queue." — Short-term memory

from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model="gpt-5.5",
    tools=[...],
    middleware=[
        SummarizationMiddleware(
            model="gpt-5.4-mini",
            trigger=("tokens", 4000),
            keep=("messages", 20)
        )
    ],
    checkpointer=checkpointer,
)

📖 "Long-term memory is built on LangGraph stores, which save data as JSON documents organized by namespace and key." — Long-term memory

📖 "Each memory is organized under a custom namespace (similar to a folder) and a distinct key (like a file name)." — Long-term memory

📖 "This structure enables hierarchical organization of memories. Cross-namespace searching is then supported through content filters." — Long-term memory

from langgraph.store.base import IndexConfig
from langgraph.store.memory import InMemoryStore

store = InMemoryStore(index=IndexConfig(embed=embed, dims=2))
user_id = "my-user"
application_context = "chitchat"
namespace = (user_id, application_context)
store.put(
    namespace,
    "a-memory",
    {
        "rules": [
            "User likes short, direct language",
            "User only speaks English & python",
        ],
        "my-key": "my-value",
    },
)
item = store.get(namespace, "a-memory")
items = store.search(
    namespace, filter={"my-key": "my-value"}, query="language preferences"
)

📖 "The runtime parameter is hidden from the tool signature (so the model doesn't see it), but the tool can access the state through it." — Short-term memory

📖 "To modify the agent's short-term memory (state) during execution, you can return state updates directly from the tools." — Short-term memory

from langchain.tools import ToolRuntime, tool

@tool
def save_user_info(user_info: UserInfo, runtime: ToolRuntime[Context]) -> str:
    """Save user info."""
    assert runtime.store is not None
    store = runtime.store
    user_id = runtime.context.user_id
    # Store data in the store (namespace, key, data)
    store.put(("users",), user_id, dict(user_info))
    return "Successfully saved user info."

📖 "By default, agents use AgentState to manage short term memory, specifically the conversation history via a messages key." — Short-term memory

📖 "You can extend AgentState to add additional fields." — Short-term memory

from langchain.agents import create_agent, AgentState
from langgraph.checkpoint.memory import InMemorySaver

class CustomAgentState(AgentState):
    user_id: str
    preferences: dict

agent = create_agent(
    "gpt-5.5",
    tools=[get_user_info],
    state_schema=CustomAgentState,
    checkpointer=InMemorySaver(),
)

result = agent.invoke(
    {
        "messages": [{"role": "user", "content": "Hello"}],
        "user_id": "user_123",
        "preferences": {"theme": "dark"}
    },
    {"configurable": {"thread_id": "1"}})

📖 "Retrieval addresses these problems by fetching relevant external knowledge at query time. This is the foundation of Retrieval-Augmented Generation (RAG): enhancing an LLM’s answers with context-specific information." — Retrieval

📖 "In 2-Step RAG, the retrieval step is always executed before the generation step." — Retrieval

flowchart LR
  subgraph TwoStep[2-Step RAG]
    A1[User Question] --> A2[Retrieve] --> A3[Generate] --> A4[Answer]
  end
  subgraph Agentic[Agentic RAG]
    B1[Question] --> B2[Agent LLM]
    B2 --> B3{Need info?}
    B3 -- Yes --> B4[Search tool]
    B4 --> B2
    B3 -- No --> B5[Final answer]
  end

📖 "Instead of retrieving documents before answering, an agent (powered by an LLM) reasons step-by-step and decides when and how to retrieve information during the interaction." — Retrieval

📖 "The only thing an agent needs to enable RAG behavior is access to one or more tools that can fetch external knowledge—such as documentation loaders, web APIs, or database queries." — Retrieval

import requests
from langchain.tools import tool
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent


@tool
def fetch_url(url: str) -> str:
    """Fetch text content from a URL"""
    response = requests.get(url, timeout=10.0)
    response.raise_for_status()
    return response.text

system_prompt = """\
Use fetch_url when you need to fetch information from a web-page; quote relevant snippets.
"""

agent = create_agent(
    model="claude-sonnet-4-6",
    tools=[fetch_url], # A tool for retrieval [!code highlight]
    system_prompt=system_prompt,
)

📖 "Context engineering is providing the right information and tools in the right format so the LLM can accomplish a task." — Context engineering in agents

📖 "When agents fail, it's usually because the LLM call inside the agent took the wrong action / didn't do what we expected." — Context engineering in agents

flowchart TD
  subgraph Controllable[可控上下文]
    MC[Model Context
transient] 
    TC[Tool Context
persistent]
    LC[Life-cycle Context
persistent]
  end
  subgraph Sources[数据源]
    RC[Runtime Context
会话级静态配置]
    ST[State
短期记忆]
    SR[Store
长期记忆]
  end
  MC --> RC & ST & SR
  TC --> RC & ST & SR
  LC --> ST & SR

📖 "The examples above use wrap_model_call to make transient updates - modifying what messages are sent to the model for a single call without changing what's saved in state." — Context engineering in agents

📖 "The summarized conversation history is permanently updated - future turns will see the summary instead of the original messages." — Context engineering in agents

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model="gpt-5.5",
    tools=[...],
    middleware=[
        SummarizationMiddleware(
            model="gpt-5.4-mini",
            trigger={"tokens": 4000},
            keep={"messages": 20},
        ),
    ],
)

📖 "Model Context Protocol (MCP) is an open protocol that standardizes how applications provide tools and context to LLMs." — Model Context Protocol (MCP)

📖 "MultiServerMCPClient is stateless by default. Each tool invocation creates a fresh MCP ClientSession, executes the tool, and then cleans up." — Model Context Protocol (MCP)

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent

async def main():
    client = MultiServerMCPClient(
        {
            "math": {
                "transport": "stdio",
                "command": "python",
                "args": ["/path/to/math_server.py"],
            },
            "weather": {
                "transport": "http",
                "url": "http://localhost:8000/mcp",
            }
        }
    )
    tools = await client.get_tools()
    agent = create_agent("claude-sonnet-4-6", tools)

📖 "MCP servers run as separate processes—they can't access LangGraph runtime information like the store, context, or agent state. Interceptors bridge this gap by giving you access to this runtime context during MCP tool execution." — Model Context Protocol (MCP)

📖 "Tool execution errors (CallToolResult(isError=True)) do not raise by default, so exception-catching interceptors never trigger on them. To catch those as exceptions here, set handle_tool_errors=False." — Model Context Protocol (MCP)

from dataclasses import dataclass
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.interceptors import MCPToolCallRequest
from langchain.agents import create_agent

@dataclass
class Context:
    user_id: str
    api_key: str

async def inject_user_context(request: MCPToolCallRequest, handler):
    """Inject user credentials into MCP tool calls."""
    runtime = request.runtime
    user_id = runtime.context.user_id
    modified_request = request.override(
        args={**request.args, "user_id": user_id}
    )
    return await handler(modified_request)

client = MultiServerMCPClient({...}, tool_interceptors=[inject_user_context])

flowchart TD
    U[用户需求] --> Q{单 agent 能否胜任?}
    Q -->|工具少/上下文可控| S[单 agent + 动态工具/prompt]
    Q -->|工具过多/需专精/需并行| M[多智能体]
    M --> C[Context management 按需暴露知识]
    M --> D[Distributed development 团队独立维护]
    M --> P[Parallelization 并发子任务]

📖 "Multi-agent systems coordinate specialized components to tackle complex workflows. However, not every complex task requires this approach—a single agent with the right (sometimes dynamic) tools and prompt can often achieve similar results." — Multi-agent

📖 "Here are the main patterns for building multi-agent systems, each suited to different use cases:" — Multi-agent

📖 "Handoffs, Skills, and Router are most efficient for single tasks (3 calls each). Subagents adds one extra call because results flow back through the main agent—this overhead provides centralized control." — Multi-agent

📖 "Stateful patterns (Handoffs, Skills) save 40-50% of calls on repeat requests. Subagents maintain consistent cost per request—this stateless design provides strong context isolation but at the cost of repeated model calls." — Multi-agent

📖 "For multi-domain tasks, patterns with parallel execution (Subagents, Router) are most efficient." — Multi-agent

Pattern     One-shot   Repeat request   Multi-domain
Subagents   4 calls    8 calls (4+4)    5 calls, 9K tokens
Handoffs    3 calls    5 calls (3+2)    7+ calls, 14K+ tokens
Skills      3 calls    5 calls (3+2)    3 calls, 15K tokens
Router      3 calls    6 calls (3+3)    5 calls, 9K tokens

📖 "A human decision then determines what happens next: the action can be approved as-is (approve), modified before running (edit), rejected with feedback (reject), or responded to directly (respond) for "ask user" style tools." — Human-in-the-loop

📖 "Use reject when the human is denying the requested action. Use respond only when the human is acting as the tool, such as answering an ask_user prompt. Do not use respond to deny side-effecting tools, because its message is treated as a successful tool result." — Human-in-the-loop

agent.invoke(
    Command(
        resume={"decisions": [{"type": "approve"}]}  # or "reject"
    ),
    config=config,  # Same thread ID to resume the paused conversation
    version="v2",
)

📖 "You must configure a checkpointer to persist the graph state across interrupts." — Human-in-the-loop

📖 "When the when predicate returns False, the call runs without interrupting. When it returns True, or when you omit when, the call pauses as usual. Calls that evaluate to False are never added to the interrupt batch, so a reviewer only sees the actions that need a decision." — Human-in-the-loop

def is_write_query(request: ToolCallRequest) -> bool:
    """Pause SQL that isn't a read-only SELECT."""
    query = request.tool_call["args"].get("query", "")
    return not query.lstrip().upper().startswith("SELECT")

📖 "Streaming is crucial for enhancing the responsiveness of applications built on LLMs. By displaying output progressively, even before a complete response is ready, streaming significantly improves user experience (UX), particularly when dealing with the latency of LLMs." — Streaming

📖 "To stream tokens as they are produced by the LLM, use stream_mode="messages"." — Streaming

from langgraph.config import get_stream_writer

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    writer = get_stream_writer()
    writer(f"Looking up data for city: {city}")
    writer(f"Acquired data for city: {city}")
    return f"It's always sunny in {city}!"

📖 "You can stream real-time updates while the agent runs and handles interrupts using stream_events(). Use stream.messages to stream LLM tokens and stream.values to check agent state snapshots for interrupts." — Human-in-the-loop

📖 "When there are multiple LLMs at any point in an agent, it's often necessary to disambiguate the source of messages as they are generated." — Streaming

if agent_name := metadata.get("lc_agent_name"):
    if agent_name != current_agent:
        print(f"🤖 {agent_name}: ")
        current_agent = agent_name

📖 "LangGraph is a low-level orchestration framework and runtime for building, managing, and deploying long-running, stateful agents." — LangGraph overview

📖 "LangGraph is focused on the underlying capabilities important for agent orchestration: durable execution, streaming, human-in-the-loop, and more." — LangGraph overview

📖 "LangGraph does not abstract prompts or architecture, and provides the following central benefits:" — LangGraph overview

from langgraph.graph import StateGraph, MessagesState, START, END

def mock_llm(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}

graph = StateGraph(MessagesState)
graph.add_node(mock_llm)
graph.add_edge(START, "mock_llm")
graph.add_edge("mock_llm", END)
graph = graph.compile()

graph.invoke({"messages": [{"role": "user", "content": "hi!"}]})

📖 "When you build an agent with LangGraph, you will first break it apart into discrete steps called nodes. Then, you will describe the different decisions and transitions from each of your nodes. Finally, you connect nodes together through a shared state that each node can read from and write to." — Thinking in LangGraph

📖 "The arrows in this diagram show possible paths, but the actual decision of which path to take happens inside each node." — Thinking in LangGraph

flowchart TD
    A[START] --> B[Read Email]
    B --> C[Classify Intent]
    C -.-> D[Doc Search]
    C -.-> E[Bug Track]
    C -.-> F[Human Review]
    D --> G[Draft Reply]
    E --> G
    F --> G
    G -.-> H[Human Review]
    G -.-> I[Send Reply]
    H --> J[END]
    I --> J[END]

📖 "State is the shared memory accessible to all nodes in your agent." — Thinking in LangGraph

📖 "A key principle: your state should store raw data, not formatted text. Format prompts inside nodes when you need them." — Thinking in LangGraph

from typing import TypedDict, Literal

class EmailClassification(TypedDict):
    intent: Literal["question", "bug", "billing", "feature", "complex"]
    urgency: Literal["low", "medium", "high", "critical"]
    topic: str
    summary: str

class EmailAgentState(TypedDict):
    email_content: str
    sender_email: str
    email_id: str
    classification: EmailClassification | None
    search_results: list[str] | None   # 原始文档块，非格式化文本
    customer_history: dict | None       # CRM 原始数据
    draft_response: str | None
    messages: list[str] | None

📖 "Transient failures get retries, LLM-recoverable errors loop back with context, user-fixable problems pause for input, and unexpected errors bubble up for debugging." — Thinking in LangGraph

from langgraph.types import RetryPolicy

workflow.add_node(
    "search_documentation",
    search_documentation,
    retry_policy=RetryPolicy(max_attempts=3, initial_interval=1.0)
)

from langgraph.types import Command

def execute_tool(state: State) -> Command[Literal["agent", "execute_tool"]]:
    try:
        result = run_tool(state['tool_call'])
        return Command(update={"tool_result": result}, goto="agent")
    except ToolError as e:
        return Command(update={"tool_result": f"Tool error: {str(e)}"}, goto="agent")

📖 "The answer involves trade-offs between resilience and observability." — Thinking in LangGraph

📖 "Smaller nodes mean more frequent checkpoints, which means less work to repeat if something goes wrong." — Thinking in LangGraph

📖 "LangGraph provides low-level supporting infrastructure for *any* long-running, stateful workflow or agent." — LangGraph overview

📖 "LangGraph is inspired by Pregel and Apache Beam." — LangGraph overview

📖 "LangGraph is built by LangChain Inc, the creators of LangChain, but can be used without LangChain." — LangGraph overview

config = {"configurable": {"thread_id": "customer_123"}}
stream = app.stream_events(initial_state, config, version="v3")
# 图会在 human_review 处暂停
print(f"human review interrupt:{stream.interrupts}")

📖 "LangGraph provides two different APIs to build agent workflows: the Graph API and the Functional API. Both APIs share the same underlying runtime and can be used together in the same application, but they are designed for different use cases and development preferences." — Choosing between Graph and Functional APIs

📖 "Both APIs provide the same core LangGraph features (persistence, streaming, human-in-the-loop, memory) but package them in different paradigms to suit different development styles and use cases." — Choosing between Graph and Functional APIs

from langgraph.func import entrypoint, task

@task
def process_user_input(user_input: str) -> dict:
    # Existing function with minimal changes
    return {"processed": user_input.lower().strip()}

@entrypoint(checkpointer=checkpointer)
def workflow(user_input: str) -> str:
    # Standard Python control flow
    processed = process_user_input(user_input).result()

    if "urgent" in processed["processed"]:
        response = handle_urgent_request(processed).result()
    else:
        response = handle_normal_request(processed).result()

    return response

📖 "Because LangGraph runs the latest deployed graph against state that has been persisted for existing threads, every change you ship is effectively a backward-compatible API change with respect to your existing checkpoints." — Backward compatibility

📖 "Unlike workflow engines that pin a run to the version of code it started with, LangGraph applies the latest graph immediately to *every* thread, both new threads and threads that resume from a checkpoint." — Backward compatibility

flowchart TD
    A[上线一次图代码改动] --> B{新代码能否对旧 State
加载并执行?}
    B -- 否 --> T[Technical compatibility
如改节点名/State key]
    B -- 是 --> C{改动是否改变了
旧 thread 的业务含义?}
    C -- 是 --> D[Business compatibility
用 flow_version 分支]
    C -- 否 --> E{是否用 Functional API
或 node 内 task/interrupt?}
    E -- 是 --> F[Non-determinism
注意 task 顺序/副作用]
    E -- 否 --> G[安全]

📖 "A LangGraph application consists of one or more graphs, a configuration file (langgraph.json), a file that specifies dependencies, and an optional .env file that specifies environment variables." — Application structure

📖 "Each graph is identified by a name (which should be unique) and a path for either: (1) the compiled graph or (2) a function that makes a graph is defined." — Application structure

{
  "dependencies": ["langchain_openai", "./your_package"],
  "graphs": {
    "my_agent": "./your_package/your_file.py:agent"
  },
  "env": "./.env"
}

📖 "The recommended pattern is to record the relevant *behavioral version* on the state at thread start, then branch on it with a conditional edge:" — Backward compatibility

📖 "This pattern only works if you set the version *at thread start*, before any branch that needs to be versioned. Setting it later means existing threads will not have it set when they need it." — Backward compatibility

def intake(state: State) -> dict:
    # Stamp new threads with the current flow version. Existing threads
    # that resume past `intake` keep whatever value was already saved.
    return {"flow_version": state.get("flow_version", 2)}


def after_triage(state: State) -> str:
    if state.get("flow_version", 1) >= 2:
        return "policy_check"
    return "respond"

📖 "A Functional API entrypoint compiles to a single node that replays the entrypoint body from the beginning when a run resumes, using cached @task results to skip work that has already been done." — Backward compatibility

📖 "Introducing non-deterministic operations outside of a @task, such as time.time(), random.random(), or a network call inlined in the entrypoint body. On replay these produce different values than they did on the first run, which can change the control flow." — Backward compatibility

1. drain：让进行中的 run 先跑完再部署改动
2. 新 @task：把新逻辑包进新的 @task，使其结果被独立 checkpoint
3. 新 graph 名：在 langgraph.json 里用新名字注册新 entrypoint，只把新 thread 路由过去

📖 "At its core, LangGraph models agent workflows as graphs." — Graph API overview

📖 "In short: *nodes do the work, edges tell what to do next*." — Graph API overview

📖 "A super-step can be considered a single iteration over the graph nodes. Nodes that run in parallel are part of the same super-step, while nodes that run sequentially belong to separate super-steps." — Graph API overview

from typing import TypedDict
from langgraph.graph import START, StateGraph

class State(TypedDict):
    value_1: str
    value_2: int

def step_1(state: State):
    return {"value_1": "a"}

builder = StateGraph(State)
builder.add_node(step_1)
builder.add_edge(START, "step_1")
graph = builder.compile()

flowchart LR
    START([START]) --> A[Node: 干活]
    A -->|Edge: 决定下一步| B[Node]
    B --> END([END])

📖 "The State consists of the [schema of the graph](#schema) as well as [reducer functions](#reducers) which specify how to apply updates to the state." — Graph API overview

📖 "If no reducer function is explicitly specified then it is assumed that all updates to that key should override it." — Graph API overview

📖 "A custom reducer combines the left and right arguments instead of replacing the state value, which is useful for accumulating values, such as appending updates to a list." — Graph API overview

from operator import add
from typing import Annotated
from typing_extensions import TypedDict

class State(TypedDict):
    foo: int
    bar: Annotated[list[str], add]

📖 "If you don't specify a reducer, every state update will overwrite the list of messages with the most recently provided value." — Graph API overview

📖 "For brand new messages, it will simply append to existing list, but it will also handle the updates for existing messages correctly." — Graph API overview

from langchain.messages import AnyMessage
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDict

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    extra_field: int

def node(state: State):
    new_message = AIMessage("Hello!")
    return {"messages": [new_message], "extra_field": 10}

📖 "In LangGraph, nodes are Python functions (either synchronous or asynchronous) that accept the following arguments:" — Graph API overview

📖 "If execution stops and later resumes (for example after an interrupt or a retry), the affected node runs again from the start of its function. Code and side effects before the pause run again." — Graph API overview

📖 "Nodes should return updates to the state directly, instead of mutating the state." — Use the graph API

from langgraph.runtime import Runtime

def node_with_runtime(state: State, runtime: Runtime[Context]):
    print("In node: ", runtime.context.user_id)
    return {"results": f"Hello, {state['input']}!"}

📖 "A node can have multiple outgoing edges. If a node has multiple outgoing edges, all of those destination nodes will be executed in parallel as a part of the next superstep." — Graph API overview

📖 "Normal Edges: Go directly from one node to the next." — Graph API overview

📖 "Similar to nodes, the routing_function accepts the current state of the graph and returns a value." — Graph API overview

def conditional_edge(state: State) -> Literal["b", "c"]:
    return state["which"]

builder.add_conditional_edges("a", conditional_edge)

📖 "Use Command when you need to both update state and route to a different node. If you only need to route without updating state, use [conditional edges](#conditional-edges) instead." — Graph API overview

📖 "Command only adds dynamic edges—static edges defined with add_edge / addEdge still execute." — Graph API overview

def node_a(state: State) -> Command[Literal["node_b", "node_c"]]:
    value = random.choice(["b", "c"])
    goto = "node_b" if value == "b" else "node_c"
    return Command(
        update={"foo": value},
        goto=goto,
    )

📖 "To support this design pattern, LangGraph supports returning Send objects from conditional edges. Send takes two arguments: first is the name of the node, and second is the state to pass to that node." — Graph API overview

from langgraph.types import Send

def continue_to_jokes(state: OverallState):
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]

builder.add_conditional_edges("generate_topics", continue_to_jokes, ["generate_joke"])

flowchart LR
  A["@entrypoint\n工作流入口"] --> B["@task\n离散工作单元"]
  B --> C["future 对象\n.result() / await"]
  C --> D["checkpoint\n保存 task 结果"]

📖 "The Functional API does not require thinking about graph structure. You can use standard Python constructs to define workflows. This will usually trim the amount of code you need to write." — Functional API overview

📖 "In the Functional API, when tasks are executed, their results are saved to an existing checkpoint associated with the given entrypoint instead of creating a new checkpoint." — Functional API overview

📖 "Both APIs share the same underlying runtime, so you can use them together in the same application." — Functional API overview

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint, task
from langgraph.types import interrupt

@task
def write_essay(topic: str) -> str:
    """Write an essay about the given topic."""
    time.sleep(1) # A placeholder for a long-running task.
    return f"An essay about topic: {topic}"

@entrypoint(checkpointer=InMemorySaver())
def workflow(topic: str) -> dict:
    """A simple workflow that writes an essay and asks for a review."""
    essay = write_essay("cat").result()
    is_approved = interrupt({
        "essay": essay,
        "action": "Please approve/reject the essay",
    })
    return {"essay": essay, "is_approved": is_approved}

📖 "@entrypoint: Marks a function as the starting point of a workflow, encapsulating logic and managing execution flow, including handling long-running tasks and interrupts." — Functional API overview

📖 "When you call a task, it returns *immediately* with a future object. A future is a placeholder for a result that will be available later." — Functional API overview

📖 "Tasks *cannot* be called directly from the main application code." — Functional API overview

@task
def is_even(number: int) -> bool:
    return number % 2 == 0

@task
def format_message(is_even: bool) -> str:
    return "The number is even." if is_even else "The number is odd."

@entrypoint(checkpointer=checkpointer)
def workflow(inputs: dict) -> str:
    """Simple workflow to classify a number."""
    even = is_even(inputs["number"]).result()
    return format_message(even).result()

📖 "When you resume a workflow run, the code does NOT resume from the same line of code where execution stopped. Execution returns to a checkpoint boundary, and the workflow replays forward until it reaches the pause again." — Functional API overview

📖 "To use features like human-in-the-loop, you must place non-deterministic work (for example, random values) and side effects (for example, file writes or API calls) in tasks." — Functional API overview

📖 "This matching is strictly index-based, so the order of the resume values should match the order of the interrupts." — Functional API overview

# 正确：把非确定性的取时间包成 task，resume 时返回同一结果
import time
from langgraph.func import task

@task
def get_time() -> float:
    return time.time()

@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
    t0 = inputs["t0"]
    t1 = get_time().result()
    delta_t = t1 - t0
    if delta_t > 1:
        result = slow_task(1).result()
        value = interrupt("question")
    else:
        result = slow_task(2).result()
        value = interrupt("question")
    return {"result": result, "value": value}

📖 "By default, the previous parameter is the return value of the previous invocation." — Functional API overview

📖 "The first value is the return value of the entrypoint, and the second value is the value that will be saved in the checkpoint. The type annotation is entrypoint.final[return_type, save_type]." — Functional API overview

@entrypoint(checkpointer=checkpointer)
def accumulate(n: int, *, previous: int | None) -> entrypoint.final[int, int]:
    previous = previous or 0
    total = previous + n
    # 返回旧的 previous 给调用方，但把新 total 存进 checkpoint
    return entrypoint.final(value=previous, save=total)

config = {"configurable": {"thread_id": "my-thread"}}
print(accumulate.invoke(1, config=config))  # 0
print(accumulate.invoke(2, config=config))  # 1
print(accumulate.invoke(3, config=config))  # 3

📖 "Tasks can be executed in parallel by invoking them concurrently and waiting for the results. This is useful for improving performance in IO bound tasks (e.g., calling APIs for LLMs)." — Use the functional API

📖 "Timeouts are supported only for async tasks and entrypoints. If you set timeout on a sync function, LangGraph raises an error when the task or entrypoint is declared." — Use the functional API

📖 "When we resume execution, we won't need to re-run the slow_task as its result is already saved in the checkpoint." — Use the functional API

import asyncio
from langgraph.errors import NodeTimeoutError
from langgraph.func import entrypoint, task
from langgraph.types import RetryPolicy

@task(
    timeout=1.0,
    retry_policy=RetryPolicy(retry_on=NodeTimeoutError),
)
async def call_api(url: str) -> str:
    await asyncio.sleep(2)
    return f"result from {url}"

@entrypoint(timeout=5.0)
async def workflow(inputs: dict) -> str:
    return await call_api(inputs["url"])

📖 "LangGraph's persistence layer gives agents short-term memory through checkpointers and long-term memory through stores." — Persistence

📖 "Most applications can use both: a checkpointer tracks the current thread, and a store tracks durable information across threads." — Persistence

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

checkpointer = InMemorySaver()
store = InMemoryStore()

graph = builder.compile(checkpointer=checkpointer, store=store)

result = graph.invoke(
    {"messages": [{"role": "user", "content": "Hi, my name is Bob."}]},
    {"configurable": {"thread_id": "thread-1"}},
)

📖 "A checkpointer saves a snapshot of graph state at each super-step, organized into threads." — Checkpointers

📖 "Understanding super-step boundaries is important for time travel, because you can only resume execution from a checkpoint (i.e., a super-step boundary)." — Checkpointers

📖 "The checkpointer uses thread_id as the primary key for storing and retrieving checkpoints." — Checkpointers

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": "", "bar":[]}, config)
# 之后共 4 个 checkpoint,最后一个: {'foo': 'b', 'bar': ['a', 'b']},无下一节点

📖 "LangGraph supports three durability modes that let you balance performance and data consistency." — Checkpointers

📖 "LangGraph persists changes synchronously before the next step starts. This ensures that LangGraph writes every checkpoint before continuing execution, providing high durability at the cost of some performance overhead." — Checkpointers

graph.stream(
    {"input": "test"},
    durability="sync"
)

📖 "When a graph node fails mid-execution at a given [super-step](#super-steps), LangGraph stores pending checkpoint writes from any other nodes that completed successfully at that super-step. When you resume graph execution from that super-step you don't re-run the successful nodes." — Checkpointers

📖 "These per-task writes are what enable [pending writes](#pending-writes) recovery: if another node in the same super-step fails, the successful nodes' writes are already durable and don't need to be re-run on resume." — Checkpointers

super-step k: [node_x (成功), node_y (失败)]
  -> node_x 的输出作为 task 条目写入 checkpoint_writes 表(pending write)
  -> 从 super-step k 恢复时: node_x 不重跑,只重跑 node_y

📖 "You can edit the graph state using update_state. This creates a new checkpoint with the updated values — it does not modify the original checkpoint." — Checkpointers

📖 "Nodes before the checkpoint are skipped (their results are already saved). Nodes after the checkpoint re-execute, including any LLM calls, API requests, or interrupts — which are always re-triggered during replay." — Checkpointers

# 取指定 checkpoint_id 的快照
config = {"configurable": {"thread_id": "1", "checkpoint_id": "1ef663ba-28fe-6528-8002-5a559208592c"}}
graph.get_state(config)

# 在历史中找由 update_state 创建的 fork
history = list(graph.get_state_history(config))
forks = [s for s in history if s.metadata["source"] == "update"]

📖 "Under the hood, checkpointing is powered by checkpointer objects that conform to BaseCheckpointSaver interface." — Checkpointers

📖 "The default serializer, JsonPlusSerializer, uses ormsgpack and JSON under the hood, which is not suitable for all types of objects." — Checkpointers

import sqlite3

from langgraph.checkpoint.serde.encrypted import EncryptedSerializer
from langgraph.checkpoint.sqlite import SqliteSaver

serde = EncryptedSerializer.from_pycryptodome_aes()  # reads LANGGRAPH_AES_KEY
checkpointer = SqliteSaver(sqlite3.connect("checkpoint.db"), serde=serde)

📖 "Stores let agents persist information across threads, including user preferences, accumulated knowledge, and facts that should survive beyond a single conversation. Unlike checkpointers, which save the full graph state scoped to one thread, stores hold arbitrary key-value data accessible from any thread." — Stores

📖 "The store works hand-in-hand with the checkpointer: the checkpointer saves state to threads, as discussed above, and the store allows you to store arbitrary information for access *across* threads." — Stores

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

checkpointer = InMemorySaver()
store = InMemoryStore()
graph = builder.compile(checkpointer=checkpointer, store=store)

📖 "Memories are namespaced by a tuple, which is (<user_id>, \"memories\") in the following example. The namespace can be any length and represent anything, does not have to be user specific." — Stores

📖 "Beyond simple retrieval, the store also supports semantic search, allowing you to find memories based on meaning rather than exact matches. To enable this, configure the store with an embedding model:" — Stores

from langchain.embeddings import init_embeddings
from langgraph.store.memory import InMemoryStore

embeddings = init_embeddings("openai:text-embedding-3-small")
store = InMemoryStore(index={"embed": embeddings, "dims": 1536})

store.put(("user_123", "memories"), "1", {"text": "I love pizza"})
store.put(("user_123", "memories"), "2", {"text": "I am a plumber"})

items = store.search(("user_123", "memories"), query="I'm hungry", limit=1)

📖 "Short-term memory (thread-level persistence) enables agents to track multi-turn conversations." — Memory

📖 "Use long-term memory to store user-specific or application-specific data across conversations." — Memory

async def call_model(state: MessagesState, runtime: Runtime[Context]):
    user_id = runtime.context.user_id
    namespace = (user_id, "memories")
    memories = await runtime.store.asearch(
        namespace, query=state["messages"][-1].content, limit=3
    )
    info = "\n".join([d.value["data"] for d in memories])
    await runtime.store.aput(
        namespace, str(uuid.uuid4()), {"data": "User prefers dark mode"}
    )

📖 "With [short-term memory](#add-short-term-memory) enabled, long conversations can exceed the LLM's context window." — Memory

📖 "One way to decide when to truncate messages is to count the tokens in the message history and truncate whenever it approaches that limit." — Memory

from langchain_core.messages.utils import trim_messages, count_tokens_approximately

def call_model(state: MessagesState):
    messages = trim_messages(
        state["messages"],
        strategy="last",
        token_counter=count_tokens_approximately,
        max_tokens=128,
        start_on="human",
        end_on=("human", "tool"),
    )
    response = model.invoke(messages)
    return {"messages": [response]}

📖 "Workflows have predetermined code paths and are designed to operate in a certain order." — Workflows and agents

📖 "Agents have more autonomy than workflows, and can make decisions about the tools they use and how to solve problems. You can still define the available toolset and guidelines for how agents behave." — Workflows and agents

flowchart LR
    START --> llm_call
    llm_call -->|有 tool_calls| tool_node
    llm_call -->|无 tool_calls| END
    tool_node --> llm_call

📖 "The Send API lets you dynamically create worker nodes and send them specific inputs. Each worker has its own state, and all worker outputs are written to a shared state key that is accessible to the orchestrator graph." — Workflows and agents

📖 "Orchestrator-worker workflows provide more flexibility and are often used when subtasks cannot be predefined the way they can with [parallelization](#parallelization)." — Workflows and agents

from langgraph.types import Send

def assign_workers(state: State):
    """Assign a worker to each section in the plan"""
    return [Send("llm_call", {"section": s}) for s in state["sections"]]

orchestrator_worker_builder.add_conditional_edges(
    "orchestrator", assign_workers, ["llm_call"]
)

📖 "It exposes graph execution through stream modes such as updates, values, messages, custom, checkpoints, tasks, and debug." — Streaming

📖 "The debug mode combines checkpoints and tasks events with additional metadata. Use checkpoints or tasks directly if you only need a subset of the debug information." — Streaming

for chunk in graph.stream(inputs, stream_mode=["updates", "custom"], version="v2"):
    if chunk["type"] == "updates":
        for node_name, state in chunk["data"].items():
            print(f"Node `{node_name}` updated: {state}")
    elif chunk["type"] == "custom":
        print(f"Custom event: {chunk['data']}")

📖 "Event streaming is the recommended in-process streaming model for most LangGraph application code. It returns a run stream object that can be consumed in multiple ways at the same time." — Event streaming

📖 "Multiple consumers can read these projections concurrently. Reading stream.messages does not consume events needed by stream.values, stream.subgraphs, or stream.output." — Event streaming

📖 "Use streaming when you need low-level access to those modes; use event streaming when application code benefits from typed projections." — Event streaming

flowchart TD
    A[Pregel engine 运行图步骤] -->|emits| B[Raw Pregel events]
    B -->|sent to| C[Event router]
    C -->|cascades through| D[Stream transformers]
    D -->|produces| E[Event Stream 投影]

stream = graph.stream_events({
    "messages": [{"role": "user", "content": "What is 42 * 17?"}],
}, version="v3")

for message in stream.messages:
    for token in message.text:
        print(token, end="", flush=True)

final_state = stream.output

📖 "Use the messages streaming mode to stream Large Language Model (LLM) outputs token by token from any part of your graph, including nodes, tools, subgraphs, or tasks." — Streaming

📖 "Use the nostream tag to exclude LLM output from the stream entirely. Invocations tagged with nostream still run and produce output; their tokens are simply not emitted in messages mode." — Streaming

async for chunk in graph.astream(
    {"topic": "cats"},
    stream_mode="messages",
    version="v2",
):
    if chunk["type"] == "messages":
        msg, metadata = chunk["data"]
        if metadata["tags"] == ["joke"]:
            print(msg.content, end="|", flush=True)

📖 "Every chunk is a StreamPart dict with a consistent shape — regardless of stream mode, number of modes, or subgraph settings:" — Streaming

📖 "With v1 (default), the output format changes based on your streaming options (single mode returns raw data, multiple modes return (mode, data) tuples, subgraphs return (namespace, data) tuples). With v2, the format is always the same:" — Streaming

# v2:统一 StreamPart
for chunk in graph.stream(inputs, stream_mode="updates", version="v2"):
    print(chunk["type"])  # "updates"
    print(chunk["ns"])    # ()
    print(chunk["data"])  # {"node_name": {"key": "value"}}

# v1:原始 dict
for chunk in graph.stream(inputs, stream_mode="updates"):
    print(chunk)  # {"node_name": {"key": "value"}}

📖 "You can use stream_mode="custom" to stream data from any LLM API—even if that API does not implement the LangChain chat model interface." — Streaming

📖 "In async code running on Python \< 3.11, get_stream_writer will not work." — Streaming

from langgraph.config import get_stream_writer

def call_arbitrary_model(state):
    writer = get_stream_writer()
    for chunk in your_custom_streaming_client(state["topic"]):
        writer({"custom_llm_chunk": chunk})
    return {"result": "completed"}

📖 "Stream transformers are the projection layer in event streaming. They observe protocol events, keep their own state, and expose derived views of a run" — Event streaming

📖 "Modes that no transformer requests are never emitted — declaring ("custom",) is what causes custom events to flow through the run at all." — Event streaming

flowchart TD
    A[Pregel modes] --> B[Events]
    B --> C[Built-in projections]
    C --> D[User transformers]
    D --> E[Run projections]

class CustomTransformer(StreamTransformer):
    required_stream_modes = ("custom",)

    def process(self, event: ProtocolEvent) -> bool:
        if event["method"] == "custom":
            ...
        return True

📖 "Interrupts allow you to pause graph execution at specific points and wait for external input before continuing. This enables human-in-the-loop patterns where you need external input to proceed." — Interrupts

📖 "Unlike static breakpoints (which pause before or after specific nodes), interrupts are dynamic: they can be placed anywhere in your code and can be conditional based on your application logic." — Interrupts

📖 "The thread_id you choose is effectively your persistent cursor. Reusing it resumes the same checkpoint; using a new value starts a brand-new thread with an empty state." — Interrupts

from langgraph.types import Command, interrupt

def approval_node(state):
    approved = interrupt("Do you approve this action?")
    return {"approved": approved}

config = {"configurable": {"thread_id": "thread-1"}}
stream = graph.stream_events({"input": "data"}, config=config, version="v3")
if stream.interrupted:
    print(stream.interrupts)   # (Interrupt(value='Do you approve this action?'),)
resumed = graph.stream_events(Command(resume=True), config=config, version="v3")

📖 "Matching is strictly index-based, so the order of interrupt calls within the node is important." — Interrupts

📖 "The result is exponential re-execution of any code inside the loop body." — Interrupts

📖 "Depending on which checkpointer is used, complex values may not be serializable (e.g. you can't serialize a function)." — Interrupts

def get_age_node(state):
    question = state.get("pending_question") or "What is your age?"
    answer = interrupt(question)  # 每次调用恰好一次
    if isinstance(answer, int) and answer > 0:
        return {"age": answer, "pending_question": None}
    return {"pending_question": f"'{answer}' is not a valid age. Please enter a positive number."}

def route(state):
    return END if state.get("age") is not None else "collect_age"

📖 "When adding subgraphs, you need to define how the parent graph and the subgraph communicate:" — Subgraphs

📖 "When the parent graph and subgraph have different state schemas (no shared keys), invoke the subgraph inside a node function." — Subgraphs

📖 "When the parent graph and subgraph share state keys, you can pass a compiled subgraph directly to add_node. No wrapper function is needed—the subgraph reads from and writes to the parent's state channels automatically." — Subgraphs

# 方式一:schema 不同 —— 节点内调用并做映射
def call_subgraph(state):
    subgraph_output = subgraph.invoke({"bar": state["foo"]})
    return {"foo": subgraph_output["bar"]}
builder.add_node("node_1", call_subgraph)

# 方式二:共享 key —— 直接作为节点
builder.add_node("node_1", subgraph)

flowchart TD
    A[父图状态] -->|schema 不同| B[wrapper 节点
映射→invoke→映射回]
    A -->|共享 key| C[add_node子图
直接读写父 channel]

📖 "Omit checkpointer or set it to None. Each call starts fresh, but within a single call the subgraph inherits the parent's checkpointer and can use interrupt() to pause and resume." — Subgraphs

📖 "Per-thread subgraphs do not support parallel tool calls. When an LLM has access to a per-thread subagent as a tool, it may try to call that tool multiple times in parallel (for example, asking the fruit expert about apples and bananas simultaneously). This causes checkpoint conflicts because both calls write to the same namespace." — Subgraphs

📖 "Use this when you want to run a subagent like a plain function call with no checkpointing overhead. The subgraph cannot pause/resume and does not benefit from" — Subgraphs

subgraph = builder.compile(checkpointer=None)   # per-invocation(默认)
subgraph = builder.compile(checkpointer=True)   # per-thread,累积记忆
subgraph = builder.compile(checkpointer=False)  # stateless,无中断/无恢复

📖 "Both work by resuming from a prior checkpoint. Nodes before the checkpoint are not re-executed (results are already saved)." — Use time-travel

📖 "Replay re-executes nodes—it doesn't just read from cache." — Use time-travel

📖 "update_state does not roll back a thread. It creates a new checkpoint that branches from the specified point. The original execution history remains intact." — Use time-travel

history = list(graph.get_state_history(config))
before_joke = next(s for s in history if s.next == ("write_joke",))

fork_config = graph.update_state(before_joke.config, values={"topic": "chickens"})
fork_result = graph.invoke(None, fork_config)   # write_joke 用新 topic 重跑

📖 "The node containing the interrupt re-executes, and interrupt() pauses for a new Command(resume=...)." — Use time-travel

📖 "By default, a subgraph inherits the parent's checkpointer. The parent treats the entire subgraph as a single super-step — there is only one parent-level checkpoint for the whole subgraph execution. Time traveling from before the subgraph re-executes it from scratch." — Use time-travel

📖 "Set checkpointer=True on the subgraph to give it its own checkpoint history. This creates checkpoints at each step within the subgraph, allowing you to time travel from a specific point inside it — for example, between two interrupts." — Use time-travel

graph.invoke({"value": []}, config)            # 停在 step_a interrupt
graph.invoke(Command(resume="Alice"), config)  # 停在 step_b interrupt

parent_state = graph.get_state(config, subgraphs=True)
sub_config = parent_state.tasks[0].state.config
fork_config = graph.update_state(sub_config, {"value": ["forked"]})
result = graph.invoke(None, fork_config)   # step_b 重跑,step_a 结果保留

flowchart LR
    plan[Plan: 选 actor] --> exec[Execution: 并行执行]
    exec --> update[Update: 提交 channel 写入]
    update -->|有 actor 被选中| plan
    update -->|无 actor / 达到最大步数| done([结束])

📖 "Pregel organizes the execution of the application into multiple steps, following the Pregel Algorithm/Bulk Synchronous Parallel model." — LangGraph runtime

📖 "Execute all selected actors in parallel, until all complete, or one fails, or a timeout is reached. During this phase, channel updates are invisible to actors until the next step." — LangGraph runtime

from langgraph.channels import LastValue, EphemeralValue
from langgraph.pregel import Pregel, NodeBuilder

node1 = NodeBuilder().subscribe_only("a").do(lambda x: x + x).write_to("b")
node2 = NodeBuilder().subscribe_only("b").do(lambda x: x + x).write_to("c")

app = Pregel(
    nodes={"node1": node1, "node2": node2},
    channels={"a": EphemeralValue(str), "b": LastValue(str), "c": EphemeralValue(str)},
    input_channels=["a"],
    output_channels=["b", "c"],
)
app.invoke({"a": "foo"})  # {'b': 'foofoo', 'c': 'foofoofoofoo'}

📖 "LastValue is the default channel type. It stores the last value written to it, overwriting any previous value." — LangGraph runtime

📖 "Without delta storage, the full list is re-serialized into every checkpoint; with DeltaChannel, only the new messages written at each step are stored." — LangGraph runtime

📖 "The reducer runs on reconstruction, not on write." — LangGraph runtime

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.channels import DeltaChannel

class State(TypedDict):
    messages: Annotated[
        list[str],
        DeltaChannel(my_reducer, snapshot_frequency=5),
    ]

📖 "By default, retry_on uses default_retry_on, which retries on any exception except the following (and their subclasses):" — Fault tolerance

📖 "For exceptions from popular HTTP libraries such as requests and httpx, it only retries on 5xx status codes." — Fault tolerance

def my_node(state: State, runtime: Runtime) -> State:
    if runtime.execution_info.node_attempt > 1:
        return {"result": call_fallback_api()}
    return {"result": call_primary_api()}

builder.add_node("my_node", my_node, retry_policy=RetryPolicy(max_attempts=3))

📖 "Node timeouts only apply to async nodes. Sync nodes with a timeout are rejected at compile time. To wrap blocking I/O, use asyncio.to_thread inside an async node." — Fault tolerance

📖 "run_timeout is a hard wall-clock cap on a single attempt. It is never refreshed, regardless of node activity:" — Fault tolerance

async def long_running_node(state: State, runtime: Runtime) -> State:
    for batch in fetch_batches():
        process(batch)
        runtime.heartbeat()
    return {"result": "done"}

builder.add_node(
    "long_running_node",
    long_running_node,
    timeout=TimeoutPolicy(idle_timeout=30, refresh_on="heartbeat"),
)

📖 "Only after retries are exhausted does the error handler run." — Fault tolerance

📖 "interrupt() raised inside a node is not routed to the error handler." — Fault tolerance

def payment_error_handler(state: State, error: NodeError) -> Command:
    return Command(
        update={"status": f"compensated_after_{error.node}: {error.error}"},
        goto="finalize",
    )

graph = (
    StateGraph(State)
    .add_node(
        "charge_payment",
        charge_payment,
        retry_policy=RetryPolicy(max_attempts=3, retry_on=ConnectionError),
        error_handler=payment_error_handler,
    )
    .add_node("finalize", finalize)
    .compile()
)

📖 "agents are useful when you want an LLM to make a decision about whether to retrieve context from a vectorstore or respond to the user directly." — Build a custom RAG agent with LangGraph

📖 "The retriever tool can return potentially irrelevant documents, which indicates a need to improve the original user question." — Build a custom RAG agent with LangGraph

flowchart TD
    START --> Q[generate_query_or_respond]
    Q -->|有 tool_calls| R[retrieve]
    Q -->|无 tool_calls| END1[END]
    R --> G{grade_documents}
    G -->|relevant| A[generate_answer]
    G -->|not relevant| W[rewrite_question]
    W --> Q
    A --> END2[END]

📖 "We can enforce a higher degree of control in LangGraph by customizing the agent." — Build a custom SQL agent

📖 "Putting these steps in dedicated nodes lets us (1) force tool-calls when needed, and (2) customize the prompts associated with each step." — Build a custom SQL agent

📖 "lets us get started quickly, but we relied on the system prompt to constrain its behavior" — Build a custom SQL agent

flowchart TD
    START --> LT[list_tables]
    LT --> CGS[call_get_schema]
    CGS --> GS[get_schema]
    GS --> GQ[generate_query]
    GQ -->|有 tool_calls| CQ[check_query]
    GQ -->|无 tool_calls| END[END]
    CQ --> RQ[run_query]
    RQ --> GQ

📖 "Make sure that your database connection permissions are always scoped as narrowly as possible for your agent's needs." — Build a custom SQL agent

📖 "This will mitigate, though not eliminate, the risks of building a model-driven system." — Build a custom SQL agent

📖 "Use narrowly scoped database permissions and add application-specific validation before executing model-generated SQL." — Build a custom SQL agent

generate_query_system_prompt = """
You are an agent designed to interact with a SQL database.
...
DO NOT make any DML statements (INSERT, UPDATE, DELETE, DROP etc.) to the database.
"""

📖 "It can be prudent to check the agent's SQL queries before they are executed for any unintended actions or inefficiencies." — Build a custom SQL agent

📖 "Below, we allow for input to approve the tool call, edit its arguments, or provide user feedback." — Build a custom SQL agent

📖 "this is required to pause and resume the run." — Build a custom SQL agent

stream = agent.stream_events(
    Command(resume={"type": "accept"}),
    # Command(resume={"type": "edit", "args": {"query": "..."}}),
    config,
    version="v3",
)

📖 "This list of companies using LangGraph and their success stories is compiled from public sources." — Case studies

LinkedIn | Social Media | Code generation; Search & discovery
→ Blog: practical-text-to-sql-for-data-analytics

📖 "LangSmith Studio is a free visual interface for developing and testing your LangChain agents from your local machine." — LangSmith Studio

📖 "The development server supports hot-reloading—make changes to prompts or tool signatures in your code, and Studio reflects them immediately." — LangSmith Studio

langgraph dev

📖 "The langgraph dev command starts Agent Server in an in-memory mode. This mode is suitable for development and testing purposes. For production use, deploy Agent Server with access to a persistent storage backend." — Run a local server

from langgraph_sdk import get_sync_client

client = get_sync_client(url="http://localhost:2024")

for chunk in client.runs.stream(
    None,  # Threadless run
    "agent", # Name of assistant. Defined in langgraph.json.
    input={
        "messages": [{
            "role": "human",
            "content": "What is LangGraph?",
        }],
    },
    stream_mode="messages-tuple",
):
    print(f"Receiving new event of type: {chunk.event}...")
    print(chunk.data)
    print("\n\n")

📖 "LangSmith Cloud provides fully managed infrastructure for stateful, long-running agents with persistent state and background execution." — Deployment

📖 "Select your application's repository. Click Submit to deploy. This may take about 15 minutes to complete. You can check the status in the Deployment details view." — Deployment

from langgraph_sdk import get_sync_client # or get_client for async

client = get_sync_client(url="your-deployment-url", api_key="your-langsmith-api-key")

for chunk in client.runs.stream(
    None,    # Threadless run
    "agent", # Name of agent. Defined in langgraph.json.
    input={
        "messages": [{
            "role": "human",
            "content": "What is LangGraph?",
        }],
    },
    stream_mode="updates",
):
    print(f"Receiving new event of type: {chunk.event}...")
    print(chunk.data)
    print("\n\n")

📖 "Traces are a series of steps that your application takes to go from input to output. Each of these individual steps is represented by a run." — LangSmith Observability

📖 "By default, the trace will be logged to the project with the name default. To configure a custom project name, see [Log to a project](#log-to-a-project)." — LangSmith Observability

import langsmith as ls

with ls.tracing_context(project_name="email-agent-test", enabled=True):
    response = agent.invoke({
        "messages": [{"role": "user", "content": "Send a welcome email"}]
    })

📖 "You may want to mask sensitive data to prevent it from being logged to LangSmith." — LangSmith Observability

📖 "This example will redact anything matching the Social Security Number format XXX-XX-XXXX from traces sent to LangSmith." — LangSmith Observability

from langchain_core.tracers.langchain import LangChainTracer
from langgraph.graph import StateGraph, MessagesState
from langsmith import Client
from langsmith.anonymizer import create_anonymizer

anonymizer = create_anonymizer([
    # Matches SSNs
    { "pattern": r"\b\d{3}-?\d{2}-?\d{4}\b", "replace": "<ssn>" }
])

tracer_client = Client(anonymizer=anonymizer)
tracer = LangChainTracer(client=tracer_client)
# Define the graph
graph = (
    StateGraph(MessagesState)
    ...
    .compile()
    .with_config({'callbacks': [tracer]})
)

graph TB
    Project -->|contains| Trace1[Trace]
    Project -->|contains| Trace2[Trace]
    Trace1 -->|contains| Run1[Run]
    Trace1 -->|contains| Run2[Run]
    Trace1 -->|part of| Thread
    Trace2 -->|part of| Thread

📖 "LangSmith groups multiple [*traces*](#traces) within a [*project*](#projects). Each trace records the sequence of steps your application takes for a single operation, which are made up of individual [*runs*](#runs)." — Observability concepts

📖 "A *project* is a container for all the traces related to a single application or service." — Observability concepts

📖 "A *trace* is a collection of runs for a single operation." — Observability concepts

📖 "A *run* is a span representing a single unit of work within your LLM application: a call to an LLM, a prompt formatting step, a retrieval call, or any other discrete operation." — Observability concepts

📖 "Each trace is limited to a maximum of 25,000 runs. Once the trace reaches this limit, LangSmith will reject any additional runs that you send for that trace." — Observability concepts

📖 "LangSmith *integrations* provide automatic tracing for popular LLM providers and agent frameworks (the equivalent of auto-instrumentation in general observability)." — Observability concepts

📖 "*Manual instrumentation* lets you add tracing to any code, regardless of the framework. Use it when you're not using a supported integration or when you need granular control over what gets traced." — Observability concepts

from langsmith import traceable

@traceable(run_type="tool")  # 手动埋点：把该函数追踪为一个 tool span
def get_context(question: str) -> str:
    return "LangSmith traces are stored for 14 days on the Developer plan."

📖 "OpenAI wrapper: wraps the OpenAI client so every LLM call is automatically logged as a nested span." — Tracing quickstart

📖 "Traceable wrapper: wraps a function so its inputs, outputs, and any nested spans appear as a single trace in LangSmith." — Tracing quickstart

📖 "If your account is in a region other than US (the default), also set LANGSMITH_ENDPOINT to the API URL for your region. Without this, your API key won't be recognized and requests will fail to authenticate." — Tracing quickstart

export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="<your-langsmith-api-key>"
export OPENAI_API_KEY="<your-openai-api-key>"

📖 "A *thread* is a sequence of traces representing a single conversation. Each turn in a multi-turn conversation is its own trace, but traces are linked by a shared identifier. To group traces into threads, pass a special metadata key (session_id or thread_id) with a unique value." — Observability concepts

📖 "*Feedback* allows you to score an individual run based on certain criteria. Each feedback entry consists of a tag and a score, and is bound to a run by a unique run ID." — Observability concepts

📖 "LangSmith (SaaS) retains trace data for 400 days from ingestion. After that, traces are permanently deleted, with limited metadata retained for usage statistics." — Observability concepts

flowchart LR
    A[Development] --> B[Testing] --> C[Deployment] --> D[Monitoring] --> E[Iteration]
    A -.Offline.-> F[Dataset/Examples]
    C -.Online.-> G[Runs/Threads]
    E -.Both.-> H[Feedback loop]

📖 "This difference in targets determines what you can evaluate: offline evaluations can check correctness against expected answers, while online evaluations focus on quality patterns, safety, and real-world behavior." — Evaluation concepts

📖 "As applications mature, both evaluation types work together in an iterative feedback loop to improve quality continuously." — Evaluation concepts

📖 "*Code evaluators* are deterministic, rule-based functions." — Evaluation concepts

📖 "Pairwise evaluation works well when directly scoring an output is difficult but comparing two outputs is straightforward." — Evaluation concepts

📖 "LLM-as-judge evaluators require careful review of scores and prompt tuning. Few-shot evaluators, which include examples of inputs, outputs, and expected grades in the grader prompt, often improve performance." — Evaluation concepts

{
  "key": "correctness",
  "score": 1,
  "comment": "Answer is semantically equivalent to the reference."
}

📖 "Reference-free evaluators assess quality without comparing to expected outputs. These work for both offline and online evaluation:" — Evaluation concepts

📖 "Reference-based evaluators require reference outputs and only work for offline evaluation:" — Evaluation concepts

📖 "A dataset is a *collection of examples* used for evaluating an application. An example is a test input, reference output pair." — Evaluation concepts

📖 "An *experiment* represents the results of evaluating a specific application version on a dataset. Each experiment captures outputs, evaluator scores, and execution traces for every example in the dataset." — Evaluation concepts

📖 "Unlike examples in datasets, runs do not include reference outputs." — Evaluation concepts

{
  "inputs": {"question": "What is the capital of France?"},
  "reference_outputs": {"answer": "Paris"},
  "metadata": {"category": "geography", "split": "test"}
}

📖 "*Benchmarking* compares multiple application versions on a curated dataset to identify the best performer." — Evaluation types

📖 "*Regression tests* measure performance consistency across application versions over time." — Evaluation types

📖 "*Backtesting* evaluates new application versions against historical production data." — Evaluation types

# 运行 LangSmith 评估型单元测试（CI 中开启 caching 以减少 LLM API 调用）
pytest tests/ --langsmith-output

📖 "One way to evaluate an agent is to assess its overall performance on a task. This basically involves treating the agent as a black box and simply evaluating whether or not it gets the job done." — Application-specific evaluation approaches

📖 "Evaluating an agent's trajectory involves evaluating all the steps an agent took." — Application-specific evaluation approaches

flowchart LR
    U[User input] --> A[assistant node/LLM]
    A --> C{tool condition}
    C -- tool selected --> T[tool node]
    T --> A
    C -- no tool --> R[Return LLM response]

📖 "Evaluation measures performance according to metrics. Metrics can be fuzzy or subjective, and prove more useful in relative terms. They typically compare systems against each other." — Evaluation concepts

📖 "Testing asserts correctness. A system can only be deployed if it passes all tests." — Evaluation concepts

📖 "a target function that takes an input dictionary and returns an output dictionary. The example.inputs field of each Example is what gets passed to the target function." — How to evaluate agents

📖 "metadata - an optional object to attach to the experiment. Pass models, prompts, and tools keys to populate the corresponding columns in the experiment table view." — How to evaluate agents

results = ls_client.evaluate(
    toxicity_classifier,
    data=dataset.name,
    evaluators=[correct],
    experiment_prefix="gpt-5.4-mini, baseline",  # optional, experiment name prefix
    description="Testing the baseline system.",  # optional, experiment description
    max_concurrency=4,  # optional, add concurrency
    metadata=EXPERIMENT_METADATA,  # optional, used to populate model/prompt/tool columns in UI
)

📖 "Evaluators are functions for scoring your application's outputs. They take in the example inputs, actual outputs, and, when present, the reference outputs." — How to evaluate agents

📖 "These evaluators will be automatically triggered with every new experiment." — How to evaluate agents

def correct(inputs: dict, outputs: dict, reference_outputs: dict) -> bool:
    return outputs["class"] == reference_outputs["label"]

📖 "Since LLM outputs are non-deterministic, multiple repetitions provide a more accurate performance estimate." — Experiment configuration

📖 "Each repetition re-runs both the target function and all evaluators." — Experiment configuration

# 让实验重复运行 3 次,以平滑 LLM 输出波动
results = ls_client.evaluate(
    toxicity_classifier,
    data=dataset.name,
    evaluators=[correct],
    num_repetitions=3,
)

📖 "The max_concurrency argument specifies the maximum number of concurrent threads for running both the target function and evaluators." — Experiment configuration

📖 "aevaluate creates a task for each example, where each task runs the target function and all evaluators for that example. The max_concurrency argument specifies the maximum number of concurrent examples to process." — Experiment configuration

# 同步 evaluate:max_concurrency 是线程数上限
ls_client.evaluate(target, data="Toxic Queries", evaluators=[correct], max_concurrency=4)

# 异步 aevaluate:max_concurrency 是同时处理的 example 数(信号量限流)
# await aevaluate(target, data="Toxic Queries", evaluators=[correct], max_concurrency=4)

📖 "*Caching* stores API call results to disk to speed up future experiments. Set the LANGSMITH_TEST_CACHE environment variable to a valid folder path with write access." — Experiment configuration

📖 "Future experiments that make identical API calls will reuse cached results instead of making new requests." — Experiment configuration

# 指向一个可写目录即可开启评估缓存
export LANGSMITH_TEST_CACHE=./.langsmith_cache

📖 "Alerts in LangSmith are project-scoped, requiring separate configuration for each monitored project." — Alerts in LangSmith

📖 "LangSmith provides threshold-based alerting on the following metrics:" — Alerts in LangSmith

📖 "The configuration in the screenshot would generate an alert when more than 5% of runs within the past 5 minutes result in errors." — Alerts in LangSmith

Total Cost: $12.50 ≥ $5.00 · avg · 30 min

📖 "LangSmith automatically records LLM token usage and costs for major providers, and also allows you to submit custom cost data for any additional components." — Cost tracking

📖 "To compute cost automatically from token usage, you need to provide token counts, the model and provider, and the model price." — Cost tracking

from langsmith import traceable, get_current_run_tree

@traceable(run_type="tool", name="get_weather")
def get_weather(city: str):
    result = {"temperature_f": 68, "condition": "sunny", "city": city}
    tool_cost = 0.0015
    run = get_current_run_tree()
    run.set(usage_metadata={"total_cost": tool_cost})
    return result

📖 "Prebuilt dashboards are created automatically for each project and cover essential metrics, such as trace count, error rates, token usage, and more." — Monitor projects with dashboards

📖 "Group by defaults to the top 5 elements by frequency, configurable up to 20." — Monitor projects with dashboards

flowchart LR
    A[triage_input node] --> B{triage.response}
    B -->|no| C[No response needed]
    B -->|email| D[Send email back]
    B -->|notify| E[Notify user]

📖 "Pairwise annotation queues (PAQs) present two runs side-by-side so reviewers can quickly decide which output is better (or if they are equivalent) against the rubric items you define." — Use annotation queues

📖 "A run is marked Completed only when every assigned reviewer has submitted their review. Queue items progress through three states: Needs Review → Needs Others' Review → Completed." — Use annotation queues

📖 "Alerts can [route](#step-4-configure-notification-channel) to Slack, PagerDuty, Dynatrace, or any HTTP endpoint via webhook." — Alerts in LangSmith

📖 "LangSmith does not perform template substitution on the request body. The auto-populated fields above are merged into the outgoing JSON as top-level keys, alongside the body you configure." — Alerts in LangSmith

{
  "Content-Type": "application/json",
  "Authorization": "Bearer xoxb-your-token-here"
}

📖 "When you deploy Agent Server, you are deploying one or more [graphs](#graphs), a database for persistence, and a [task queue](#task-queue)." — Agent Server

📖 "Agent Server persists three types of data, all backed by PostgreSQL by default:" — Agent Server

📖 "Durability mode controls checkpoint frequency—async (default) writes after each step; exit stores only the final state." — Agent Server

flowchart LR
    G[Graph 蓝图] --> AS[Agent Server]
    AS -->|assistants/threads/runs/cron| PG[(PostgreSQL)]
    AS -->|checkpoints 短期记忆| PG
    AS -->|store 长期记忆| PG
    AS -->|信令/取消/流式 pub-sub| R[(Redis 仅临时数据)]

📖 "API servers handle client requests (creating runs, reading thread state, streaming results) but do not execute agent code themselves." — Agent Server

📖 "Queue workers are the execution engine. They listen to the durable task queue, execute your graph code, and write checkpoints." — Agent Server

📖 "The queue enforces that at most 1 run can be executed for a given thread at one time." — Agent Server

📖 "Each worker handles up to N_JOBS_PER_WORKER runs concurrently (default: 10), so a single worker container serves many runs in parallel." — Agent Server

flowchart TB
    User -->|request| API[API Server]
    API -->|create pending run| DB[(Postgres)]
    API -->|notify| Redis[(Redis)]
    Redis -->|wake| QL[Queue Loop]
    QL -->|claim next run| DB
    QL --> W[Worker 执行 graph]
    W -->|checkpoints/status| DB
    W -->|publish events| Redis
    Redis -->|stream events| API
    API -->|SSE response| User

📖 "Agent Server supports three runtime configurations:" — Agent Server

📖 "The API server manages the task queue directly with no separate queue workers. This is the default for self-hosted deployments and is suitable for development and low-traffic use cases." — Agent Server

📖 "Use this for large-scale deployments with high concurrency requirements." — Agent Server

# 开发/低流量:single host(默认,无需额外配置)
# 生产扩展:split API and queue
queue:
  enabled: true   # 开启独立 queue worker,API 与执行分主机
# 大规模高并发:distributed runtime(编排进程 + 执行进程分离)

📖 "In LangSmith, when you save changes to a prompt, you're essentially creating a new version or a "Prompt Commit." These commits are what can trigger webhooks." — How to sync prompts with GitHub

📖 "LangSmith webhooks for prompt commits are generally triggered at the workspace level." — How to sync prompts with GitHub

📖 "* Version Control: Keep your prompts versioned alongside your application code in a familiar system." — How to sync prompts with GitHub

@app.post("/webhook/commit", status_code=201, tags=["GitHub Webhooks"])
async def handle_webhook_direct_commit(payload: WebhookPayload = Body(...)):
    # payload 内含 prompt_id / prompt_name / commit_hash / manifest
    # 先 GET 现有文件 SHA,再 PUT 提交 manifest 到指定分支
    github_response = await commit_manifest_to_github(payload)
    return {"message": "Webhook received and manifest committed directly to GitHub successfully."}

📖 "Pick a environment based on where you want the control plane and data plane (Agent Servers and their databases) to run. All infrastructure types use the same Agent Server runtime." — LangSmith Deployment

📖 "LangSmith Deployment is framework-agnostic which means you can deploy agents built with:" — LangSmith Deployment

需求 → 选型
- 托管省心、agent 全托管         → Cloud(Plus+)
- agent 在自己 VPC、控制面托管   → Hybrid
- 完全数据驻留 / air-gapped      → Self-hosted with control plane(Enterprise)
- 只要 agent 运行时、无控制面     → Standalone Agent Server(自带 PG/Redis/license)