第 5 章 从原型到生产：契约、工具与可靠执行

Deep Agents

%pip -q install -U deepagents langchain langchain-core langchain-openai tavily-python langgraph

import os
from dotenv import load_dotenv


load_dotenv()

OPENROUTER_API_KEY = os.getenv('OPENROUTER_API_KEY')
TAVILY_API_KEY = os.getenv('TAVILY_API_KEY')
EXA_API_KEY = os.getenv('EXA_API_KEY')

!git clone https://github.com/langchain-ai/deepagents.git

%cd /content/deepagents/examples/content-builder-agent

/content/deepagents/examples/content-builder-agent

%ls

!python content_writer.py "Create a LinkedIn post about AI agents in recruiting in 2026 and how this can be used to automate job writing"

用 LangGraph 集成 MCP

用 LangGraph 集成 MCP：通过 Model Context Protocol 扩展 Agents

挑战： 你想让你的 LangGraph agents 访问外部工具和资源——网页搜索、数据库、API——但直接在 agent 代码中管理这些集成会造成紧耦合和维护上的麻烦。每一个新能力都需要改动代码，而且不同工具有不同的接口。

解决方案： Model Context Protocol（MCP）提供了一种标准化方式，把工具和资源暴露给 AI 应用。借助 langchain-mcp-adapters，你可以把 MCP servers 无缝集成到 LangGraph agents 中，让它们获得强大的外部能力，而无需让 agent 代码变得杂乱。

这个 notebook 演示：

1. MCP 原理： 什么是 MCP，为什么它很重要
2. 简单 Server： 构建一个基本的文件管理器 MCP server
3. LangGraph 集成： 在 StateGraph agents 中使用 MCP tools
4. 生产级示例： 带类型化 IO、安全性和治理能力的 SQLite 库存管理器
5. 第三方集成： 使用外部 MCP servers 的快速示例

最佳实践

1. 传输方式选择

• stdio：最适合本地开发、单用户场景
• HTTP/Streamable HTTP：更适合 Web 服务器、多用户场景

2. 并发安全

• 并发 tool calls 使用每次调用独立的连接（而非全局连接）
• 启用 SQLite WAL 模式以获得更好的并发能力
• 使用原子更新来防止竞态条件

3. 治理

• 在 MCP server 边界强制执行策略，而不是在模型内部
• MCP servers 是多 agent 系统的治理层

4. 类型化 IO

• 输入和输出都使用 Pydantic models，以确保明确的契约
• 这可以防止静默失败，并让 tool 行为可预测

安装

首先，安装所需的包：

# Install required packages
%pip install langchain-mcp-adapters langgraph langchain-core langchain-openai mcp fastmcp pydantic

理解 MCP

Model Context Protocol（MCP） 是一个开放协议，标准化了 AI 应用与外部工具和资源交互的方式。可以把它想象成一个通用适配器，让你的 agents 无需了解实现细节即可访问各种能力。

核心概念

• MCP Servers： 提供工具和资源（例如文件操作、数据库访问、网页搜索）
• MCP Clients： 连接服务器并把它们的能力暴露给 AI 应用
• 传输方式（Transports）： 通信方法（stdio、HTTP、SSE、streamable HTTP）
• 工具（Tools）： agent 可以调用的函数（例如 read_file、search_web）

为什么 MCP 很重要

• 解耦： 工具位于独立的服务器中，而不是在 agent 代码里
• 可复用： 一个 MCP server 可以为多个 agents 服务
• 标准化： 在不同工具之间提供一致的接口
• 治理： MCP servers 在模型外部强制执行策略和不变量

简单的 MCP Server - 文件管理器

让我们从简单的文件管理器 MCP server 开始，理解基础知识。它演示了核心模式：用 Pydantic models 定义工具，并通过 FastMCP 暴露它们。这个简单 server 演示：

• 使用 Pydantic models 的类型化 IO
• 用 FastMCP 定义基本工具
• 标准的 MCP server 模式

# Create a simple file manager MCP server
file_manager_code = '''
"""
Simple File Manager MCP Server

A minimal example showing:
- Typed IO with Pydantic models
- Basic file operations
- FastMCP server setup
"""

from pathlib import Path
from typing import Optional
from pydantic import BaseModel, Field
from mcp.server.fastmcp import FastMCP

# Pydantic models for typed IO
class ReadFileRequest(BaseModel):
    """Read a file."""
    file_path: str = Field(..., description="Path to file to read")

class WriteFileRequest(BaseModel):
    """Write content to a file."""
    file_path: str = Field(..., description="Path to file to write")
    content: str = Field(..., description="Content to write")

class ListFilesRequest(BaseModel):
    """List files in a directory."""
    directory: str = Field(..., description="Directory path")
    pattern: Optional[str] = Field(None, description="Optional glob pattern (e.g., '*.py')")

# Create MCP server
mcp = FastMCP("FileManager")

@mcp.tool()
def read_file(request: ReadFileRequest) -> str:
    """Read the contents of a file."""
    path = Path(request.file_path)
    if not path.exists():
        raise FileNotFoundError(f"File not found: {request.file_path}")
    return path.read_text(encoding="utf-8")

@mcp.tool()
def write_file(request: WriteFileRequest) -> dict:
    """Write content to a file. Creates file if it doesn't exist."""
    path = Path(request.file_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(request.content, encoding="utf-8")
    return {"status": "success", "file_path": str(path), "bytes_written": len(request.content)}

@mcp.tool()
def list_files(request: ListFilesRequest) -> list[str]:
    """List files in a directory."""
    dir_path = Path(request.directory)
    if not dir_path.exists():
        raise FileNotFoundError(f"Directory not found: {request.directory}")

    if request.pattern:
        files = list(dir_path.glob(request.pattern))
    else:
        files = list(dir_path.iterdir())

    return [str(f.relative_to(dir_path)) for f in files if f.is_file()]

if __name__ == "__main__":
    mcp.run(transport="stdio")
'''

# Write the server file
from pathlib import Path
file_manager_path = Path("file_manager_server.py")
file_manager_path.write_text(file_manager_code)
print(f"Created {file_manager_path}")

Created file_manager_server.py

连接 MCP Server 并与 LangGraph 一起使用

现在让我们连接 MCP server，并将其与 LangGraph 集成。这正是 MCP 的真正强大之处——你的 agent 可以在不了解实现细节的情况下使用外部工具。

设置你的 API 密钥（或使用环境变量）

import os
from dotenv import load_dotenv


load_dotenv()

OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
EXA_API_KEY = os.environ.get("EXA_API_KEY", "")

给 OutStream.fileno 打 Monkey patch

这能让 Jupyter 的打印保持正常，只是让 fileno() 在 Jupyter Notebooks 中不再抛错。注意：在普通的 Python 脚本中不需要这一步。

from ipykernel.iostream import OutStream

def _safe_fileno(self):
    name = getattr(self, "name", "")
    if "stderr" in name:
        return 2
    return 1

OutStream.fileno = _safe_fileno

from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition
from langchain.chat_models import init_chat_model

# Initialize model
try:
    model = init_chat_model("openrouter:anthropic/claude-3.5-haiku")
except:
    model = init_chat_model("openai:gpt-4o-mini")

# Create MCP client and connect to file manager server
client = MultiServerMCPClient(
    {
        "file_manager": {
            "command": "python",
            "args": [os.path.abspath("file_manager_server.py")],
            "transport": "stdio",
        }
    }
)


# Load tools from MCP server
tools = await client.get_tools()

print(f" Loaded {len(tools)} tools from MCP server:")
for tool in tools:
    print(f"  - {tool.name}: {tool.description}")

# Define the agent node
def call_model(state: MessagesState):
    """Call the model with tools bound."""
    response = model.bind_tools(tools).invoke(state["messages"])
    return {"messages": [response]}

# Build the graph
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "call_model")
builder.add_conditional_edges("call_model", tools_condition)
builder.add_edge("tools", "call_model")

graph = builder.compile()
print("\n LangGraph agent created with MCP tools")

Loaded 3 tools from MCP server:
  - read_file: Read the contents of a file.
  - write_file: Write content to a file. Creates file if it doesn't exist.
  - list_files: List files in a directory.

 LangGraph agent created with MCP tools

用文件操作测试 agent

response = await graph.ainvoke({
    "messages": [("user", "Create a file called 'test.txt' with content 'Hello from MCP!', then read it back.")]
})

# Display the response
from langchain_core.messages import AIMessage
for message in response["messages"]:
    if isinstance(message, AIMessage):
        print("Agent Response:")
        print(message.content)
        if message.tool_calls:
            print("\nTool Calls:")
            for tool_call in message.tool_calls:
                print(f"  - {tool_call['name']}({tool_call['args']})")

Agent Response:


Tool Calls:
  - write_file({'request': {'file_path': 'test.txt', 'content': 'Hello from MCP!'}})
  - read_file({'request': {'file_path': 'test.txt'}})
Agent Response:
The file 'test.txt' has been created with the content 'Hello from MCP!'. When read back, the content is: **Hello from MCP!**

生产级示例 - 库存管理器

现在让我们看一个生产就绪的示例：一个演示真实世界模式的 SQLite 库存管理器。它展示了 MCP servers 如何提供持久化状态、安全性保证和策略执行。

这个库存管理器演示了企业级模式：

• 部门拥有系统 → MCP servers 暴露它们
• Agent 系统消费 → 稳定、类型化的 tool 契约
• MCP = 互操作层 → 多个 MAS 无需共享代码即可集成
• 边界处的治理 → 策略在 server 中执行，而不是在模型内

# Create a production-ready inventory manager MCP server
inventory_server_code = '''
"""
SQLite Inventory Manager MCP Server

Production patterns:
- Typed IO with Pydantic models (input and output)
- Safety: Atomic updates prevent negative stock
- Concurrency: Per-call connections, SQLite WAL mode
- Governance: Policy gates enforced at MCP boundary
"""

import sqlite3
import json
from pathlib import Path
from typing import Optional, List
from contextlib import contextmanager
from pydantic import BaseModel, Field, field_validator
from mcp.server.fastmcp import FastMCP

# ============================================================================
# Pydantic Models
# ============================================================================

class ItemCreate(BaseModel):
    """Create a new inventory item."""
    name: str = Field(..., min_length=1, max_length=200)
    initial_stock: int = Field(..., ge=0)
    unit_price: float = Field(..., ge=0)
    category: Optional[str] = None

class StockAdjustment(BaseModel):
    """Adjust stock quantity."""
    item_id: int = Field(..., gt=0)
    quantity: int = Field(..., description="Positive adds, negative removes")
    reason: str = Field(..., min_length=1)
    approval_token: Optional[str] = None

    @field_validator("quantity")
    @classmethod
    def quantity_not_zero(cls, v):
        if v == 0:
            raise ValueError("Quantity must be non-zero")
        return v

class ItemQuery(BaseModel):
    """Query items."""
    item_id: Optional[int] = None
    category: Optional[str] = None
    low_stock_threshold: Optional[int] = None

class ItemOut(BaseModel):
    """Item output model."""
    id: int
    name: str
    stock: int
    unit_price: float
    category: Optional[str]

# ============================================================================
# Database Setup
# ============================================================================

DB_PATH = Path("inventory.db")

def init_database():
    """Initialize database schema."""
    conn = sqlite3.connect(str(DB_PATH), timeout=10)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA synchronous=NORMAL")
    conn.execute("PRAGMA foreign_keys=ON")

    conn.execute("""
        CREATE TABLE IF NOT EXISTS items (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            stock INTEGER NOT NULL DEFAULT 0 CHECK(stock >= 0),
            unit_price REAL NOT NULL CHECK(unit_price >= 0),
            category TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)

    conn.execute("""
        CREATE TABLE IF NOT EXISTS audit_log (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            item_id INTEGER,
            event_type TEXT NOT NULL,
            event_data TEXT NOT NULL,
            timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (item_id) REFERENCES items(id)
        )
    """)

    conn.commit()
    conn.close()

init_database()

def connect():
    """Create a new connection for each tool call (safe for concurrency)."""
    conn = sqlite3.connect(str(DB_PATH), timeout=10)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA synchronous=NORMAL")
    conn.execute("PRAGMA foreign_keys=ON")
    return conn

@contextmanager
def transaction():
    """Context manager for transactions with per-call connections."""
    conn = connect()
    try:
        conn.execute("BEGIN")
        yield conn
        conn.execute("COMMIT")
    except Exception:
        conn.execute("ROLLBACK")
        raise
    finally:
        conn.close()

def log_event(conn, item_id: Optional[int], event_type: str, event_data: dict):
    """Append event to audit log using active connection."""
    conn.execute(
        "INSERT INTO audit_log (item_id, event_type, event_data) VALUES (?, ?, ?)",
        (item_id, event_type, json.dumps(event_data))
    )

# ============================================================================
# Policy Enforcement
# ============================================================================

def enforce_policy(adjustment: StockAdjustment):
    """Enforce business policies - MCP servers are the governance boundary."""
    # Policy: Large reductions require approval
    if adjustment.quantity < -100:
        if not adjustment.approval_token or adjustment.approval_token != "APPROVED_BY_MANAGER":
            raise ValueError(
                f"Large stock reduction ({abs(adjustment.quantity)} units) requires approval_token."
            )

    # Policy: Negative adjustments must reference an order
    if adjustment.quantity < 0:
        if not adjustment.reason.startswith("ORDER:"):
            raise ValueError(
                "Negative adjustments must include order reference. Format: 'ORDER:  - '"
            )

# ============================================================================
# MCP Server
# ============================================================================

mcp = FastMCP("InventoryManager")

@mcp.tool()
def create_item(item: ItemCreate) -> ItemOut:
    """Create a new inventory item."""
    with transaction() as conn:
        cursor = conn.execute(
            "INSERT INTO items (name, stock, unit_price, category) VALUES (?, ?, ?, ?)",
            (item.name, item.initial_stock, item.unit_price, item.category)
        )
        item_id = cursor.lastrowid

        log_event(conn, item_id, "item_created", {
            "name": item.name,
            "initial_stock": item.initial_stock
        })

        row = conn.execute(
            "SELECT id, name, stock, unit_price, category FROM items WHERE id = ?",
            (item_id,)
        ).fetchone()

        return ItemOut(
            id=row["id"],
            name=row["name"],
            stock=row["stock"],
            unit_price=row["unit_price"],
            category=row["category"]
        )

@mcp.tool()
def adjust_stock(adjustment: StockAdjustment) -> ItemOut:
    """Adjust stock quantity. Enforces no negative stock and policy gates."""
    # Enforce policy gates (governance boundary)
    enforce_policy(adjustment)

    with transaction() as conn:
        # Atomic update: only succeeds if stock won't go negative
        cur = conn.execute(
            "UPDATE items SET stock = stock + ? WHERE id = ? AND stock + ? >= 0",
            (adjustment.quantity, adjustment.item_id, adjustment.quantity)
        )

        if cur.rowcount == 0:
            exists = conn.execute("SELECT 1 FROM items WHERE id = ?", (adjustment.item_id,)).fetchone()
            if not exists:
                raise ValueError(f"Item {adjustment.item_id} not found")

            current = conn.execute("SELECT stock FROM items WHERE id = ?", (adjustment.item_id,)).fetchone()
            current_stock = current["stock"] if current else 0
            raise ValueError(
                f"Insufficient stock. Current: {current_stock}, requested: {adjustment.quantity}"
            )

        updated = conn.execute("SELECT stock FROM items WHERE id = ?", (adjustment.item_id,)).fetchone()
        new_stock = updated["stock"]

        log_event(conn, adjustment.item_id, "stock_adjusted", {
            "adjustment": adjustment.quantity,
            "new_stock": new_stock,
            "reason": adjustment.reason
        })

        row = conn.execute(
            "SELECT id, name, stock, unit_price, category FROM items WHERE id = ?",
            (adjustment.item_id,)
        ).fetchone()

        return ItemOut(
            id=row["id"],
            name=row["name"],
            stock=row["stock"],
            unit_price=row["unit_price"],
            category=row["category"]
        )

@mcp.tool()
def list_items(query: ItemQuery) -> List[ItemOut]:
    """Query inventory items."""
    conn = connect()
    try:
        conditions = []
        params = []

        if query.item_id:
            conditions.append("id = ?")
            params.append(query.item_id)
        if query.category:
            conditions.append("category = ?")
            params.append(query.category)
        if query.low_stock_threshold is not None:
            conditions.append("stock <= ?")
            params.append(query.low_stock_threshold)

        where = "WHERE " + " AND ".join(conditions) if conditions else ""

        rows = conn.execute(
            f"SELECT id, name, stock, unit_price, category FROM items {where} ORDER BY name",
            params
        ).fetchall()

        return [
            ItemOut(
                id=row["id"],
                name=row["name"],
                stock=row["stock"],
                unit_price=row["unit_price"],
                category=row["category"]
            )
            for row in rows
        ]
    finally:
        conn.close()

if __name__ == "__main__":
    mcp.run(transport="stdio")
'''

# Write the server file
inventory_path = Path("inventory_server.py")
inventory_path.write_text(inventory_server_code)
print(f"Created {inventory_path}")
print("\nKey patterns demonstrated:")
print("  • Typed IO: Pydantic models for contracts")
print("  • Safety: Atomic updates prevent race conditions")
print("  • Concurrency: Per-call connections, WAL mode")
print("  • Governance: Policy gates enforced at server boundary")

Created inventory_server.py

Key patterns demonstrated:
  • Typed IO: Pydantic models for contracts
  • Safety: Atomic updates prevent race conditions
  • Concurrency: Per-call connections, WAL mode
  • Governance: Policy gates enforced at server boundary

# Create agent with inventory manager
inventory_client = MultiServerMCPClient(
    {
        "inventory": {
            "command": "python",
            "args": [os.path.abspath("inventory_server.py")],
            "transport": "stdio",
        }
    }
)

inventory_tools = await inventory_client.get_tools()

def call_model_inventory(state: MessagesState):
    response = model.bind_tools(inventory_tools).invoke(state["messages"])
    return {"messages": [response]}

builder_inv = StateGraph(MessagesState)
builder_inv.add_node("call_model", call_model_inventory)
builder_inv.add_node("tools", ToolNode(inventory_tools))
builder_inv.add_edge(START, "call_model")
builder_inv.add_conditional_edges("call_model", tools_condition)
builder_inv.add_edge("tools", "call_model")

inventory_graph = builder_inv.compile()

# Demo: Create item, adjust stock, test safety and policy gates
print("=== Inventory Manager Demo ===\n")

# 1. Create an item
print("1. Creating item...")
response1 = await inventory_graph.ainvoke({
    "messages": [("user", "Create a new item: 'Laptop Pro' with initial stock 10, price $1299.99, category 'Electronics'")]
})
print("   ✅ Item created\n")

# 2. Adjust stock (positive)
print("2. Adding stock...")
response2 = await inventory_graph.ainvoke({
    "messages": [("user", "Add 5 units to item ID 1, reason: 'Received shipment'")]
})
print("   ✅ Stock increased\n")

# 3. Test safety: Try to remove more than available
print("3. Testing safety invariant (prevent negative stock)...")
try:
    await inventory_graph.ainvoke({
        "messages": [("user", "Remove 1000 units from item ID 1, reason: 'ORDER: TEST-001 - Test order'")]
    })
    print("   ⚠️  Safety check failed!")
except Exception as e:
    print(f"   ✅ Safety check passed: {str(e)[:60]}...\n")

# 4. Test policy: Large reduction without approval
print("4. Testing policy gate (large reduction requires approval)...")
try:
    await inventory_graph.ainvoke({
        "messages": [("user", "Reduce stock by 150 units for item ID 1, reason: 'ORDER: BULK-001 - Bulk order'")]
    })
    print("   ⚠️  Policy check failed!")
except Exception as e:
    print(f"   ✅ Policy gate enforced: {str(e)[:60]}...\n")

print("=== Key Insights ===")
print("  • MCP servers enforce safety invariants (no negative stock)")
print("  • Policy gates are enforced at the server boundary, not in the model")
print("  • Multiple agents can safely share the same database through MCP")

=== Inventory Manager Demo ===

1. Creating item...
   ✅ Item created

2. Adding stock...
   ✅ Stock increased

3. Testing safety invariant (prevent negative stock)...
   ✅ Safety check passed: Error executing tool adjust_stock: Large stock reduction (10...

4. Testing policy gate (large reduction requires approval)...
   ✅ Policy gate enforced: Error executing tool adjust_stock: Large stock reduction (15...

=== Key Insights ===
  • MCP servers enforce safety invariants (no negative stock)
  • Policy gates are enforced at the server boundary, not in the model
  • Multiple agents can safely share the same database through MCP

第三方 MCP Servers

许多服务都提供 MCP servers 以便集成。这里有一个使用 Exa 的 MCP server 进行网页搜索的快速示例：

# Example: Combine inventory manager with Exa search
EXA_MCP_URL = "https://mcp.exa.ai/mcp"


if EXA_API_KEY:
    # Combine multiple MCP servers
    combined_client = MultiServerMCPClient(
        {
            "inventory": {
                "command": "python",
                "args": [os.path.abspath("inventory_server.py")],
                "transport": "stdio",
            },
            "exa": {
                "transport": "http",
                "url": EXA_MCP_URL,
                "headers": {"Authorization": f"Bearer {EXA_API_KEY}"},
            }
        }
    )

    combined_tools = await combined_client.get_tools()
    print(f"✅ Loaded {len(combined_tools)} tools from multiple MCP servers")
    print("   You can now use both inventory management and web search in the same agent")
else:
    print("⚠️  Set EXA_API_KEY to test third-party MCP integration")
    print("   Example: os.environ['EXA_API_KEY'] = 'your-key-here'")

✅ Loaded 5 tools from multiple MCP servers
   You can now use both inventory management and web search in the same agent

产品可靠性定律

关于本 notebook

这个 notebook 把多 agent 系统背后的可靠性数学讲清楚了。它实现了两个简单模型。第一个遵循产品可靠性定律，在系统工程中正式化为 Lusser's law，它展示了当概率组件在没有校验的情况下顺序组合时，系统准确率是如何衰减的。应用到多 agent 流水线上，它给你：

$$ P(\text { system success })=\prod_{i=1}^N p_i $$

第二个模型在这个基线上做了扩展，在每个 agent 边界引入校验，说明在错误传播之前就捕获它们会如何从根本上改变系统级的可靠性。目标不是覆盖所有真实的失败模式，而是让你直观地理解为什么多 agent 系统默认是脆弱的，以及架构选择如何直接影响到可靠性和成本。

import pandas as pd
from typing import List


def reliability_df_product_rule(
    p: float,
    max_agents: int
) -> pd.DataFrame:
    """
    Generate a DataFrame showing system reliability decay
    under the product reliability rule.
    """
    data = []
    for n in range(1, max_agents + 1):
        system_p = p ** n
        data.append({
            "num_agents": n,
            "per_agent_accuracy": p,
            "system_accuracy": system_p,
            "error_rate": 1.0 - system_p
        })
    return pd.DataFrame(data)


def reliability_df_with_validation(
    p: float,
    validation_catch_rate: float,
    max_agents: int
) -> pd.DataFrame:
    """
    Generate a DataFrame showing system reliability
    with validation at each agent boundary.
    """
    p_eff = p + (1.0 - p) * validation_catch_rate

    data = []
    for n in range(1, max_agents + 1):
        system_p = p_eff ** n
        data.append({
            "num_agents": n,
            "per_agent_accuracy": p_eff,
            "system_accuracy": system_p,
            "error_rate": 1.0 - system_p
        })
    return pd.DataFrame(data)

p = 0.98
v = 0.9
max_agents = 10

df_product = reliability_df_product_rule(p, max_agents)
df_validation = reliability_df_with_validation(p, v, max_agents)

df_product

num_agents  per_agent_accuracy  system_accuracy  error_rate
0           1                0.98         0.980000    0.020000
1           2                0.98         0.960400    0.039600
2           3                0.98         0.941192    0.058808
3           4                0.98         0.922368    0.077632
4           5                0.98         0.903921    0.096079
5           6                0.98         0.885842    0.114158
6           7                0.98         0.868126    0.131874
7           8                0.98         0.850763    0.149237
8           9                0.98         0.833748    0.166252
9          10                0.98         0.817073    0.182927

df_validation

num_agents  per_agent_accuracy  system_accuracy  error_rate
0           1               0.998         0.998000    0.002000
1           2               0.998         0.996004    0.003996
2           3               0.998         0.994012    0.005988
3           4               0.998         0.992024    0.007976
4           5               0.998         0.990040    0.009960
5           6               0.998         0.988060    0.011940
6           7               0.998         0.986084    0.013916
7           8               0.998         0.984112    0.015888
8           9               0.998         0.982143    0.017857
9          10               0.998         0.980179    0.019821

用 Pydantic 保证 Agent 一致性

关于本 notebook

这个 notebook 为构建生产就绪的 AI agents 提供了一个蓝图。它解决了"在我的机器上能用"和"在大规模下可靠运行"之间的关键差距。

问题： agents 在生产环境中经常静默失败。当 LLM 输出稍有变化、JSON 解析出错，或者不同的服务提供商返回不一致的类型（字符串而不是整数）时，就会导致数小时的脆弱代码调试。

解决方案： 通过把 Pydantic models 用作技术契约，我们在系统边界定义了数据的精确形态。这可以防止无效状态传播、简化服务提供商切换，并确保系统具备自愈能力。

---

生产环境的技术支柱

1. 确定性策略管理

要保证一次运行从头到尾都是稳定的，校验策略必须在开始时就被固定下来。

• 策略哈希（Policy Hashing）： 活动配置被哈希并存储在运行状态中。
• 一致性： 即使全局配置在部署中途发生变化，正在进行的运行也遵循它们初始化时的规则。
• 审计追踪（Audit Trails）： 每条数据库记录都包含一个 policy_hash，让开发者能精确看到是哪一套校验规则生成了那条具体数据。

2. 性能优化的校验

校验不应成为瓶颈。

• 模型缓存： Pydantic models 按 (mode, policy_hash) 组合构建一次并缓存，避免在"热路径"上重复生成类。
• 直接传递字典： 节点把原始数据直接传给工厂函数，消除执行图中不必要的解析开销。

3. 显式的状态管理

我们不用错误计数来推断成功，而是使用布尔校验标志（例如 prompt_ok、image_ok）。

• 清晰的失败模式： 路由逻辑直接检查标志。
• 人类可读的日志： 重试逻辑先自增再检查上限，确保日志清楚地记录从"第 1 次尝试"到"第 N 次尝试"。

4. 边界分离

我们严格区分工具结果（Tool Results）和数据库记录（Database Records）。

• 契约独立： 用于校验 tool 输出的 schema 可以独立于用于持久化存储的 schema 演进。
• 类型安全契约： 数据库记录接受已校验的 BaseModel 实例，而不是原始字典或 Any 类型。

---

架构设计模式

该实现遵循三块结构：

组件	职责
治理配置（Governance Config）	使用 OmegaConf 进行集中式策略管理，支持模式切换（严格/宽松）。
契约与闸门（Contracts & Gates）	缓存的 Pydantic 工厂，作为针对不可信数据的防御边界。
图编排（Graph Wiring）	管理状态持久化、checkpointers 和重试逻辑的 LangGraph 工作流。

---

Pydantic 与 Instructor 的战略对比

Pydantic 在图中保护系统边界，而 Instructor 则把恢复能力下移到生成的源头。

特性	LangGraph + Pydantic	Instructor
恢复逻辑	恢复逻辑位于**图（Graph）**中的 refine 节点。	恢复逻辑位于客户端（Client），通过自动重试实现。
可见性	每一步重试都是 trace 中一个独立的节点。	重试在 LLM 调用内部处理。
复杂度	较高（需要条件路由和循环）。	较低（从生成到下一步是线性流程）。
最适合	复杂的多 agent 系统和审计密集型任务。	结构化数据提取和简单的"修复"循环。

---

关键实现细节

这个 notebook 演示了几个生产关键修复：

1. 稳定的 Schema 哈希： 动态生成的模型使用显式标题，确保 JSON schema 引用跨进程保持稳定。
2. 激进的规范化： 校验关键字和策略值在哈希前会被排序并转小写，避免不必要的缓存未命中。
3. 可追溯性： run_id 字段把每条数据库记录关联到具体的 LangGraph thread_id 和 checkpoint 状态。

%%capture
!pip install langgraph langchain-core langchain-openai typing-extensions instructor

import os
from dotenv import load_dotenv


load_dotenv()

OPENROUTER_API_KEY = os.getenv('OPENROUTER_API_KEY')

导入

from __future__ import annotations

# Standard library
import json
import hashlib
from datetime import datetime, timezone
from operator import add
from typing import Any, Annotated, Literal, Optional, List
from typing_extensions import TypedDict

# LangChain / LangGraph
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver, MemorySaver

# Third party
from omegaconf import OmegaConf, DictConfig
from pydantic import BaseModel, Field, ValidationError, field_validator
import instructor

问题：静默失败

让我们从一个常见场景开始：一个处理用户数据的 agent。没有 Pydantic 时，类型不匹配会静默失败，或在后面造成难以理解的错误。

静默失败 - 字符串与整数：你的数据库期望一个整数，但 LLM 返回了一个字符串。

def process_user_data_brittle(data: dict):
    """Without validation, this fails silently or crashes unpredictably."""
    user_id = data.get("user_id")  # Could be "123" (string) or 123 (int)
    age = data.get("age")  # Could be "25" or 25

    # This works in staging (maybe the test data is correct)
    # But fails in production when the LLM returns strings
    result = user_id * 2  # TypeError if user_id is "123"
    database_query = f"SELECT * FROM users WHERE age > {age}"  # SQL injection risk if age is string

    return {"processed_id": result, "query": database_query}

# Simulate what happens when an LLM returns inconsistent data
llm_response_1 = {"user_id": 123, "age": 25}  # ✅ Works (staging)
llm_response_2 = {"user_id": "123", "age": "25"}  # ❌ Fails silently (production)

print("Response 1 (staging):")
try:
    result = process_user_data_brittle(llm_response_1)
    print(f"  Success: {result}")
except Exception as e:
    print(f"  Error: {e}")

print("\nResponse 2 (production):")
try:
    result = process_user_data_brittle(llm_response_2)
    print(f"  Success: {result}")
except Exception as e:
    print(f"  Error: {type(e).__name__}: {e}")

Response 1 (staging):
  Success: {'processed_id': 246, 'query': 'SELECT * FROM users WHERE age > 25'}

Response 2 (production):
  Success: {'processed_id': '123123', 'query': 'SELECT * FROM users WHERE age > 25'}

解决方案：Pydantic 校验

Pydantic 会在边界立即捕获这些错误，防止它们在整个系统中传播。

# Define the exact shape of your data
class UserData(BaseModel):
    """Technical spec: defines exactly what data we expect."""
    user_id: int = Field(..., description="User ID must be an integer")
    age: int = Field(..., ge=0, le=150, description="Age must be between 0 and 150")
    email: Optional[str] = Field(None, description="Optional email address")

    class Config:
        # Automatically convert strings to ints when possible
        # This handles provider differences gracefully
        str_strip_whitespace = True

def process_user_data_safe(data: dict):
    """With Pydantic, validation happens at the boundary."""
    try:
        # Validation happens here - fails fast if data is wrong
        user = UserData(**data)

        # Now we can trust the types
        result = user.user_id * 2
        database_query = f"SELECT * FROM users WHERE age > {user.age}"

        return {"processed_id": result, "query": database_query, "validated": user}
    except ValidationError as e:
        return {"error": "Validation failed", "details": str(e)}

# Same test cases
llm_response_1 = {"user_id": 123, "age": 25}
llm_response_2 = {"user_id": "123", "age": "25"}  # Strings - Pydantic converts them!
llm_response_3 = {"user_id": "abc", "age": "25"}  # Invalid - Pydantic catches it!

print("Response 1 (valid ints):")
result = process_user_data_safe(llm_response_1)
print(f"  {result}")

print("\nResponse 2 (string numbers - Pydantic converts):")
result = process_user_data_safe(llm_response_2)
print(f"  {result}")

print("\nResponse 3 (invalid data - Pydantic catches):")
result = process_user_data_safe(llm_response_3)
print(f"  {result}")

Response 1 (valid ints):
  {'processed_id': 246, 'query': 'SELECT * FROM users WHERE age > 25', 'validated': UserData(user_id=123, age=25, email=None)}

Response 2 (string numbers - Pydantic converts):
  {'processed_id': 246, 'query': 'SELECT * FROM users WHERE age > 25', 'validated': UserData(user_id=123, age=25, email=None)}

Response 3 (invalid data - Pydantic catches):
  {'error': 'Validation failed', 'details': "1 validation error for UserData\nuser_id\n  Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='abc', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.12/v/int_parsing"}

/tmp/ipykernel_1461/2645436559.py:2: PydanticDeprecatedSince20: Support for class-based `config` is deprecated, use ConfigDict instead. Deprecated in Pydantic V2.0 to be removed in V3.0. See Pydantic V2 Migration Guide at https://errors.pydantic.dev/2.12/migration/
  class UserData(BaseModel):

服务提供商的差异 - 真实世界的问题

不同的 LLM 服务提供商返回的 JSON 结构略有不同。没有校验时，你需要为每个提供商编写自定义解析逻辑。

# Simulate different provider responses
class ProviderResponse(BaseModel):
    """Unified model that works across all providers."""
    task: str
    priority: int = Field(ge=1, le=5)
    tags: List[str] = Field(default_factory=list)
    metadata: dict = Field(default_factory=dict)

# Provider A: Returns nested structure
provider_a_response = {
    "task": "Write blog post",
    "priority": "3",  # String instead of int
    "tags": ["ai", "agents"],
    "metadata": {"source": "user"}
}

# Provider B: Returns flat structure with different field names
provider_b_response = {
    "task": "Write blog post",
    "priority": 3,
    "tags": "ai,agents",  # Comma-separated string instead of list
    "extra": {"source": "user"}  # Different key name
}

# Provider C: Returns valid JSON but missing fields
provider_c_response = {
    "task": "Write blog post",
    "priority": 3
    # Missing tags and metadata
}

def parse_provider_response_brittle(data: dict, provider: str):
    """Without Pydantic, you need custom parsing for each provider."""
    if provider == "A":
        priority = int(data.get("priority", 0))
        tags = data.get("tags", [])
    elif provider == "B":
        priority = data.get("priority", 0)
        tags = data.get("tags", "").split(",") if isinstance(data.get("tags"), str) else data.get("tags", [])
    else:
        priority = data.get("priority", 0)
        tags = data.get("tags", [])

    return {"task": data.get("task"), "priority": priority, "tags": tags}

def parse_provider_response_safe(data: dict):
    """With Pydantic, one model handles all providers."""
    # Pydantic automatically:
    # - Converts string "3" to int 3
    # - Handles missing fields (uses defaults)
    # - Validates types and constraints
    try:
        # Handle provider B's comma-separated tags
        if isinstance(data.get("tags"), str):
            data["tags"] = [tag.strip() for tag in data["tags"].split(",") if tag.strip()]

        # Handle provider B's different metadata key
        if "extra" in data and "metadata" not in data:
            data["metadata"] = data.pop("extra")

        return ProviderResponse(**data)
    except ValidationError as e:
        return {"error": str(e)}

print("Provider A (string priority):")
result = parse_provider_response_safe(provider_a_response)
print(f"  {result}")

print("\nProvider B (comma-separated tags, different key):")
result = parse_provider_response_safe(provider_b_response)
print(f"  {result}")

print("\nProvider C (missing fields):")
result = parse_provider_response_safe(provider_c_response)
print(f"  {result}")

Provider A (string priority):
  task='Write blog post' priority=3 tags=['ai', 'agents'] metadata={'source': 'user'}

Provider B (comma-separated tags, different key):
  task='Write blog post' priority=3 tags=['ai', 'agents'] metadata={'source': 'user'}

Provider C (missing fields):
  task='Write blog post' priority=3 tags=[] metadata={}

为什么 Pydantic 对 Agents 很重要

1. 防止静默失败：在边界捕获类型不匹配，而不是在代码深处
2. 服务提供商一致性：一个模型处理所有提供商的变化（字符串 vs 整数、不同的键等）
3. 类型安全：你的 IDE 和类型检查器能理解你的数据结构
4. 自文档化：模型本身就是规范——无需单独的文档
5. LangGraph 集成：已校验的 state 防止 agent 步骤之间发生数据损坏

最佳实践

• 在边界定义模型：数据进入系统的地方（LLM 响应、API 调用、用户输入）
• 使用 Field 约束：ge、le、pattern 用于校验规则
• 处理提供商差异：Pydantic 的类型强制转换能处理常见的变化
• 快速失败：让校验错误立即冒出来，不要试图"修复"无效数据

所有概念的整合

治理配置与策略选择

GOVERNANCE_CONFIG = """
# Runtime mode: "strict", "moderate", or "lenient"
mode: "strict"

# Mode-specific policies
policies:
  strict:
    max_retries: 3
    prompt:
      min_length: 50
      required_keywords: ["pydantic", "validation", "contract"]
      forbid_markdown: true  # Real constraint: no code blocks in prompts
      require_style_tag: true  # Must include style descriptor
      size_px_min: 512
      size_px_max: 1536
    image_output:
      allowed_mime: ["image/png"]
      max_bytes: 2000000
    db:
      serialize_mode: "json"
      schema_version: 1

  moderate:
    max_retries: 2
    prompt:
      min_length: 30
      required_keywords: ["pydantic"]
      forbid_markdown: false
      require_style_tag: false
      size_px_min: 256
      size_px_max: 2048
    image_output:
      allowed_mime: ["image/png", "image/jpeg"]
      max_bytes: 5000000
    db:
      serialize_mode: "json"
      schema_version: 1

  lenient:
    max_retries: 1
    prompt:
      min_length: 20
      required_keywords: []
      forbid_markdown: false
      require_style_tag: false
      size_px_min: 256
      size_px_max: 4096
    image_output:
      allowed_mime: ["image/png", "image/jpeg", "image/webp"]
      max_bytes: 10000000
    db:
      serialize_mode: "json"
      schema_version: 1
"""

cfg: DictConfig = OmegaConf.create(GOVERNANCE_CONFIG)

def get_active_policy(cfg: DictConfig) -> DictConfig:
    """Get the active policy based on current mode."""
    mode = str(cfg.mode)
    return cfg.policies[mode]


def sha256_text(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def hash_policy(policy_dict: dict[str, Any]) -> str:
    """Generate deterministic hash of policy for versioning."""
    canonical = json.dumps(policy_dict, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    return sha256_text(canonical)[:16]

契约与闸门（带缓存）

# Cache Pydantic models by (mode, policy_hash) to avoid stale validators
# Returns: (ImagePrompt, GeneratedImageToolResult, GeneratedImageRecord)
_model_cache: dict[tuple[str, str], tuple[type[BaseModel], type[BaseModel], type[BaseModel]]] = {}


def build_image_prompt_model(policy_dict: dict[str, Any], mode: str, policy_hash: str):
    """Factory: Build Pydantic model from policy dict. Cached by mode and policy_hash."""
    prompt_cfg = policy_dict["prompt"]
    min_len = int(prompt_cfg["min_length"])
    required = [str(x).lower() for x in prompt_cfg.get("required_keywords", [])]
    size_min = int(prompt_cfg["size_px_min"])
    size_max = int(prompt_cfg["size_px_max"])
    forbid_md = bool(prompt_cfg.get("forbid_markdown", False))
    require_style = bool(prompt_cfg.get("require_style_tag", False))

    class ImagePrompt(BaseModel):
        model_config = {"title": f"ImagePrompt_{mode}_{policy_hash}"}  # Stable schema refs

        prompt: str = Field(..., min_length=min_len)
        size_px: int = Field(default=1024, ge=size_min, le=size_max)

        @field_validator("prompt")
        @classmethod
        def enforce_keywords(cls, v: str) -> str:
            if required:
                low = v.lower()
                missing = [k for k in required if k not in low]
                if missing:
                    raise ValueError(f"Prompt missing required keywords: {missing}")
            return v

        @field_validator("prompt")
        @classmethod
        def forbid_markdown_blocks(cls, v: str) -> str:
            # Only check for code blocks (triple backticks), not inline formatting
            if forbid_md and "```" in v:
                raise ValueError("Prompt must not contain markdown code blocks")
            return v

        @field_validator("prompt")
        @classmethod
        def require_style_descriptor(cls, v: str) -> str:
            if require_style:
                style_tags = ["minimal", "vector", "illustration", "photo", "diagram"]
                low = v.lower()
                if not any(tag in low for tag in style_tags):
                    raise ValueError(f"Prompt must include a style tag: {style_tags}")
            return v

    return ImagePrompt


def build_generated_image_model(policy_dict: dict[str, Any], mode: str, policy_hash: str):
    """Factory: Build Pydantic model from policy dict. Cached by mode and policy_hash."""
    image_cfg = policy_dict["image_output"]
    allowed_mime = {str(x) for x in image_cfg["allowed_mime"]}
    max_bytes = int(image_cfg["max_bytes"])

    class GeneratedImageToolResult(BaseModel):
        """Tool result contract: validates output from image generation tool."""
        model_config = {"title": f"GeneratedImageToolResult_{mode}_{policy_hash}"}  # Stable schema refs

        asset_id: str = Field(..., min_length=8)
        mime_type: str
        bytes_size: int = Field(..., ge=1, le=max_bytes)
        width_px: int = Field(..., ge=1)
        height_px: int = Field(..., ge=1)

        @field_validator("mime_type")
        @classmethod
        def validate_mime(cls, v: str) -> str:
            if v not in allowed_mime:
                raise ValueError(f"mime_type must be one of {sorted(allowed_mime)}")
            return v

    return GeneratedImageToolResult


def build_generated_image_record_model(policy_dict: dict[str, Any], mode: str, policy_hash: str):
    """Factory: Build DB record model for image storage. Separate from tool result contract."""
    image_cfg = policy_dict["image_output"]
    allowed_mime = {str(x) for x in image_cfg["allowed_mime"]}
    max_bytes = int(image_cfg["max_bytes"])

    class GeneratedImageRecord(BaseModel):
        """DB record contract: validates stored image fields. May drift independently from tool contract."""
        model_config = {"title": f"GeneratedImageRecord_{mode}_{policy_hash}"}  # Stable schema refs

        asset_id: str = Field(..., min_length=8)
        mime_type: str
        bytes_size: int = Field(..., ge=1, le=max_bytes)
        width_px: int = Field(..., ge=1)
        height_px: int = Field(..., ge=1)

        @field_validator("mime_type")
        @classmethod
        def validate_mime(cls, v: str) -> str:
            if v not in allowed_mime:
                raise ValueError(f"mime_type must be one of {sorted(allowed_mime)}")
            return v

    return GeneratedImageRecord


def get_models_for_policy(policy_dict: dict[str, Any], mode: str, policy_hash: str):
    """Get cached models for a policy. Cache key includes policy_hash to avoid stale validators.

    Returns: (ImagePrompt, GeneratedImageToolResult, GeneratedImageRecord)
    """
    cache_key = (mode, policy_hash)
    if cache_key not in _model_cache:
        _model_cache[cache_key] = (
            build_image_prompt_model(policy_dict, mode, policy_hash),
            build_generated_image_model(policy_dict, mode, policy_hash),  # Tool result contract
            build_generated_image_record_model(policy_dict, mode, policy_hash),  # DB record contract
        )
    return _model_cache[cache_key]


class MediaRecord(BaseModel):
    """DB record contract with schema versioning and audit trail.

    Note: image field accepts any BaseModel instance (GeneratedImageRecord is built dynamically
    per policy mode, but we enforce it's a validated Pydantic model via type annotation).
    """
    record_id: str
    created_at: str
    prompt_hash: str
    prompt_text: str
    image: BaseModel  # Typed contract (GeneratedImageRecord instance), validated before construction
    record_schema_version: int = Field(default=1)
    policy_hash: str  # Hash of policy snapshot for audit
    schema_hash: str  # Combined hash of schemas + policy semantics (for migration safety)
    run_id: str  # Run identifier tying DB record to LangGraph thread_id and checkpointer state

    def to_json_dict(self) -> dict[str, Any]:
        """Convert to JSON-compatible dict for DB storage using single point of truth."""
        # Defensive check: ensure image is a BaseModel (Pydantic validates, but explicit is clearer)
        if not isinstance(self.image, BaseModel):
            raise TypeError(f"image must be a BaseModel instance, got {type(self.image)}")

        # Use model_dump for entire record, but explicitly serialize nested image contract
        # This makes the contract boundary clear: we intentionally serialize the validated image
        data = self.model_dump(mode="json")
        data["image"] = self.image.model_dump(mode="json")  # Explicit contract serialization
        return data


class GraphState(TypedDict):
    messages: Annotated[list[BaseMessage], add]
    execution_log: Annotated[list[str], add]  # Uses additive reducer: every node appends

    image_prompt: dict[str, Any]
    generated_image: dict[str, Any]

    retry_count: int
    last_error: str
    prompt_ok: bool  # Explicit validation status
    image_ok: bool   # Explicit validation status

    db_record: dict[str, Any]
    active_policy: dict[str, Any]  # Pinned policy for this run
    active_mode: str  # Mode snapshot for this run
    policy_hash: str  # Hash of pinned policy (for cache key and audit)
    schema_hash: str  # Combined hash of schemas + policy semantics (for migration safety)
    max_retries: int  # Pinned max_retries from policy (avoid repeated lookups)
    run_id: str  # Run identifier tying DB record to LangGraph thread_id and checkpointer state

构建节点

def node_initialize_policy(state: GraphState):
    """Pin policy at START to ensure deterministic validation throughout run."""
    policy = get_active_policy(cfg)
    policy_dict = OmegaConf.to_container(policy, resolve=True)
    mode = str(cfg.mode)

    # Compute policy hash for caching and audit
    policy_hash = hash_policy(policy_dict)

    # Pre-build and cache models for this policy
    ImagePrompt, GeneratedImageToolResult, GeneratedImageRecord = get_models_for_policy(policy_dict, mode, policy_hash)

    # Extract policy semantics that affect validation but aren't in JSON Schema
    # (e.g., required_keywords, forbid_markdown, require_style_tag)
    # Normalize aggressively for hash stability
    prompt_cfg = policy_dict["prompt"]
    policy_semantics = {
        "required_keywords": sorted([str(x).lower() for x in prompt_cfg.get("required_keywords", [])]),  # Normalized
        "forbid_markdown": bool(prompt_cfg.get("forbid_markdown", False)),
        "require_style_tag": bool(prompt_cfg.get("require_style_tag", False)),
        "max_retries": int(policy_dict["max_retries"]),
        # If style_tags list exists in config, normalize it here too
        # "style_tags": sorted([str(x).lower() for x in prompt_cfg.get("style_tags", [])]),
    }

    # Compute combined schema hash: schemas + policy semantics for migration safety
    # This ensures schema_hash changes when validation behavior changes, even if JSON Schema doesn't
    prompt_schema = ImagePrompt.model_json_schema()
    tool_schema = GeneratedImageToolResult.model_json_schema()
    record_schema = GeneratedImageRecord.model_json_schema()
    combined = {
        "schemas": {
            "ImagePrompt": prompt_schema,
            "GeneratedImageToolResult": tool_schema,
            "GeneratedImageRecord": record_schema,
        },
        "policy_semantics": policy_semantics,
    }
    schema_hash = sha256_text(
        json.dumps(combined, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    )[:16]

    # Use run_id from state if present (set from thread_id in config), otherwise generate
    run_id = state.get("run_id") or f"run_{sha256_text(f'{mode}_{datetime.now(timezone.utc).isoformat()}')[:12]}"

    return {
        "active_policy": policy_dict,
        "active_mode": mode,
        "policy_hash": policy_hash,
        "schema_hash": schema_hash,
        "max_retries": int(policy_dict["max_retries"]),  # Store for routing (avoid repeated lookups)
        "run_id": run_id,  # Run identifier for DB record and checkpoint tracing
        "prompt_ok": False,
        "image_ok": False,
        "execution_log": [f"Initialized: mode={mode}, policy_hash={policy_hash}, schema_hash={schema_hash}, run_id={run_id}"],
    }


def node_content_writer(state: GraphState):
    """Agent A: Writes content draft."""
    return {
        "execution_log": ["A: Content draft written"],
    }


def node_propose_image_prompt(state: GraphState):
    """Agent B: Proposes image prompt (may be invalid initially)."""
    return {
        "image_prompt": {
            "prompt": "Short",  # Will fail validation
            "size_px": 1024,
        },
        "execution_log": ["B: Proposed image prompt"],
    }


def node_prompt_validation_gate(state: GraphState):
    """Validation Gate: Enforces Pydantic contract with retry logic."""
    policy_dict = state["active_policy"]
    mode = state["active_mode"]
    policy_hash = state["policy_hash"]
    ImagePrompt, _, _ = get_models_for_policy(policy_dict, mode, policy_hash)

    raw_prompt = state.get("image_prompt", {})
    retries = int(state.get("retry_count", 0))
    max_retries = int(state["max_retries"])  # Use pinned value from state

    try:
        valid_prompt = ImagePrompt(**raw_prompt)
        return {
            "image_prompt": valid_prompt.model_dump(),
            "retry_count": 0,
            "last_error": "",
            "prompt_ok": True,  # Explicit success flag
            "execution_log": [f"C: Prompt validated on attempt {retries + 1}"],
        }
    except ValidationError as e:
        error_msg = e.errors()[0]["msg"]

        # Increment retry count first, then check (makes log numbering clearer)
        new_retries = retries + 1

        if new_retries > max_retries:
            return {
                "retry_count": new_retries,  # Update so routing can detect exceeded state
                "prompt_ok": False,
                "execution_log": [f"C: Max retries ({max_retries}) exceeded on attempt {new_retries}. Last error: {error_msg}"],
                "last_error": error_msg,
            }

        return {
            "retry_count": new_retries,
            "last_error": error_msg,
            "prompt_ok": False,
            "execution_log": [f"C: Validation failed on attempt {new_retries}: {error_msg}"],
        }


def node_refine_prompt(state: GraphState):
    """Agent D: Refines prompt based on validation error."""
    policy_dict = state["active_policy"]
    last_error = state.get("last_error", "")

    prompt_cfg = policy_dict["prompt"]
    min_len = int(prompt_cfg["min_length"])
    required = [str(x) for x in prompt_cfg.get("required_keywords", [])]
    require_style = bool(prompt_cfg.get("require_style_tag", False))

    # Build prompt that satisfies all requirements
    style_part = "minimal vector illustration" if require_style else ""
    refined_prompt = {
        "prompt": f"A high-resolution technical illustration showing Pydantic validation contracts in a multi-agent system. {style_part} " +
                  " ".join([f"Focus on {kw}." for kw in required]) +
                  f" Clean style. Minimum {min_len} characters.",
        "size_px": 1024,
    }

    return {
        "image_prompt": refined_prompt,
        "execution_log": ["D: Prompt refined based on validation error"],
    }


def node_generate_image(state: GraphState):
    """Agent E: Calls image generation tool."""
    prompt = state["image_prompt"]["prompt"]

    tool_output = {
        "asset_id": sha256_text(prompt)[:12],
        "mime_type": "image/png",
        "bytes_size": 150000,
        "width_px": state["image_prompt"]["size_px"],
        "height_px": state["image_prompt"]["size_px"],
    }

    return {
        "generated_image": tool_output,
        "execution_log": ["E: Image generation tool executed"],
    }


def node_image_output_validation_gate(state: GraphState):
    """Validation Gate: Validates tool output before DB storage."""
    policy_dict = state["active_policy"]
    mode = state["active_mode"]
    policy_hash = state["policy_hash"]
    _, GeneratedImageToolResult, _ = get_models_for_policy(policy_dict, mode, policy_hash)

    raw_image = state.get("generated_image", {})

    try:
        valid_image = GeneratedImageToolResult(**raw_image)
        return {
            "generated_image": valid_image.model_dump(),
            "image_ok": True,  # Explicit success flag
            "execution_log": ["F: Image tool output validated"],
        }
    except ValidationError as e:
        error_msg = e.errors()[0]["msg"]
        return {
            "generated_image": {},
            "image_ok": False,
            "execution_log": [f"F: Image tool output rejected: {error_msg}"],
        }


def node_prepare_db_record(state: GraphState):
    """Prepares validated record for DB storage with JSON serialization."""
    policy_dict = state["active_policy"]
    mode = state["active_mode"]
    policy_hash = state["policy_hash"]
    schema_hash = state["schema_hash"]
    _, _, GeneratedImageRecord = get_models_for_policy(policy_dict, mode, policy_hash)

    prompt_text = state["image_prompt"]["prompt"]
    prompt_hash = sha256_text(prompt_text)

    # Convert tool result to DB record contract (may have different validation rules)
    # In this example they're identical, but in production they can drift independently
    image_record = GeneratedImageRecord(**state["generated_image"])

    record = MediaRecord(
        record_id=f"media_{prompt_hash[:12]}",
        created_at=datetime.now(timezone.utc).isoformat(),
        prompt_hash=prompt_hash,
        prompt_text=prompt_text,
        image=image_record,  # Typed contract (GeneratedImageRecord BaseModel), not dict
        record_schema_version=int(policy_dict["db"]["schema_version"]),
        policy_hash=policy_hash,  # Use stored hash from init
        schema_hash=schema_hash,  # Use stored hash from init
        run_id=state["run_id"],  # Run identifier tying DB record to thread_id and checkpointer
    )

    db_dict = record.to_json_dict()

    return {
        "db_record": db_dict,
        "execution_log": ["G: DB record prepared with JSON serialization"],
    }


def node_persist_to_db(state: GraphState):
    """Persists validated, JSON-serialized record to DB."""
    record = state["db_record"]
    return {
        "execution_log": [f"H: Persisted to DB: record_id={record['record_id']}"],
    }


def should_retry_prompt(state: GraphState) -> str:
    """Route based on explicit validation status."""
    if state.get("prompt_ok", False):
        return "ok"

    max_retries = int(state["max_retries"])  # Use pinned value from state
    retries = int(state.get("retry_count", 0))

    # Check if we've exceeded max retries (node increments first, then checks)
    if retries > max_retries:
        return "end"
    return "fix"


def should_continue_with_image(state: GraphState) -> str:
    """Route based on explicit validation status."""
    return "ok" if state.get("image_ok", False) else "end"

构建图

memory = InMemorySaver()
workflow = StateGraph(GraphState)

workflow.add_node("init_policy", node_initialize_policy)
workflow.add_node("writer", node_content_writer)
workflow.add_node("propose_prompt", node_propose_image_prompt)
workflow.add_node("prompt_gate", node_prompt_validation_gate)
workflow.add_node("refine_prompt", node_refine_prompt)
workflow.add_node("generate_image", node_generate_image)
workflow.add_node("image_gate", node_image_output_validation_gate)
workflow.add_node("prepare_db", node_prepare_db_record)
workflow.add_node("persist_db", node_persist_to_db)

workflow.add_edge(START, "init_policy")
workflow.add_edge("init_policy", "writer")
workflow.add_edge("writer", "propose_prompt")
workflow.add_edge("propose_prompt", "prompt_gate")

workflow.add_conditional_edges(
    "prompt_gate",
    should_retry_prompt,
    {"fix": "refine_prompt", "ok": "generate_image", "end": END},
)
workflow.add_edge("refine_prompt", "prompt_gate")

workflow.add_edge("generate_image", "image_gate")
workflow.add_conditional_edges(
    "image_gate",
    should_continue_with_image,
    {"ok": "prepare_db", "end": END},
)

workflow.add_edge("prepare_db", "persist_db")
workflow.add_edge("persist_db", END)

app = workflow.compile(checkpointer=memory)

运行应用

print("=" * 80)
print("PRODUCTION SYSTEM DEMONSTRATION")
print("=" * 80)
# Note: We print from cfg here for initial display, but actual execution uses pinned values from state
policy = get_active_policy(cfg)
print(f"Initial Mode: {cfg.mode}")
print(f"Max Retries: {policy.max_retries}")
print(f"Prompt Min Length: {policy.prompt.min_length}")
print(f"Required Keywords: {policy.prompt.required_keywords}")
print(f"Forbid Markdown: {policy.prompt.forbid_markdown}")
print(f"Require Style Tag: {policy.prompt.require_style_tag}")
print("=" * 80)
print()

config = {"configurable": {"thread_id": "production_run_001"}}

# Set run_id from thread_id to tie DB record to LangGraph checkpoint state
initial_state: GraphState = {
    "messages": [HumanMessage(content="Generate a blog cover image")],
    "execution_log": [],
    "image_prompt": {},
    "generated_image": {},
    "retry_count": 0,
    "last_error": "",
    "prompt_ok": False,
    "image_ok": False,
    "db_record": {},
    "active_policy": {},
    "active_mode": "",
    "policy_hash": "",
    "schema_hash": "",
    "max_retries": 0,
    "run_id": config["configurable"]["thread_id"],  # Tie DB record to LangGraph thread_id
}

print("--- Running Complete Workflow ---")
result = app.invoke(initial_state, config)

print("\n--- Execution Log ---")
for line in result["execution_log"]:
    print(f"  {line}")

print("\n--- DB Record (JSON-Compatible with Versioning) ---")
print(json.dumps(result["db_record"], indent=2, ensure_ascii=False))

print("\n--- Policy Snapshot (Pinned at Start) ---")
print(f"  Mode: {result['active_mode']}")  # Use pinned value from state, not cfg.mode
print(f"  Max Retries: {result.get('max_retries', 'N/A')}")  # Use pinned value from state
print(f"  Run ID: {result.get('run_id', 'N/A')}")  # Run identifier for tracing
print(f"  Policy Hash: {result['db_record'].get('policy_hash', 'N/A')}")
print(f"  Schema Hash: {result['db_record'].get('schema_hash', 'N/A')}")
print(f"  Schema Version: {result['db_record'].get('record_schema_version', 'N/A')}")

print("\n--- Fault Tolerance: Checkpoint Inspection ---")
snapshot = app.get_state(config)
print(f"  Last node: {snapshot.next if hasattr(snapshot, 'next') else 'N/A'}")
print(f"  Policy pinned: {bool(snapshot.values.get('active_policy'))}")
print(f"  Run ID: {snapshot.values.get('run_id', 'N/A')}")  # Ties DB record to checkpoint
print(f"  Policy hash: {snapshot.values.get('policy_hash', 'N/A')}")
print(f"  Schema hash: {snapshot.values.get('schema_hash', 'N/A')}")

================================================================================
PRODUCTION SYSTEM DEMONSTRATION
================================================================================
Initial Mode: strict
Max Retries: 3
Prompt Min Length: 50
Required Keywords: ['pydantic', 'validation', 'contract']
Forbid Markdown: True
Require Style Tag: True
================================================================================

--- Running Complete Workflow ---

--- Execution Log ---
  Initialized: mode=strict, policy_hash=64a35d3ba43027ae, schema_hash=7720b7473856aab3, run_id=production_run_001
  A: Content draft written
  B: Proposed image prompt
  C: Validation failed on attempt 1: String should have at least 50 characters
  D: Prompt refined based on validation error
  C: Prompt validated on attempt 2
  E: Image generation tool executed
  F: Image tool output validated
  G: DB record prepared with JSON serialization
  H: Persisted to DB: record_id=media_ad4f8f9896c2

--- DB Record (JSON-Compatible with Versioning) ---
{
  "record_id": "media_ad4f8f9896c2",
  "created_at": "2026-06-10T12:39:15.804294+00:00",
  "prompt_hash": "ad4f8f9896c2184e5f2a7861d1eac320a0927a12753e12d489397a324a8d90ae",
  "prompt_text": "A high-resolution technical illustration showing Pydantic validation contracts in a multi-agent system. minimal vector illustration Focus on pydantic. Focus on validation. Focus on contract. Clean style. Minimum 50 characters.",
  "image": {
    "asset_id": "ad4f8f9896c2",
    "mime_type": "image/png",
    "bytes_size": 150000,
    "width_px": 1024,
    "height_px": 1024
  },
  "record_schema_version": 1,
  "policy_hash": "64a35d3ba43027ae",
  "schema_hash": "7720b7473856aab3",
  "run_id": "production_run_001"
}

--- Policy Snapshot (Pinned at Start) ---
  Mode: strict
  Max Retries: 3
  Run ID: production_run_001
  Policy Hash: 64a35d3ba43027ae
  Schema Hash: 7720b7473856aab3
  Schema Version: 1

--- Fault Tolerance: Checkpoint Inspection ---
  Last node: ()
  Policy pinned: True
  Run ID: production_run_001
  Policy hash: 64a35d3ba43027ae
  Schema hash: 7720b7473856aab3

Pydantic 与 Instructor：边界保护与源头恢复

在你的 LangGraph 工作流中，Pydantic 充当边界闸门。它能防止无效数据传播，但恢复逻辑仍然位于编排层，例如一个独立的 refine 节点和显式的重试路由。

当你希望把恢复能力移到更靠近源头的位置时，Instructor 就很有用。Instructor 没有把结构化输出校验当作独立的工作流关注点，而是把 schema 执行与生成耦合在一起。模型会被反复要求生成满足同一个固定 Pydantic 契约的输出，并使用校验错误作为反馈。这减少了图的分支，消除了对专门"修复器"节点的需求，并让结构化输出行为在不同模型和服务提供商之间更加一致。

换句话说，Pydantic 保护系统边界。Instructor 则减少了稳定触达该边界所需的工作。

Instructor：在源头恢复

使用与之前相同的 Pydantic 模型（复用策略驱动的模型）

policy = get_active_policy(cfg)
policy_dict = OmegaConf.to_container(policy, resolve=True)
mode = str(cfg.mode)
policy_hash = hash_policy(policy_dict)

ImagePrompt, _, _ = get_models_for_policy(policy_dict, mode, policy_hash)

# Create Instructor client using OpenAI client (compatible with OpenRouter)
# Instructor works by patching the provider's create() method
# Architecture: Pydantic Model → Provider Handler → Dispatcher → Provider Client → Retry Layer → Instructor (patched)
from openai import OpenAI
model_slug = os.environ.get("OPENROUTER_MODEL", "anthropic/claude-3.5-haiku")

# Create OpenAI client configured for OpenRouter
client = instructor.from_provider(
    f"openrouter/{model_slug}",
    base_url="https://openrouter.ai/api/v1",
    mode=instructor.Mode.TOOLS,
)

print("=" * 80)
print("Automatic Recovery at Generation Time")
print("=" * 80)
print(f"Policy Mode: {mode}")
print(f"Max Retries: {policy.max_retries}")
print(f"Prompt Min Length: {policy.prompt.min_length}")
print(f"Required Keywords: {policy.prompt.required_keywords}")
print("=" * 80)
print()

# Instructor automatically retries with validation error feedback
# No need for separate refine nodes or conditional routing
print("--- Generating Image Prompt with Instructor ---")
print("Instructor will automatically retry if validation fails, using error messages as feedback.")
print()

try:
    # Instructor handles validation internally - no manual retry logic needed
    # The patched client's chat.completions.create() now supports response_model and max_retries
    # Flow: create() → retry layer → process_response → validation → reask if needed
    # When validation fails, Instructor's handle_reask_kwargs() appends error feedback
    result = client.create(                       # was client.chat.completions.create(...)
        messages=[
            {"role": "user",
            "content": "Generate an image prompt for a blog cover about Pydantic validation in multi-agent systems."}
        ],
        response_model=ImagePrompt,               # triggers schema/tool wiring + parsing
        max_retries=policy.max_retries,           # tenacity-backed validation retries
        temperature=0.2,
    )

    # Instructor returns the parsed Pydantic model directly (with _raw_response attached)
    print(f"✓ Successfully generated valid prompt after automatic retries:")
    print(f"  Prompt: {result.prompt[:100]}...")
    print(f"  Length: {len(result.prompt)} characters")
    print(f"  Size: {result.size_px}px")
    # Access raw provider response if needed: result._raw_response
    print()
    print("Key difference: No separate 'refine_prompt' node needed!")
    print("Instructor handled validation and retry logic internally.")

except Exception as e:
    print(f"✗ Failed after max retries: {e}")

================================================================================
Automatic Recovery at Generation Time
================================================================================
Policy Mode: strict
Max Retries: 3
Prompt Min Length: 50
Required Keywords: ['pydantic', 'validation', 'contract']
================================================================================

--- Generating Image Prompt with Instructor ---
Instructor will automatically retry if validation fails, using error messages as feedback.

✓ Successfully generated valid prompt after automatic retries:
  Prompt: A futuristic digital landscape showcasing a contract-based validation system for multi-agent AI inte...
  Length: 568 characters
  Size: 1024px

Key difference: No separate 'refine_prompt' node needed!
Instructor handled validation and retry logic internally.

用 Instructor 简化 LangGraph 工作流

# Simplified state - no retry_count, last_error, or prompt_ok needed
class SimplifiedGraphState(TypedDict):
    messages: Annotated[list[BaseMessage], add]
    execution_log: Annotated[list[str], add]
    image_prompt: dict[str, Any]
    generated_image: dict[str, Any]

def node_propose_prompt_with_instructor(state: SimplifiedGraphState):
    """Generate prompt using Instructor - validation and retry happen here."""
    # Instructor handles validation internally, so we get a valid prompt directly
    # The patched client automatically retries with validation error feedback
    result = client.chat.completions.create(
        model=os.environ.get("OPENROUTER_MODEL", "anthropic/claude-3.5-haiku"),
        messages=[
            {
                "role": "user",
                "content": "Generate an image prompt for a blog cover about Pydantic validation in multi-agent systems."
            }
        ],
        response_model=ImagePrompt,
        max_retries=policy.max_retries,
        temperature=0.2,
    )

    return {
        "image_prompt": result.model_dump(),
        "execution_log": ["B: Generated valid prompt with Instructor (automatic retry on validation errors)"],
    }

def node_generate_image_simple(state: SimplifiedGraphState):
    """Generate image - no validation gate needed if Instructor ensures valid input."""
    prompt = state["image_prompt"]["prompt"]

    tool_output = {
        "asset_id": sha256_text(prompt)[:12],
        "mime_type": "image/png",
        "bytes_size": 150000,
        "width_px": state["image_prompt"]["size_px"],
        "height_px": state["image_prompt"]["size_px"],
    }

    return {
        "generated_image": tool_output,
        "execution_log": ["E: Image generation tool executed"],
    }

# Build simplified graph - no refine node, no conditional retry routing
simplified_workflow = StateGraph(SimplifiedGraphState)

simplified_workflow.add_node("propose_prompt", node_propose_prompt_with_instructor)
simplified_workflow.add_node("generate_image", node_generate_image_simple)

# Simple linear flow - no branching for retries
simplified_workflow.add_edge(START, "propose_prompt")
simplified_workflow.add_edge("propose_prompt", "generate_image")
simplified_workflow.add_edge("generate_image", END)

simplified_app = simplified_workflow.compile(checkpointer=MemorySaver())

print("=" * 80)
print("Running Simplified LangGraph Workflow with Instructor")
print("=" * 80)
print()

initial_state = {
    "messages": [HumanMessage(content="Generate a blog cover image")],
    "execution_log": [],
    "image_prompt": {},
    "generated_image": {},
}

result = simplified_app.invoke(initial_state, {"configurable": {"thread_id": "instructor_demo"}})

print("--- Execution Log ---")
for line in result["execution_log"]:
    print(f"  {line}")

print()
print("--- Generated Prompt ---")
print(f"  Prompt: {result['image_prompt'].get('prompt', 'N/A')[:100]}...")
print(f"  Length: {len(result['image_prompt'].get('prompt', ''))} characters")

print()
print("=" * 80)
print("GRAPH COMPARISON:")
print("=" * 80)
print("Previous (LangGraph + Pydantic):")
print("  START → init_policy → writer → propose_prompt → prompt_gate")
print("  prompt_gate → [fix: refine_prompt → prompt_gate] | [ok: generate_image]")
print("  generate_image → image_gate → [ok: prepare_db] | [end: END]")
print("  prepare_db → persist_db → END")
print()
print("With Instructor:")
print("  START → propose_prompt → generate_image → END")
print()
print("Nodes removed: init_policy, writer, prompt_gate, refine_prompt,")
print("               image_gate, prepare_db, persist_db")
print("Conditional edges removed: retry routing logic")
print("=" * 80)

================================================================================
Running Simplified LangGraph Workflow with Instructor
================================================================================

--- Execution Log ---
  B: Generated valid prompt with Instructor (automatic retry on validation errors)
  E: Image generation tool executed

--- Generated Prompt ---
  Prompt: A futuristic digital landscape [vector] showcasing a data contract validation process in a multi-age...
  Length: 456 characters

================================================================================
GRAPH COMPARISON:
================================================================================
Previous (LangGraph + Pydantic):
  START → init_policy → writer → propose_prompt → prompt_gate
  prompt_gate → [fix: refine_prompt → prompt_gate] | [ok: generate_image]
  generate_image → image_gate → [ok: prepare_db] | [end: END]
  prepare_db → persist_db → END

With Instructor:
  START → propose_prompt → generate_image → END

Nodes removed: init_policy, writer, prompt_gate, refine_prompt,
               image_gate, prepare_db, persist_db
Conditional edges removed: retry routing logic
================================================================================