Back to Details

agent-workflow

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

LangChain / LangGraph

LangChain / LangGraph

Overview

概述

LangChain is the standard framework for building applications powered by large language models. It provides abstractions for chat models, prompt templates, output parsers, tool integration, and retrieval augmented generation. LangChain Expression Language (LCEL) enables declarative composition of these components into chains using a pipe syntax.
LangGraph extends LangChain with a graph-based runtime for building stateful, multi-step agent workflows. It models agent logic as a state graph where nodes are functions, edges define control flow, and state is persisted across turns via checkpointing. LangGraph is the recommended approach for any agent that requires loops, branching, tool-calling cycles, human-in-the-loop intervention, or long-running conversations.
Key characteristics:
  • LangChain: model abstraction, prompt management, output parsing, tool definitions, retrieval
  • LangGraph: state machines for agents, persistence, streaming, human-in-the-loop, multi-agent orchestration
  • LCEL: composable pipe syntax for simple prompt-model-parser chains
  • Production-ready: built-in checkpointing, error handling, streaming, and observability
LangChain是构建大语言模型（LLM）驱动应用的标准框架。它为聊天模型、提示词模板、输出解析器、工具集成以及检索增强生成（RAG）提供了抽象能力。LangChain表达式语言（LCEL）允许通过管道语法将这些组件声明式地组合成链。
LangGraph是LangChain的扩展，提供了基于图的运行时环境，用于构建有状态的多步骤Agent工作流。它将Agent逻辑建模为状态图，其中节点是函数，边定义控制流，状态通过检查点机制在多轮交互中持久化。对于需要循环、分支、工具调用周期、人在回路干预或长对话的Agent，LangGraph是推荐的实现方案。
核心特性：
  • LangChain：模型抽象、提示词管理、输出解析、工具定义、检索能力
  • LangGraph：Agent状态机、状态持久化、流式输出、人在回路、多Agent编排
  • LCEL：用于构建简单提示词-模型-解析器链的可组合管道语法
  • 生产就绪：内置检查点、错误处理、流式输出和可观测性

Chat Models

聊天模型

Chat models are the primary interface to LLMs. Always import from 
langchain_<provider>
packages rather than the deprecated 
langchain.chat_models
.
python
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
聊天模型是与LLM交互的主要接口。请始终从
langchain_<provider>
包导入，而非已弃用的
langchain.chat_models
。
python
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

OpenAI

OpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm = ChatOpenAI(model="gpt-4o", temperature=0)

Anthropic

Anthropic

llm = ChatAnthropic(model="claude-sonnet-4-5-20250929", temperature=0)
undefined
llm = ChatAnthropic(model="claude-sonnet-4-5-20250929", temperature=0)
undefined

Message Types

消息类型

LangChain uses typed message objects for chat interactions:
python
from langchain_core.messages import (
    HumanMessage,
    AIMessage,
    SystemMessage,
    ToolMessage,
)

messages = [
    SystemMessage(content="Answer concisely."),
    HumanMessage(content="What is LangGraph?"),
]

response = llm.invoke(messages)  # returns AIMessage
AIMessage
includes 
content
(the text response), 
tool_calls
(list of tool invocations), and 
response_metadata
(token usage, model info).
LangChain使用类型化的消息对象进行聊天交互：
python
from langchain_core.messages import (
    HumanMessage,
    AIMessage,
    SystemMessage,
    ToolMessage,
)

messages = [
    SystemMessage(content="Answer concisely."),
    HumanMessage(content="What is LangGraph?"),
]

response = llm.invoke(messages)  # returns AIMessage
AIMessage
包含
content
（文本响应）、
tool_calls
（工具调用列表）和
response_metadata
（令牌使用量、模型信息）。

Prompt Templates

提示词模板

Use 
ChatPromptTemplate
to create reusable, parameterized prompts:
python
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "Answer questions about {topic}. Be concise."),
    MessagesPlaceholder("chat_history", optional=True),
    ("human", "{question}"),
])
使用
ChatPromptTemplate
创建可复用的参数化提示词：
python
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "Answer questions about {topic}. Be concise."),
    MessagesPlaceholder("chat_history", optional=True),
    ("human", "{question}"),
])

Invoke with variables

传入变量调用

messages = prompt.invoke({ "topic": "Python", "question": "What is asyncio?", "chat_history": [], })

`MessagesPlaceholder` injects a list of messages at that position -- essential for conversation history and agent scratchpads. Mark it `optional=True` when the variable may be absent.
messages = prompt.invoke({ "topic": "Python", "question": "What is asyncio?", "chat_history": [], })

`MessagesPlaceholder`会在对应位置注入消息列表——这对于对话历史和Agent草稿本至关重要。当变量可能不存在时，将其标记为`optional=True`。

Output Parsers

输出解析器

Parse LLM output into structured formats:
python
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
将LLM输出解析为结构化格式：
python
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field

Simple string output

简单字符串输出

parser = StrOutputParser()
parser = StrOutputParser()

Pydantic structured output

Pydantic结构化输出

class MovieReview(BaseModel): title: str = Field(description="Movie title") rating: int = Field(description="Rating from 1 to 10", ge=1, le=10) summary: str = Field(description="Brief summary")
pydantic_parser = PydanticOutputParser(pydantic_object=MovieReview)
class MovieReview(BaseModel): title: str = Field(description="Movie title") rating: int = Field(description="Rating from 1 to 10", ge=1, le=10) summary: str = Field(description="Brief summary")
pydantic_parser = PydanticOutputParser(pydantic_object=MovieReview)

JSON output

JSON输出

json_parser = JsonOutputParser(pydantic_object=MovieReview)

For structured output with tool-calling models, prefer `model.with_structured_output(MovieReview)` over manual parser injection -- it uses the model's native function-calling capability.

```python
structured_llm = llm.with_structured_output(MovieReview)
result = structured_llm.invoke("Review the movie Inception")
json_parser = JsonOutputParser(pydantic_object=MovieReview)

对于支持工具调用的模型的结构化输出，优先使用`model.with_structured_output(MovieReview)`而非手动注入解析器——它会利用模型原生的函数调用能力。

```python
structured_llm = llm.with_structured_output(MovieReview)
result = structured_llm.invoke("Review the movie Inception")

result is a MovieReview instance

result是MovieReview实例


For Pydantic structured output, consult the `pydantic` skill.

关于Pydantic结构化输出的更多内容，请参考`pydantic`技能。

LangChain Expression Language (LCEL)

LangChain表达式语言（LCEL）

LCEL composes components using the pipe (
|
) operator. Each component in the chain must be a 
Runnable
.
python
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_messages([
    ("system", "Translate the following to {language}."),
    ("human", "{text}"),
])

chain = prompt | ChatOpenAI(model="gpt-4o") | StrOutputParser()

result = chain.invoke({"language": "French", "text": "Hello, world!"})
LCEL使用管道（
|
）运算符组合组件。链中的每个组件必须是
Runnable
类型。
python
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_messages([
    ("system", "Translate the following to {language}."),
    ("human", "{text}"),
])

chain = prompt | ChatOpenAI(model="gpt-4o") | StrOutputParser()

result = chain.invoke({"language": "French", "text": "Hello, world!"})

Key Runnables

核心Runnable组件

python
from langchain_core.runnables import RunnablePassthrough, RunnableLambda, RunnableParallel
python
from langchain_core.runnables import RunnablePassthrough, RunnableLambda, RunnableParallel

Pass input through unchanged (useful for forwarding data)

保持输入不变（用于转发数据）

RunnablePassthrough()
RunnablePassthrough()

Wrap any function as a Runnable

将任意函数包装为Runnable

RunnableLambda(lambda x: x["text"].upper())
RunnableLambda(lambda x: x["text"].upper())

Run multiple chains in parallel, merge results

并行运行多个链，合并结果

RunnableParallel( summary=summary_chain, keywords=keywords_chain, )
undefined
RunnableParallel( summary=summary_chain, keywords=keywords_chain, )
undefined

Chaining with Context

带上下文的链组合

A common pattern passes retrieved context alongside the original question:
python
from langchain_core.runnables import RunnablePassthrough

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)
LCEL chains support 
.invoke()
, 
.ainvoke()
, 
.stream()
, 
.astream()
, 
.batch()
, and 
.abatch()
out of the box.
一种常见模式是将检索到的上下文与原始问题一起传递：
python
from langchain_core.runnables import RunnablePassthrough

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)
LCEL链默认支持
.invoke()
、
.ainvoke()
、
.stream()
、
.astream()
、
.batch()
和
.abatch()
方法。

Tool Creation

工具创建

Tools give LLMs the ability to call external functions. Define tools with clear names and descriptions so the model knows when and how to use them.
工具让LLM具备调用外部函数的能力。定义工具时要提供清晰的名称和描述，以便模型了解何时以及如何使用它们。

@tool Decorator

@tool装饰器

python
from langchain_core.tools import tool

@tool
def search_database(query: str, limit: int = 10) -> list[dict]:
    """Search the product database by query string.

    Args:
        query: The search query to match against product names and descriptions.
        limit: Maximum number of results to return.
    """
    return db.search(query, limit=limit)
The decorator extracts the function signature and docstring to build the tool schema automatically.
python
from langchain_core.tools import tool

@tool
def search_database(query: str, limit: int = 10) -> list[dict]:
    """Search the product database by query string.

    Args:
        query: The search query to match against product names and descriptions.
        limit: Maximum number of results to return.
    """
    return db.search(query, limit=limit)
该装饰器会自动提取函数签名和文档字符串来构建工具的Schema。

StructuredTool with Pydantic Schema

带Pydantic Schema的StructuredTool

For complex argument validation, provide an explicit 
args_schema
:
python
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

class SearchArgs(BaseModel):
    query: str = Field(description="Search query string")
    category: str = Field(description="Product category to filter")
    limit: int = Field(default=10, ge=1, le=100)

search_tool = StructuredTool.from_function(
    func=search_database,
    name="search_database",
    description="Search the product database with filters",
    args_schema=SearchArgs,
)
对于复杂的参数验证，提供显式的
args_schema
：
python
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field

class SearchArgs(BaseModel):
    query: str = Field(description="Search query string")
    category: str = Field(description="Product category to filter")
    limit: int = Field(default=10, ge=1, le=100)

search_tool = StructuredTool.from_function(
    func=search_database,
    name="search_database",
    description="Search the product database with filters",
    args_schema=SearchArgs,
)

Binding Tools to Models

将工具绑定到模型

python
tools = [search_database, get_weather, calculate]
python
tools = [search_database, get_weather, calculate]

Bind tools to a model

将工具绑定到模型

llm_with_tools = llm.bind_tools(tools)
llm_with_tools = llm.bind_tools(tools)

Force a specific tool

强制使用特定工具

llm_with_tools = llm.bind_tools(tools, tool_choice="search_database")
llm_with_tools = llm.bind_tools(tools, tool_choice="search_database")

Parse tool calls from the response

从响应中解析工具调用

response = llm_with_tools.invoke("Find laptops under $1000") for tool_call in response.tool_calls: print(tool_call["name"], tool_call["args"])

See `references/tools.md` for async tools, error handling, built-in tools, retriever tools, and advanced patterns.
response = llm_with_tools.invoke("Find laptops under $1000") for tool_call in response.tool_calls: print(tool_call["name"], tool_call["args"])

关于异步工具、错误处理、内置工具、检索器工具以及高级模式，请参考`references/tools.md`。

LangGraph Fundamentals

LangGraph基础

LangGraph models agent workflows as directed graphs. The three core concepts are state, nodes, and edges.
LangGraph将Agent工作流建模为有向图。三个核心概念是状态、节点和边。

State Definition

状态定义

Define the graph state as a 
TypedDict
. Use 
Annotated
fields with reducer functions to control how node outputs merge into existing state.
python
import operator
from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages


class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]  # append messages
    context: str                                           # overwrite on each update
    iteration_count: Annotated[int, operator.add]          # sum values
  • add_messages
    -- appends new messages, deduplicates by ID, and handles 
    RemoveMessage
    .
  • operator.add
    -- sums numeric values from successive updates.
  • No annotation -- the latest value overwrites the previous one.
将图的状态定义为
TypedDict
。使用带有归约函数的
Annotated
字段来控制节点输出如何合并到现有状态中。
python
import operator
from typing import Annotated, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages


class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]  # 追加消息
    context: str                                           # 每次更新时覆盖
    iteration_count: Annotated[int, operator.add]          # 求和
  • add_messages
    ——追加新消息，按ID去重，并处理
    RemoveMessage
    。
  • operator.add
    ——对连续更新的数值求和。
  • 无注解——最新值覆盖之前的值。

Nodes

节点

Nodes are functions (sync or async) that receive the current state and return a partial state update:
python
from langchain_core.messages import AIMessage

async def agent_node(state: AgentState) -> dict:
    """Call the LLM with the current messages."""
    response = await llm_with_tools.ainvoke(state["messages"])
    return {"messages": [response]}

async def process_node(state: AgentState) -> dict:
    """Post-process the agent output."""
    last_message = state["messages"][-1]
    return {"context": last_message.content, "iteration_count": 1}
Nodes return only the keys being updated. The reducer merges the partial update into the full state.
节点是接收当前状态并返回部分状态更新的函数（同步或异步）：
python
from langchain_core.messages import AIMessage

async def agent_node(state: AgentState) -> dict:
    """使用当前消息调用LLM。"""
    response = await llm_with_tools.ainvoke(state["messages"])
    return {"messages": [response]}

async def process_node(state: AgentState) -> dict:
    """后处理Agent的输出。"""
    last_message = state["messages"][-1]
    return {"context": last_message.content, "iteration_count": 1}
节点只返回需要更新的键。归约函数会将部分更新合并到完整状态中。

Edges and Conditional Routing

边和条件路由

python
from langgraph.graph import StateGraph, END

graph = StateGraph(AgentState)
python
from langgraph.graph import StateGraph, END

graph = StateGraph(AgentState)

Add nodes

添加节点

graph.add_node("agent", agent_node) graph.add_node("tools", tool_node) graph.add_node("process", process_node)
graph.add_node("agent", agent_node) graph.add_node("tools", tool_node) graph.add_node("process", process_node)

Static edge

静态边

graph.add_edge("tools", "agent")
graph.add_edge("tools", "agent")

Conditional edge -- route based on state

条件边——根据状态路由

def should_continue(state: AgentState) -> str: last_message = state["messages"][-1] if last_message.tool_calls: return "tools" return "process"
graph.add_conditional_edges("agent", should_continue)
def should_continue(state: AgentState) -> str: last_message = state["messages"][-1] if last_message.tool_calls: return "tools" return "process"
graph.add_conditional_edges("agent", should_continue)

Terminal edge

终止边

graph.add_edge("process", END)
graph.add_edge("process", END)

Set entry point

设置入口点

graph.set_entry_point("agent")
undefined
graph.set_entry_point("agent")
undefined

Graph Compilation

图编译

Compile the graph into a runnable application, optionally with a checkpointer for state persistence:
python
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)
将图编译为可运行的应用，可选地使用检查点工具实现状态持久化：
python
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

Invocation

调用

python
from langchain_core.messages import HumanMessage

config = {"configurable": {"thread_id": "user-123"}}

result = await app.ainvoke(
    {"messages": [HumanMessage(content="Find laptops under $1000")]},
    config=config,
)
The 
thread_id
links invocations to a persistent conversation. The checkpointer stores state between calls, enabling multi-turn conversations.
python
from langchain_core.messages import HumanMessage

config = {"configurable": {"thread_id": "user-123"}}

result = await app.ainvoke(
    {"messages": [HumanMessage(content="Find laptops under $1000")]},
    config=config,
)
thread_id
将调用与持久化的对话关联起来。检查点工具会在多次调用之间保存状态，支持多轮对话。

Basic Agent Pattern

基础Agent模式

The canonical LangGraph agent follows a tool-calling loop: the LLM decides whether to call tools or respond directly.
python
from langgraph.prebuilt import ToolNode

tools = [search_database, get_weather]
tool_node = ToolNode(tools)
llm_with_tools = llm.bind_tools(tools)

async def call_model(state: AgentState) -> dict:
    response = await llm_with_tools.ainvoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: AgentState) -> str:
    if state["messages"][-1].tool_calls:
        return "tools"
    return END

graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")

agent = graph.compile(checkpointer=MemorySaver())
See 
assets/graph-template.py
for a complete production-ready template with FastAPI integration.
标准的LangGraph Agent遵循工具调用循环：LLM决定是调用工具还是直接响应。
python
from langgraph.prebuilt import ToolNode

tools = [search_database, get_weather]
tool_node = ToolNode(tools)
llm_with_tools = llm.bind_tools(tools)

async def call_model(state: AgentState) -> dict:
    response = await llm_with_tools.ainvoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: AgentState) -> str:
    if state["messages"][-1].tool_calls:
        return "tools"
    return END

graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")

agent = graph.compile(checkpointer=MemorySaver())
完整的生产就绪模板（含FastAPI集成）请参考
assets/graph-template.py
。

Checkpointing and Persistence

检查点与持久化

Checkpointers save the graph state after every node execution, enabling conversation memory, fault recovery, and time-travel debugging.
python
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
检查点工具会在每个节点执行后保存图的状态，支持对话记忆、故障恢复和时间旅行调试。
python
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver

In-memory (development)

内存存储（开发环境）

checkpointer = MemorySaver()
checkpointer = MemorySaver()

SQLite (single-server persistence)

SQLite存储（单服务器持久化）

checkpointer = AsyncSqliteSaver.from_conn_string("checkpoints.db")
checkpointer = AsyncSqliteSaver.from_conn_string("checkpoints.db")

PostgreSQL (production -- requires langgraph-checkpoint-postgres)

PostgreSQL存储（生产环境——需要langgraph-checkpoint-postgres）

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver checkpointer = AsyncPostgresSaver.from_conn_string(database_url)

Always provide a `thread_id` in the config to isolate conversations:

```python
config = {"configurable": {"thread_id": "session-abc123"}}
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver checkpointer = AsyncPostgresSaver.from_conn_string(database_url)

始终在配置中提供`thread_id`以隔离对话：

```python
config = {"configurable": {"thread_id": "session-abc123"}}

Human-in-the-Loop

人在回路

Interrupt graph execution to request human approval or input before proceeding:
python
undefined
在继续执行之前中断图的执行，请求人工批准或输入：
python
undefined

Interrupt before the tool node executes

在工具节点执行前中断

app = graph.compile( checkpointer=checkpointer, interrupt_before=["tools"], )
app = graph.compile( checkpointer=checkpointer, interrupt_before=["tools"], )

Run until interrupted

运行直到中断

result = await app.ainvoke( {"messages": [HumanMessage(content="Delete all records")]}, config=config, )
result = await app.ainvoke( {"messages": [HumanMessage(content="Delete all records")]}, config=config, )

Inspect pending tool calls, get human approval, then resume

检查待处理的工具调用，获取人工批准，然后恢复执行

result = await app.ainvoke(None, config=config)

Use `interrupt_after` to pause after a node completes, allowing inspection of its output before the graph continues.
result = await app.ainvoke(None, config=config)

使用`interrupt_after`在节点完成后暂停，允许在图继续执行前检查其输出。

Streaming

流式输出

LangGraph supports multiple streaming modes for real-time output.
LangGraph支持多种流式输出模式以实现实时输出。

Stream State Updates

流式状态更新

python
async for event in app.astream(
    {"messages": [HumanMessage(content="Summarize this document")]},
    config=config,
):
    for node_name, output in event.items():
        print(f"Node '{node_name}': {output}")
python
async for event in app.astream(
    {"messages": [HumanMessage(content="Summarize this document")]},
    config=config,
):
    for node_name, output in event.items():
        print(f"节点 '{node_name}': {output}")

Stream Individual Tokens

流式单个令牌

python
async for event in app.astream_events(
    {"messages": [HumanMessage(content="Write a poem")]},
    config=config,
    version="v2",
):
    if event["event"] == "on_chat_model_stream":
        token = event["data"]["chunk"].content
        if token:
            print(token, end="", flush=True)
astream_events()
provides fine-grained events for every component in the graph: model tokens, tool starts/ends, retriever results, and custom events.
python
async for event in app.astream_events(
    {"messages": [HumanMessage(content="Write a poem")]},
    config=config,
    version="v2",
):
    if event["event"] == "on_chat_model_stream":
        token = event["data"]["chunk"].content
        if token:
            print(token, end="", flush=True)
astream_events()
为图中的每个组件提供细粒度事件：模型令牌、工具启动/结束、检索器结果和自定义事件。

Integration with FastAPI

与FastAPI集成

Mount a LangGraph agent as a FastAPI endpoint for production serving:
python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_core.messages import HumanMessage

app = FastAPI()

@app.post("/chat")
async def chat(request: ChatRequest):
    config = {"configurable": {"thread_id": request.thread_id}}
    result = await agent.ainvoke(
        {"messages": [HumanMessage(content=request.message)]},
        config=config,
    )
    return {"response": result["messages"][-1].content}

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    config = {"configurable": {"thread_id": request.thread_id}}

    async def event_generator():
        async for event in agent.astream_events(
            {"messages": [HumanMessage(content=request.message)]},
            config=config,
            version="v2",
        ):
            if event["event"] == "on_chat_model_stream":
                token = event["data"]["chunk"].content
                if token:
                    yield f"data: {token}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")
For full FastAPI application structure and patterns, consult the 
fastapi
skill. For async patterns in agents, consult the 
async-patterns
skill.
将LangGraph Agent挂载为FastAPI端点以用于生产部署：
python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_core.messages import HumanMessage

app = FastAPI()

@app.post("/chat")
async def chat(request: ChatRequest):
    config = {"configurable": {"thread_id": request.thread_id}}
    result = await agent.ainvoke(
        {"messages": [HumanMessage(content=request.message)]},
        config=config,
    )
    return {"response": result["messages"][-1].content}

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    config = {"configurable": {"thread_id": request.thread_id}}

    async def event_generator():
        async for event in agent.astream_events(
            {"messages": [HumanMessage(content=request.message)]},
            config=config,
            version="v2",
        ):
            if event["event"] == "on_chat_model_stream":
                token = event["data"]["chunk"].content
                if token:
                    yield f"data: {token}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")
关于完整的FastAPI应用结构和模式，请参考
fastapi
技能。关于Agent中的异步模式，请参考
async-patterns
技能。

Further Reading

进一步阅读

  • references/langgraph-workflows.md
    -- complex graph patterns, multi-agent systems, production deployment
  • references/tools.md
    -- tool creation, built-in tools, error handling, retriever tools
  • assets/graph-template.py
    -- production-ready agent template with FastAPI integration
Cross-references:
  • pydantic skill -- structured output models, validators, serialization
  • async-patterns skill -- async/await best practices, concurrency primitives
  • fastapi skill -- API endpoints, middleware, dependency injection
  • references/langgraph-workflows.md
    ——复杂图模式、多Agent系统、生产部署
  • references/tools.md
    ——工具创建、内置工具、错误处理、检索器工具
  • assets/graph-template.py
    ——含FastAPI集成的生产就绪Agent模板
交叉引用：
  • pydantic技能——结构化输出模型、验证器、序列化
  • async-patterns技能——async/await最佳实践、并发原语
  • fastapi技能——API端点、中间件、依赖注入