LangChain 生产发布检查表：可执行预检、故障注入与并发会话实测

"""Write a tiny fixture application, optionally with one injected release fault.
 
The clean project is what a release candidate should look like: pinned dependencies that match the
environment, a rollback lock, no legacy imports, no secrets in the tree, a bounded model client,
one conversation thread per user and one lock per thread.
"""
from importlib.metadata import version
from pathlib import Path
 
PINNED = ["langchain", "langchain-core", "langchain-openai", "langgraph"]
 
APP = '''"""Fixture service: one chat endpoint with a conversation thread per user."""
import contextlib
import os
import threading
from collections import defaultdict
 
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
 
 
def build_llm():
    return ChatOpenAI(model="gpt-test", api_key=os.environ.get("OPENAI_API_KEY", "unset"){llm_kwargs})
 
 
def build_agent(model=None):
    return create_agent(model or build_llm(), tools=[], checkpointer=InMemorySaver())
 
 
class Service:
    def __init__(self, agent):
        self.agent = agent
        self._locks = defaultdict(threading.Lock)
 
    def _config(self, user_id):
        return {{"configurable": {{"thread_id": {thread_id}}}}}
 
    def handle(self, user_id, text):
        guard = {guard}
        with guard:
            result = self.agent.invoke({{"messages": [HumanMessage(text)]}}, self._config(user_id))
        return result["messages"][-1].content
 
    def history(self, user_id):
        state = self.agent.get_state(self._config(user_id))
        return [m.content for m in state.values.get("messages", []) if isinstance(m, HumanMessage)]
'''
 
 
def write_project(root: Path, *, pins=None, previous_lock=True, legacy_import=False, secret=None,
                  env_file=False, ignore_env=True, llm_kwargs=", timeout=20, max_retries=1", shared_thread=False,
                  use_lock=True) -> None:
    root.mkdir(parents=True, exist_ok=True)
    lines = pins if pins is not None else [f"{name}=={version(name)}" for name in PINNED]
    (root / "requirements.txt").write_text("\n".join(lines) + "\n", encoding="utf-8")
    if previous_lock:
        (root / "requirements.previous.txt").write_text("langchain==0.3.27\nlangchain-core==0.3.76\n", encoding="utf-8")
    ignored = [".venv/", ".env", "__pycache__/"] if ignore_env else [".venv/", "__pycache__/"]
    (root / ".gitignore").write_text("\n".join(ignored) + "\n", encoding="utf-8")
    (root / ".env.example").write_text("OPENAI_API_KEY=\n", encoding="utf-8")
    thread_id = '"default"' if shared_thread else "user_id"
    guard = "self._locks[" + thread_id + "]" if use_lock else "contextlib.nullcontext()"
    source = APP.format(llm_kwargs=llm_kwargs, thread_id=thread_id, guard=guard)
    if secret:
        source += f'\nOPENAI_API_KEY = "{secret}"\n'
    (root / "app.py").write_text(source, encoding="utf-8")
    if legacy_import:
        (root / "legacy_module.py").write_text("from langchain.chains import LLMChain\n", encoding="utf-8")
    if env_file:
        (root / ".env").write_text("OPENAI_API_KEY=placeholder\n", encoding="utf-8")

"""A scripted chat model: it answers with every human message it was given, after a short delay.
 
The reply makes conversation state visible: if a reply contains a message the caller never sent,
that message came from somebody else's history. No network and no credentials are involved.
"""
import time
 
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.outputs import ChatGeneration, ChatResult
 
 
class EchoModel(BaseChatModel):
    delay: float = 0.0
 
    @property
    def _llm_type(self) -> str:
        return "echo"
 
    def _generate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult:
        time.sleep(self.delay)
        humans = [m.content for m in messages if isinstance(m, HumanMessage)]
        return ChatResult(generations=[ChatGeneration(message=AIMessage(content="seen:" + ",".join(humans)))])
 
    def bind_tools(self, tools, **kwargs):
        return self

"""Release preflight: seven executable checks over a project directory.
 
usage: python preflight.py <project-dir> [--limit SECONDS]
 
Exit code 0 when every check passes, 1 otherwise. Each check returns (ok, message); the message
names the first offending item so a failing release can be fixed without reading the script.
"""
import ast
import importlib
import importlib.util
import re
import sys
import uuid
from concurrent.futures import ThreadPoolExecutor
from importlib.metadata import PackageNotFoundError, version
from pathlib import Path
 
HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE.parent / "model-call-latency-budget"))  # reuse the worst-case latency model
 
from budget import latency_range  # noqa: E402
from doubles import EchoModel  # noqa: E402
 
PIN = re.compile(r"^([A-Za-z0-9_.\-]+)==([^=\s;]+)$")
SECRET_PATTERNS = {
    "OpenAI-style key": re.compile(r"sk-[A-Za-z0-9_\-]{16,}"),
    "AWS access key id": re.compile(r"AKIA[0-9A-Z]{16}"),
    "private key block": re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
}
 
 
def requirement_lines(path: Path) -> list[str]:
    return [line.strip() for line in path.read_text(encoding="utf-8").splitlines() if line.strip() and not line.startswith("#")]
 
 
def check_pins(root: Path, limit: float):
    for line in requirement_lines(root / "requirements.txt"):
        match = PIN.match(line)
        if not match:
            return False, f"not pinned with '==': {line}"
        name, pinned = match.groups()
        try:
            installed = version(name)
        except PackageNotFoundError:
            return False, f"{name} is pinned but not installed"
        if installed != pinned:
            return False, f"{name}: pinned {pinned}, installed {installed}"
    return True, "all requirements pinned and equal to the installed versions"
 
 
def check_rollback_lock(root: Path, limit: float):
    previous = root / "requirements.previous.txt"
    if not previous.exists():
        return False, "requirements.previous.txt is missing"
    for line in requirement_lines(previous):
        if not PIN.match(line):
            return False, f"rollback lock not pinned: {line}"
    return True, "rollback lock present and fully pinned"
 
 
def resolves(module: str, name: str) -> bool:
    try:
        return hasattr(importlib.import_module(module), name)
    except ImportError:
        return False
 
 
def check_imports_resolve(root: Path, limit: float):
    """Every langchain/langgraph import in the project must resolve in the release environment."""
    for path in sorted(root.rglob("*.py")):
        for node in ast.walk(ast.parse(path.read_text(encoding="utf-8"))):
            if isinstance(node, ast.ImportFrom) and node.module and node.module.split(".")[0].startswith(("langchain", "langgraph")):
                for alias in node.names:
                    if not resolves(node.module, alias.name):
                        return False, f"{path.name}:{node.lineno} cannot import {alias.name} from {node.module}"
    return True, "every langchain/langgraph import resolves in this environment"
 
 
def check_secrets(root: Path, limit: float):
    gitignore = root / ".gitignore"
    if not gitignore.exists() or ".env" not in gitignore.read_text(encoding="utf-8").split():
        return False, ".gitignore does not list .env"
    if (root / ".env").exists():
        return False, ".env is present in the release tree"
    for path in sorted(root.rglob("*")):
        if path.is_file() and path.suffix in {".py", ".txt", ".md", ".yaml", ".yml", ".json", ".toml", ".example"}:
            text = path.read_text(encoding="utf-8", errors="ignore")
            for label, pattern in SECRET_PATTERNS.items():
                if pattern.search(text):
                    return False, f"{label} found in {path.name}"
    return True, ".env ignored and absent; no key-like strings in source files"
 
 
def load_app(root: Path):
    spec = importlib.util.spec_from_file_location(f"fixture_app_{uuid.uuid4().hex}", root / "app.py")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module
 
 
def check_call_budget(root: Path, limit: float):
    llm = load_app(root).build_llm()
    timeout = llm.request_timeout
    if timeout is None:
        return False, "no timeout: the worst case of one failing call is unbounded"
    per_attempt = max(timeout) if isinstance(timeout, tuple) else float(timeout)
    retries = 2 if llm.max_retries is None else llm.max_retries  # None means the SDK default of 2
    _, worst = latency_range(per_attempt, retries)
    if worst > limit:
        return False, f"worst case {worst:.1f}s exceeds the {limit:.0f}s limit (timeout={per_attempt:g}, max_retries={retries})"
    return True, f"worst case {worst:.1f}s within the {limit:.0f}s limit (timeout={per_attempt:g}, max_retries={retries})"
 
 
def check_session_isolation(root: Path, limit: float):
    app = load_app(root)
    service = app.Service(app.build_agent(EchoModel()))
    for user in ("u0", "u1", "u2"):
        reply = service.handle(user, f"tok-{user}")
        if reply != f"seen:tok-{user}":
            return False, f"{user} received a reply built from other messages: {reply}"
    return True, "three users, three separate conversation histories"
 
 
def check_same_session_serialization(root: Path, limit: float):
    app = load_app(root)
    service = app.Service(app.build_agent(EchoModel(delay=0.05)))
    with ThreadPoolExecutor(5) as pool:
        list(pool.map(lambda i: service.handle("alice", f"m{i}"), range(5)))
    kept = len(service.history("alice"))
    if kept != 5:
        return False, f"5 concurrent messages to one thread, only {kept} kept in its history"
    return True, "5 concurrent messages to one thread, all 5 kept"
 
 
CHECKS = {
    "pins": check_pins,
    "rollback_lock": check_rollback_lock,
    "imports_resolve": check_imports_resolve,
    "secrets": check_secrets,
    "call_budget": check_call_budget,
    "session_isolation": check_session_isolation,
    "same_session_serialization": check_same_session_serialization,
}
 
 
def run_checks(root: Path, limit: float = 60.0) -> dict[str, tuple[bool, str]]:
    return {name: check(root, limit) for name, check in CHECKS.items()}
 
 
def main(argv: list[str]) -> int:
    if not argv or argv[0].startswith("--"):
        print(__doc__)
        return 2
    limit = float(argv[argv.index("--limit") + 1]) if "--limit" in argv else 60.0
    results = run_checks(Path(argv[0]), limit)
    for name, (ok, message) in results.items():
        print(f"{'PASS' if ok else 'FAIL'} {name}: {message}")
    return 0 if all(ok for ok, _ in results.values()) else 1
 
 
if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))

"""Run the preflight CLI entry point on a clean project and on a project with two faults."""
import tempfile
from importlib.metadata import version
from pathlib import Path
 
from preflight import main
from project_factory import PINNED, write_project
 
pins = [f"{name}=={version(name)}" for name in PINNED]
cases = [
    ("clean release candidate", {}),
    ("release candidate with two faults", {"pins": ["langchain>=1.0"] + pins[1:], "llm_kwargs": ""}),
]
for label, faults in cases:
    with tempfile.TemporaryDirectory() as tmp:
        write_project(Path(tmp), **faults)
        print(f"$ python preflight.py <{label}>")
        code = main([tmp])
        print(f"exit code: {code}\n")

$ python preflight.py <clean release candidate>
PASS pins: all requirements pinned and equal to the installed versions
PASS rollback_lock: rollback lock present and fully pinned
PASS imports_resolve: every langchain/langgraph import resolves in this environment
PASS secrets: .env ignored and absent; no key-like strings in source files
PASS call_budget: worst case 40.5s within the 60s limit (timeout=20, max_retries=1)
PASS session_isolation: three users, three separate conversation histories
PASS same_session_serialization: 5 concurrent messages to one thread, all 5 kept
exit code: 0

$ python preflight.py <release candidate with two faults>
FAIL pins: not pinned with '==': langchain>=1.0
PASS rollback_lock: rollback lock present and fully pinned
PASS imports_resolve: every langchain/langgraph import resolves in this environment
PASS secrets: .env ignored and absent; no key-like strings in source files
FAIL call_budget: no timeout: the worst case of one failing call is unbounded
PASS session_isolation: three users, three separate conversation histories
PASS same_session_serialization: 5 concurrent messages to one thread, all 5 kept
exit code: 1

"""Check the checks: inject one release fault at a time and see which preflight check reacts.
 
A preflight that has never failed proves nothing. Each fault below must be caught by exactly the
check named next to it, and the clean project must pass everything.
"""
import sys
import tempfile
from importlib.metadata import version
from pathlib import Path
 
from preflight import CHECKS, run_checks
from project_factory import PINNED, write_project
 
CLEAN = [f"{name}=={version(name)}" for name in PINNED]
UNPINNED = ["langchain>=1.0"] + CLEAN[1:]
DRIFTED = [CLEAN[0], "langchain-core==0.3.0"] + CLEAN[2:]
 
FAULTS = [
    ("unpinned dependency", {"pins": UNPINNED}, "pins"),
    ("pin differs from installed version", {"pins": DRIFTED}, "pins"),
    ("no rollback lock", {"previous_lock": False}, "rollback_lock"),
    ("legacy import langchain.chains", {"legacy_import": True}, "imports_resolve"),
    ("hard-coded API key in app.py", {"secret": "sk-test-0123456789abcdef"}, "secrets"),
    (".env left in the release tree", {"env_file": True}, "secrets"),
    (".gitignore does not list .env", {"ignore_env": False}, "secrets"),
    ("model client without timeout", {"llm_kwargs": ""}, "call_budget"),
    ("timeout=60 with default retries", {"llm_kwargs": ", timeout=60"}, "call_budget"),
    ("one shared thread_id for all users", {"shared_thread": True}, "session_isolation"),
    ("no per-thread lock", {"use_lock": False}, "same_session_serialization"),
]
 
 
def evaluate(**faults) -> dict:
    with tempfile.TemporaryDirectory() as tmp:
        write_project(Path(tmp), **faults)
        return run_checks(Path(tmp))
 
 
def main() -> int:
    mismatches = 0
    results = evaluate()
    clean_ok = all(ok for ok, _ in results.values())
    print(f"{'clean project':<38} -> {'all ' + str(len(CHECKS)) + ' checks pass' if clean_ok else 'UNEXPECTED FAILURE'}")
    mismatches += 0 if clean_ok else 1
    for label, faults, expected in FAULTS:
        results = evaluate(**faults)
        failed = [name for name, (ok, _) in results.items() if not ok]
        verdict = "" if failed == [expected] else f"   <-- expected only {expected}"
        mismatches += 1 if failed != [expected] else 0
        detail = "; ".join(f"FAIL {name}: {results[name][1]}" for name in failed) or "no check failed"
        print(f"{label:<38} -> {detail}{verdict}")
    print(f"faults injected: {len(FAULTS)}, caught by exactly the expected check: {len(FAULTS) - mismatches}")
    return 1 if mismatches else 0
 
 
if __name__ == "__main__":
    sys.exit(main())

clean project                          -> all 7 checks pass
unpinned dependency                    -> FAIL pins: not pinned with '==': langchain>=1.0
pin differs from installed version     -> FAIL pins: langchain-core: pinned 0.3.0, installed 1.7.0
no rollback lock                       -> FAIL rollback_lock: requirements.previous.txt is missing
legacy import langchain.chains         -> FAIL imports_resolve: legacy_module.py:1 cannot import LLMChain from langchain.chains
hard-coded API key in app.py           -> FAIL secrets: OpenAI-style key found in app.py
.env left in the release tree          -> FAIL secrets: .env is present in the release tree
.gitignore does not list .env          -> FAIL secrets: .gitignore does not list .env
model client without timeout           -> FAIL call_budget: no timeout: the worst case of one failing call is unbounded
timeout=60 with default retries        -> FAIL call_budget: worst case 181.5s exceeds the 60s limit (timeout=60, max_retries=2)
one shared thread_id for all users     -> FAIL session_isolation: u1 received a reply built from other messages: seen:tok-u0,tok-u1
no per-thread lock                     -> FAIL same_session_serialization: 5 concurrent messages to one thread, only 1 kept in its history
faults injected: 11, caught by exactly the expected check: 11

"""Conversation state under concurrency: checkpointers, thread_id and the same-thread race."""
import sqlite3
import tempfile
import threading
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
 
from doubles import EchoModel
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
 
 
def distinct_users(checkpointer, users: int = 20):
    """Each user has its own thread_id. Returns (replies built from foreign messages, error classes)."""
    agent = create_agent(EchoModel(delay=0.05), tools=[], checkpointer=checkpointer)
 
    def one(i: int):
        try:
            reply = agent.invoke({"messages": [HumanMessage(f"tok{i}")]}, {"configurable": {"thread_id": f"user-{i}"}})
            return None if reply["messages"][-1].content == f"seen:tok{i}" else "leak"
        except Exception as exc:
            return type(exc).__name__
 
    with ThreadPoolExecutor(users) as pool:
        outcomes = list(pool.map(one, range(users)))
    return outcomes.count("leak"), sorted({o for o in outcomes if o and o != "leak"}), sum(o not in (None, "leak") for o in outcomes)
 
 
def same_thread(checkpointer, locked: bool, calls: int = 5) -> int:
    """Send `calls` messages to one thread at the same time; return how many human messages survive."""
    agent = create_agent(EchoModel(delay=0.05), tools=[], checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "alice"}}
    lock = threading.Lock()
 
    def one(i: int) -> None:
        if locked:
            with lock:
                agent.invoke({"messages": [HumanMessage(f"m{i}")]}, config)
        else:
            agent.invoke({"messages": [HumanMessage(f"m{i}")]}, config)
 
    with ThreadPoolExecutor(calls) as pool:
        list(pool.map(one, range(calls)))
    messages = agent.get_state(config).values["messages"]
    return sum(isinstance(m, HumanMessage) for m in messages)
 
 
tmp = Path(tempfile.mkdtemp())
print("20 users at once, one thread_id each")
leaks, kinds, errors = distinct_users(InMemorySaver())
print(f"  InMemorySaver                                  leaked replies: {leaks}, failed calls: {errors}")
leaks, kinds, errors = distinct_users(SqliteSaver(sqlite3.connect(tmp / "default.db")))
print(f"  SqliteSaver, default connection                leaked replies: {leaks}, failed calls: {errors} {kinds}")
leaks, kinds, errors = distinct_users(SqliteSaver(sqlite3.connect(tmp / "shared.db", check_same_thread=False)))
print(f"  SqliteSaver, check_same_thread=False           leaked replies: {leaks}, failed calls: {errors}")
 
print("5 messages to ONE thread_id at the same time")
kept = same_thread(InMemorySaver(), locked=False)
print(f"  InMemorySaver, no lock                         human messages kept: {kept} of 5")
kept = same_thread(SqliteSaver(sqlite3.connect(tmp / "race.db", check_same_thread=False)), locked=False)
print(f"  SqliteSaver, no lock                           human messages kept: {kept} of 5")
kept = same_thread(InMemorySaver(), locked=True)
print(f"  InMemorySaver, one lock per thread             human messages kept: {kept} of 5")

20 users at once, one thread_id each
  InMemorySaver                                  leaked replies: 0, failed calls: 0
  SqliteSaver, default connection                leaked replies: 0, failed calls: 20 ['ProgrammingError']
  SqliteSaver, check_same_thread=False           leaked replies: 0, failed calls: 0
5 messages to ONE thread_id at the same time
  InMemorySaver, no lock                         human messages kept: 1 of 5
  SqliteSaver, no lock                           human messages kept: 1 of 5
  InMemorySaver, one lock per thread             human messages kept: 5 of 5