usehook
创建时间: 2026-08-28最后更新: 2026-08-28

1. 模型

在项目中，我们通过 langchain-openai 调用 DeepSeek 的 OpenAI 兼容接口。模型配置可以拆成连接信息、默认生成参数和允许使用的模型：

NOTE

我们之前选择了在 .env 中配置模型参数，具体使用时，可以根据个人喜好选择

config.toml
01
[model]

02
provider = "openai"

03
name = "deepseek-v4-flash"

04
base_url = "https://api.deepseek.com"

05
temperature = 0.2

06
max_tokens = 2048

07
timeout_seconds = 30

08
max_retries = 2

09
allowed_models = [

10
    "deepseek-v4-flash",

11
    "deepseek-v4-pro",

12
]

这些字段的含义如下：

字段含义是否通常固定在启动配置
providerLangChain 使用哪种供应商集成是
name传给模型接口的模型标识可以是默认值
base_urlAPI 基础地址是
temperature输出随机程度可以允许受控覆盖
max_tokens单次输出上限可以允许受控覆盖
timeout_seconds等待上游响应的超时是
max_retries可重试错误的最大重试次数是
allowed_models服务端允许调用的模型列表是

provider 与模型名称不是一回事。使用 OpenAI 兼容接口时，LangChain 仍然需要 OpenAI 集成来组织请求，但 name 会原样交给实际服务商：

app/agent/model.py
01
from langchain_openai import ChatOpenAI

02


03
from app.core.settings import Settings

04


05


06
def create_model(settings: Settings) -> ChatOpenAI:

07
    api_key = settings.model.api_key

08
    if api_key is None:

09
        raise RuntimeError("模型 API Key 未配置")

10


11
    return ChatOpenAI(

12
        model=settings.model.name,

13
        api_key=api_key.get_secret_value(),

14
        base_url=str(settings.model.base_url),

15
        temperature=settings.model.temperature,

16
        max_tokens=settings.model.max_tokens,

17
        timeout=settings.model.timeout_seconds,

18
        max_retries=settings.model.max_retries,

19
    )

ChatOpenAI 封装的是消息转换、HTTP 调用、流式输出、重试和响应对象构造等客户端逻辑。base_url 指向 DeepSeek 后，请求实际发送给 DeepSeek，不会因为类名中有 OpenAI 就改成调用 OpenAI 官方服务。

不同服务商对兼容参数的支持并不完全一致。TOML 能成功解析、LangChain 构造函数能接收参数，不代表上游一定会按相同方式执行。模型是否支持温度、工具调用、结构化输出、推理参数和流式 usage，需要以当前服务商文档与真实测试为准。

LangChain 也提供了专门的 langchain-deepseek 集成。只需要兼容接口中的基础文本能力时，可以继续使用 ChatOpenAI；如果要读取 reasoning_content 等 DeepSeek 扩展字段，或者使用供应商特有能力，应优先选择 ChatDeepSeek。此时还要安装 langchain-deepseek，并把 provider、模型工厂和依赖声明一起调整，不能只替换类名。

2. Settings

模型 Settings 可以使用 Pydantic 校验范围：

app/core/settings.py
01
from typing import Literal

02


03
from pydantic import BaseModel, Field, HttpUrl, SecretStr, model_validator

04


05


06
class ModelSettings(BaseModel):

07
    provider: Literal["openai"] = "openai"

08
    name: str = Field(min_length=1)

09
    base_url: HttpUrl

10
    api_key: SecretStr | None = None

11
    temperature: float = Field(default=0.2, ge=0, le=2)

12
    max_tokens: int = Field(default=2048, gt=0, le=100_000)

13
    timeout_seconds: float = Field(default=30, gt=0, le=300)

14
    max_retries: int = Field(default=2, ge=0, le=10)

15
    allowed_models: list[str] = Field(min_length=1)

16


17
    @model_validator(mode="after")

18
    def validate_default_model(self) -> "ModelSettings":

19
        if self.name not in self.allowed_models:

20
            raise ValueError("默认模型必须位于 allowed_models 中")

21
        return self

这里把 provider 限制为当前项目真正接入的 openai，这样拼写错误会在启动时暴露。max_tokens 的 100_000 也是项目为了控制成本和响应时间设定的上限，不代表 DeepSeek 接口本身的最大输出能力。

API Key 仍然通过环境提供：

.env.example
1
MODEL__API_KEY=your-api-key

不要在 config.toml 中写真实密钥：

insecure-config.toml
1
# 不要提交这种配置

2
[model]

3
api_key = "sk-real-secret"

配置中的 name 是默认模型，allowed_models 是服务端允许列表。用户可以请求切换模型，但不能把任意字符串直接传给供应商：

model-policy.toml
1
[model]

2
name = "deepseek-v4-flash"

3
allowed_models = [

4
    "deepseek-v4-flash",

5
    "deepseek-v4-pro",

6
]
app/services/model_policy.py
1
from app.core.settings import Settings

2


3


4
def resolve_model(requested: str | None, settings: Settings) -> str:

5
    model = requested or settings.model.name

6
    if model not in settings.model.allowed_models:

7
        raise ValueError(f"不支持的模型：{model}")

8
    return model

如果不同模型使用不同 base_url 或 API Key，仅切换模型名不够。此时应配置完整模型档案，并在服务端按受控 ID 选择，不能让客户端提交任意 URL 和密钥。

3. 动态参数

配置要区分进程级配置、业务输入、运行上下文和生成参数：

类别示例
进程级配置base_url、API Key、HTTP 超时、最大重试、连接池
业务输入用户当前发送的问题
运行上下文会话 thread_id、用户身份和请求 ID
生成参数允许范围内的模型、温度、输出长度和输出格式

进程级配置适合在应用启动时创建模型客户端。其余内容随请求变化，不应通过修改全局 Settings 实现，否则两个并发请求可能互相覆盖。

LangChain 模型可以通过 bind() 派生本次调用参数：

app/services/chat.py
01
from collections.abc import Sequence

02


03
from langchain_core.messages import BaseMessage

04
from langchain_openai import ChatOpenAI

05


06


07
async def reply(

08
    model: ChatOpenAI,

09
    messages: Sequence[BaseMessage],

10
    temperature: float,

11
) -> str:

12
    request_model = model.bind(temperature=temperature)

13
    response = await request_model.ainvoke(messages)

14
    return response.text

bind() 返回一个带有本次调用参数的新 Runnable，不会修改 TOML，也不会改写共享模型实例。请求参数仍需通过 Pydantic Schema 限制范围。

LangChain 的 init_chat_model() 还支持 configurable fields，让调用时通过 Runnable config 切换受控字段。它适合一个图中需要声明式切换模型的场景，但不要把所有字段都暴露为 configurable：

configurable-model.py
01
from langchain.chat_models import init_chat_model

02


03
from app.core.settings import Settings

04


05


06
def create_configurable_model(settings: Settings):

07
    api_key = settings.model.api_key

08
    if api_key is None:

09
        raise RuntimeError("模型 API Key 未配置")

10


11
    return init_chat_model(

12
        model=settings.model.name,

13
        model_provider=settings.model.provider,

14
        base_url=str(settings.model.base_url),

15
        api_key=api_key.get_secret_value(),

16
        configurable_fields=("model", "temperature", "max_tokens"),

17
        config_prefix="chat",

18
    )

调用时传入的字段仍应由服务端白名单生成：

invoke-configurable.py
01
response = await model.ainvoke(

02
    messages,

03
    config={

04
        "configurable": {

05
            "chat_model": selected_model,

06
            "chat_temperature": temperature,

07
            "chat_max_tokens": max_tokens,

08
        },

09
    },

10
)

Runnable config 是一次调用的上下文，不等于应用 TOML。前者随请求变化，后者在进程启动时确定默认策略。

4. LangGraph

LangGraph 配置可以分为图构建参数、持久化参数和每次运行参数：

langgraph-config.toml
1
[langgraph]

2
recursion_limit = 50

3
checkpointer = "memory"

4
store = "memory"

5


6
[langgraph.postgres]

7
pool_size = 10

8
pool_timeout_seconds = 30

各字段职责：

字段作用
recursion_limit限制一次图运行最多继续多少步，防止错误循环无限执行
checkpointer保存每个 thread 的图状态与检查点
store保存可跨 thread 查询的长期数据
pool_size生产数据库连接池容量
pool_timeout_seconds获取数据库连接的等待上限

checkpointer 与 Store 不相同：

persistence-boundary.txt
1
checkpointer：保存一次会话 thread 内的状态、步骤和恢复点

2
store：保存跨 thread 使用的长期记忆或共享数据

开发和测试可以使用内存实现。下面的示例同时补全了最小图结构，model 会在节点中真正参与调用：

app/graph.py
01
from langchain_core.language_models import BaseChatModel

02
from langgraph.checkpoint.memory import InMemorySaver

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

04
from langgraph.store.memory import InMemoryStore

05


06


07
def create_memory_graph(model: BaseChatModel):

08
    async def call_model(state: MessagesState):

09
        response = await model.ainvoke(state["messages"])

10
        return {"messages": [response]}

11


12
    builder = StateGraph(MessagesState)

13
    builder.add_node("model", call_model)

14
    builder.add_edge(START, "model")

15
    builder.add_edge("model", END)

16


17
    checkpointer = InMemorySaver()

18
    store = InMemoryStore()

19


20
    return builder.compile(

21
        checkpointer=checkpointer,

22
        store=store,

23
    )

内存状态只存在于当前进程。重启会丢失，多 worker 之间也不共享，因此不适合需要可靠恢复的生产服务。这段函数固定创建内存实现，不会读取 TOML；在完整项目中，还需要由资源工厂读取 settings.langgraph.checkpointer 和 settings.langgraph.store，再选择对应实现。只修改配置字段，并不会自动切换后端。

生产环境可以在 TOML 中选择 PostgreSQL 后端，但连接 URL 仍通过敏感环境变量提供。双下划线表示嵌套层级，因此下面的变量对应 settings.langgraph.postgres.url：

postgres-secret.env
1
LANGGRAPH__POSTGRES__URL=postgresql://user:password@db/agent

图编译时注入 checkpointer。每次调用时提供 thread_id 和递归限制：

invoke-graph.py
1
result = await graph.ainvoke(

2
    {"messages": messages},

3
    config={

4
        "configurable": {

5
            "thread_id": conversation_id,

6
        },

7
        "recursion_limit": settings.langgraph.recursion_limit,

8
    },

9
)

thread_id 是业务会话标识，必须来自当前已授权用户可访问的会话。不能让攻击者猜一个 ID 就读取其他用户的图状态。配置文件只能提供命名规则或默认策略，不能代替数据库授权检查。

5. 追踪

LangSmith tracing 常通过环境变量启用：

tracing.env
1
LANGSMITH_TRACING=true

2
LANGSMITH_API_KEY=your-langsmith-key

3
LANGSMITH_PROJECT=fastapi-langchain-development

4
LANGSMITH_TRACING_SAMPLING_RATE=0.1

TOML 可以保存非敏感策略：

tracing.toml
1
[observability]

2
tracing_enabled = false

3
project = "fastapi-langchain-development"

4
sampling_rate = 0.1

但是，仅写 TOML 不会自动启用 LangSmith。sampling_rate 也不是 LangSmith 会自动查找的 TOML 字段。部署平台可以直接提供上面的 LangSmith 环境变量；如果项目把非敏感策略统一放在 TOML 中，就要在创建模型、Agent 或图之前完成一次显式映射：

app/core/observability.py
01
import os

02


03
from app.core.settings import Settings

04


05


06
def configure_langsmith(settings: Settings) -> None:

07
    observability = settings.observability

08
    os.environ["LANGSMITH_TRACING"] = str(

09
        observability.tracing_enabled,

10
    ).lower()

11
    os.environ["LANGSMITH_PROJECT"] = observability.project

12
    os.environ["LANGSMITH_TRACING_SAMPLING_RATE"] = str(

13
        observability.sampling_rate,

14
    )

这段代码只映射公开策略，LANGSMITH_API_KEY 仍由运行环境提供。若项目改为手动创建 LangSmith Client，也可以把采样率传给 tracing_sampling_rate，再通过 tracing context 使用该 Client；两种接入方式选择一种即可。

追踪配置需要按环境管理：

环境建议
test默认关闭，避免上传测试数据
development开发者明确开启
staging可以较高采样，验证完整链路
production根据隐私、成本和排障需求设置采样

LangSmith API Key 必须由环境提供。项目名称可以进入 TOML，但生产环境仍可覆盖。

追踪可能包含提示词、工具输入输出、模型回答和用户数据。启用前要确认脱敏、访问控制、保留时间与隐私策略。SecretStr 只保护配置对象打印，不会自动从 tracing 中移除用户内容。

6. 部署

本章讲的是 Python 应用如何使用 TOML 管理自己的运行配置。LangGraph CLI 和 LangSmith Deployment 还有独立的 langgraph.json：

langgraph.json
1
{

2
  "dependencies": ["."],

3
  "graphs": {

4
    "assistant": "./app/graph.py:graph"

5
  },

6
  "env": "./.env"

7
}

它告诉 LangGraph 工具：

  • 应安装哪些项目依赖；
  • 哪些图需要对外提供；
  • 图对象或工厂函数位于哪个模块；
  • 本地开发从哪里读取环境变量。

langgraph.json 与 pyproject.toml 可以同时存在：

langgraph-files.txt
1
pyproject.toml  -> Python 依赖和工具配置

2
config.toml     -> 我们的应用默认策略

3
.env            -> 本地密钥和环境覆盖

4
langgraph.json  -> LangGraph CLI/部署的图入口配置

不能因为喜欢 TOML 就把 langgraph.json 改成 langgraph.toml，LangGraph CLI 默认不会识别自创格式。也不要在 langgraph.json 的 env 字段中写密钥值，它应指向环境文件或由部署平台提供环境变量。

如果当前项目只通过 FastAPI 自己托管编译图，不使用 LangGraph CLI/Agent Server，就不必提前创建 langgraph.json。是否需要它取决于部署方式。

启动校验

AI 配置不仅要检查类型，还要检查组合关系：

app/core/settings.py
01
from typing import Literal

02


03
from pydantic import BaseModel, Field, SecretStr, model_validator

04


05


06
class PostgresSettings(BaseModel):

07
    url: SecretStr | None = None

08
    pool_size: int = Field(default=10, gt=0)

09
    pool_timeout_seconds: float = Field(default=30, gt=0)

10


11


12
class LangGraphSettings(BaseModel):

13
    checkpointer: Literal["memory", "postgres"] = "memory"

14
    store: Literal["memory", "postgres"] = "memory"

15
    recursion_limit: int = Field(default=50, gt=0, le=500)

16
    postgres: PostgresSettings = Field(default_factory=PostgresSettings)

17


18
    @model_validator(mode="after")

19
    def validate_backend(self) -> "LangGraphSettings":

20
        uses_postgres = (

21
            self.checkpointer == "postgres" or self.store == "postgres"

22
        )

23
        if uses_postgres and self.postgres.url is None:

24
            raise ValueError("PostgreSQL 后端需要数据库 URL")

25
        return self

生产启动还应执行有超时的连通性检查，但不要让检查产生昂贵模型回答。可以验证：

  1. 数据库能建立连接并执行轻量查询；
  2. checkpointer 所需表是否准备完成；
  3. 模型凭据是否存在；
  4. 必要时执行极小的模型连接测试；
  5. tracing 开启时 LangSmith 配置是否完整。

失败信息应指出哪个组件不可用，但不能包含 API Key、数据库密码或完整 Authorization 头。

重试也要有边界。配置 max_retries = 10 可能把一次不可恢复的配置错误拖成很长超时。认证失败、模型不存在等客户端错误通常不应反复重试；网络抖动和限流才可能适合退避重试。

正在验证登录状态
请稍候，验证完成后将继续显示文章内容
0
订阅