从原型到量产：我在构建真正的智能代理系统中学到的经验

本指南将逐步展示如何从原始材料构建出可运行的系统，内容源自《从原型到生产：我使用LangChain和LangGraph构建真实智能体AI系统的经验》。重点在于可操作的步骤、明确的检查点，以及可直接放入代码库的代码，无需猜测其用途。

停止构建AI原型，开始推出能在现实世界中运行的AI系统。

在“停止构建AI原型”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比冗长的脚本，更应采用小型、可测试的单元。当某个步骤失败时，故障原因应能指向具体的责任主体，而非复杂的流程链。对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。

为何选择LangChain和LangGraph？

在 Why LangChain 和 LangGraph 阶段，应在修改代码之前明确输入内容、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务上的完整性。

架构：从文档到决策

在“基于文档的架构”阶段，应在修改代码之前明确输入内容、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。对于会产生支出或更改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。

Document Upload
      ↓
Chunking Pipeline (LangChain TextSplitter)
      ↓
Embedding Generation (Azure OpenAI ada-002)
      ↓
Vector Store (Azure AI Search — Hybrid Search)
      ↓
LangGraph Agent Orchestration
    ├── Classifier Agent (GPT-4)
    ├── Extractor Agent (GPT-4)
    ├── Validator Agent (GPT-3.5)
    └── HITL Node (Human-in-the-Loop)
      ↓
Structured Output → Backend → Database

第一层：RAG流程

在第一层RAG阶段，修改代码之前需明确输入内容、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审核。 需注明实际用于支撑答案的段落。若没有引用，操作人员就无法区分是幻觉内容还是索引缺失导致的错误。

分块策略

在分块策略阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及错误处理都是产品不可或缺的部分，而非后续需要补充的内容。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的逻辑连接并不等同于业务功能的完整性。 在分块策略阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 应将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功判定标准，绝不允许出现无声无息的半完成状态。

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=102,  # 20% overlap
    separators=["\n\n", "\n", ". ", " ", ""],
    length_function=len,
)chunks = splitter.split_documents(documents)

嵌入与向量存储

在处理嵌入与向量存储阶段时，首先明确相关规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外费用。 在调整提示词之前，先使用固定的问题集测试召回率。仅仅更换提示词往往无法改善较差的检索效果。

from langchain_openai import AzureOpenAIEmbeddings
from langchain_community.vectorstores import AzureSearch

embeddings = AzureOpenAIEmbeddings(
    azure_deployment="text-embedding-ada-002",
    azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
    api_key=os.getenv("AZURE_OPENAI_KEY"),
    api_version="2024-02-01",
)vector_store = AzureSearch(
    azure_search_endpoint=os.getenv("AZURE_SEARCH_ENDPOINT"),
    azure_search_key=os.getenv("AZURE_SEARCH_KEY"),
    index_name="tax-documents",
    embedding_function=embeddings.embed_query,
)

混合搜索——你做出的最重要优化

在处理“混合搜索最优化”阶段时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个位置，这样操作人员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作人员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。

retriever = vector_store.as_retriever(
    search_type="hybrid",
    search_kwargs={
        "k": 5,
        "score_threshold": 0.75,
    }
)

Redis嵌入缓存——成本降低40%

在处理 Redis Embedding Cache 40 阶段时，首先需明确相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词很难解决检索效果不佳的问题。 在处理 Redis Embedding Cache 40 阶段时，首先需明确相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 应将这一阶段视为输入与经过验证的输出之间的契约。为相关产物命名，定义成功判定标准，绝不允许出现无声无息的部分完成情况。

from langchain.cache import RedisSemanticCache
from langchain.globals import set_llm_cache
import redis

redis_client = redis.Redis(
    host=os.getenv("REDIS_HOST"),
    port=6380,
    password=os.getenv("REDIS_PASSWORD"),
    ssl=True,
)set_llm_cache(
    RedisSemanticCache(
        redis_=redis_client,
        embedding=embeddings,
        score_threshold=0.99,  # Very high — only cache exact or near-exact matches
    )
)

第二层：LangGraph Agent协调机制

将第二层的LangGraph Agent阶段视为可度量的对象使用效果最佳。在扩大范围之前，先记录一份理想的操作流程、一个故障案例以及回滚说明。 在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 需保持图结构的扁平化与类型化。嵌套的数据结构会掩盖具体是哪个节点设置了哪个字段，且在中断后会导致流程无法继续。

为何选择LangGraph而非简单链式结构？

将 LangGraph 的“为何简单阶段不如高级阶段”这一现象视为可测量的特征来分析最为有效。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 应将配置与应用程序代码分开。环境文件、密钥存储和功能标志应集中存放于一个位置，这样操作人员无需查看整个图结构即可进行审计。 需保持图状态的结构化与类型化。嵌套的数据块会掩盖是哪个节点修改了哪个字段，且在中断后会导致无法继续处理。

定义状态

将“定义状态”阶段视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一份理想状态下的流程日志、一个故障案例以及回滚说明。 同时记录正常流程与恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的一部分，而非后续需要补充的内容。 保持图结构的层次简单且类型明确。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在中断后导致流程无法继续。 将“定义状态”阶段视为输入与已验证输出之间的契约。为相关文档命名，明确成功标准，绝不允许出现无声无息的半完成状态。

from typing import TypedDict, Annotated, List, Optional
from langgraph.graph import StateGraph, END
import operator

class DocumentProcessingState(TypedDict):
    # Input
    document_id: str
    document_text: str
    document_chunks: List[str]

    # Processing
    document_type: Optional[str]
    extracted_fields: Optional[dict]
    validation_errors: List[str]
    retry_count: int

    # Output
    confidence_score: float
    requires_human_review: bool
    processing_notes: str
    final_result: Optional[dict]

构建图结构

在“构建图结构”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。对于那些会消耗资金或修改生产数据的节点，必须经过人工审批。仅靠编译时的配置并不足以确保业务的完整性。

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

workflow = StateGraph(DocumentProcessingState)# Add nodes
workflow.add_node("classifier", classify_document)
workflow.add_node("extractor", extract_fields)
workflow.add_node("validator", validate_extraction)
workflow.add_node("human_review", flag_for_human_review)
workflow.add_node("finalizer", finalize_result)# Define edges
workflow.set_entry_point("classifier")workflow.add_conditional_edges(
    "classifier",
    route_after_classification,
    {
        "extract": "extractor",
        "human_review": "human_review",
        "unknown": END,
    }
)workflow.add_conditional_edges(
    "extractor",
    route_after_extraction,
    {
        "validate": "validator",
        "retry": "extractor",
        "human_review": "human_review",
    }
)workflow.add_conditional_edges(
    "validator",
    route_after_validation,
    {
        "finalize": "finalizer",
        "retry": "extractor",
        "human_review": "human_review",
    }
)workflow.add_edge("finalizer", END)
workflow.add_edge("human_review", END)# Compile with checkpointing for resumability
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

分类节点

在“分类器节点”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置，以便操作人员无需查看整个流程即可进行审核。 对于涉及资金支出或修改生产数据的节点，需设置人工审批环节。编译时的连接方式并不等同于业务功能的完整性。

from langchain_openai import AzureChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field

class ClassificationResult(BaseModel):
    document_type: str = Field(
        description="One of: TAX_RETURN, PAYSLIP, TRAVEL_EXPENSE, IMMIGRATION_DOC, INVOICE, UNKNOWN"
    )
    confidence_score: float = Field(
        description="Confidence between 0.0 and 1.0"
    )
    reasoning: str = Field(
        description="Brief reasoning for the classification"
    )llm = AzureChatOpenAI(
    azure_deployment="gpt-4",
    azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
    api_key=os.getenv("AZURE_OPENAI_KEY"),
    api_version="2024-02-15-preview",
    temperature=0,  # Always 0 for classification — determinism matters
)classification_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are an expert tax document classifier.

Classify the document and return ONLY valid JSON matching the schema.
NEVER hallucinate values not present in the document.
If you cannot classify with confidence > 0.7, return UNKNOWN.{format_instructions}"""),
    ("human", "Document text:\n\n{document_text}"),
])parser = JsonOutputParser(pydantic_object=ClassificationResult)classifier_chain = classification_prompt | llm | parserdef classify_document(state: DocumentProcessingState) -> DocumentProcessingState:
    try:
        result = classifier_chain.invoke({
            "document_text": state["document_text"][:3000],  # Truncate for context window
            "format_instructions": parser.get_format_instructions(),
        })

        state["document_type"] = result["document_type"]
        state["confidence_score"] = result["confidence_score"]
        state["processing_notes"] += f"Classified as {result['document_type']} with confidence {result['confidence_score']:.2f}. "

    except Exception as e:
        state["document_type"] = "UNKNOWN"
        state["confidence_score"] = 0.0
        state["processing_notes"] += f"Classification failed: {str(e)}. "

    return state

路由功能——LangGraph的核心

在“路由功能核心”阶段，修改代码之前需明确输入参数、该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新执行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的逻辑连接并不等同于业务功能的完整性。 在“路由功能核心”阶段，修改代码之前需明确输入参数、该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新执行该步骤，而无需猜测隐藏状态。 应将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功判定标准，绝不允许出现无声无息的半完成状态。

def route_after_classification(state: DocumentProcessingState) -> str:
    if state["document_type"] == "UNKNOWN":
        return "unknown"
    elif state["confidence_score"] < 0.7:
        state["requires_human_review"] = True
        return "human_review"
    else:
        return "extract"

def route_after_extraction(state: DocumentProcessingState) -> str:
    if state["retry_count"] >= 3:
        state["requires_human_review"] = True
        return "human_review"
    elif not state.get("extracted_fields"):
        state["retry_count"] = state.get("retry_count", 0) + 1
        return "retry"
    else:
        return "validate"def route_after_validation(state: DocumentProcessingState) -> str:
    if state["validation_errors"]:
        if state["retry_count"] >= 3:
            state["requires_human_review"] = True
            return "human_review"
        state["retry_count"] = state.get("retry_count", 0) + 1
        return "retry"
    return "finalize"

第三层：生产环境可靠性模式

在处理第三层的生产环境可靠性阶段时，首先需明确相关约定：所需输入、成功标志以及部分故障时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在代码从演示环境迁移到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次对同一LLM调用收费。

结合Polly的断路器机制（C#后端）

在使用 Polly 的 Circuit Breaker 阶段时，首先需明确相关约定：所需的输入参数、成功信号，以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改不会偏离原有设计。 配置应与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中管理，这样操作人员无需查看整个系统结构即可进行审计。 在耗时较高的步骤之后设置检查点。当操作人员重新尝试后续节点时，恢复流程不应再次对同一次 LLM 调用计费。

// In .NET Core — Polly circuit breaker
services.AddHttpClient<ILLMGateway, LLMGateway>()
    .AddPolicyHandler(
        Policy<HttpResponseMessage>
            .Handle<HttpRequestException>()
            .OrResult(r => (int)r.StatusCode >= 500)
            .CircuitBreakerAsync(
                handledEventsAllowedBeforeBreaking: 5,
                durationOfBreak: TimeSpan.FromSeconds(60),
                onBreak: (ex, duration) =>
                    logger.LogWarning("Circuit breaker opened for {Duration}s", duration),
                onReset: () =>
                    logger.LogInformation("Circuit breaker reset")
            )
    )
    .AddPolicyHandler(
        Policy<HttpResponseMessage>
            .Handle<HttpRequestException>()
            .WaitAndRetryAsync(
                retryCount: 3,
                sleepDurationProvider: attempt =>
                    TimeSpan.FromSeconds(Math.Pow(2, attempt)) + // Exponential
                    TimeSpan.FromMilliseconds(new Random().Next(0, 1000)) // Jitter
            )
    );

并发限流（Python）

在处理 Python 并发限流阶段时，首先需明确接口规范：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。 每次调用时都要记录请求 ID、模型 ID 以及延迟时间。如果没有这些记录，间歇性的服务错误就会被视为应用程序的缺陷。 在处理 Python 并发限流阶段时，首先需明确接口规范：所需输入、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 应将此阶段视为输入与经过验证的输出之间的契约。为相关产物命名，定义成功判定标准，并杜绝无声的半完成状态。

import asyncio
from asyncio import Semaphore

class ThrottledDocumentProcessor:
    def __init__(self, max_concurrent: int = 50):
        self._semaphore = Semaphore(max_concurrent)

    async def process_document(
        self,
        document_id: str,
        document_text: str
    ) -> dict:
        async with self._semaphore:
            config = {"configurable": {"thread_id": document_id}}
            result = await app.ainvoke(
                {
                    "document_id": document_id,
                    "document_text": document_text,
                    "document_chunks": [],
                    "validation_errors": [],
                    "retry_count": 0,
                    "confidence_score": 0.0,
                    "requires_human_review": False,
                    "processing_notes": "",
                    "final_result": None,
                },
                config=config,
            )
            return result

GPT-4向GPT-3.5的回退方案

将GPT-4到GPT-3.5的转换过程视为可测量的流程时效果最佳。在扩大范围之前，需记录一份理想的测试案例、一个失败案例以及回滚说明。 在功能结果旁同时记录处理时间以及token或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 保持数据结构扁平且类型明确。嵌套的数据块会掩盖具体是哪个节点设置了哪一字段，且在中断后会导致流程无法继续。

async def classify_with_fallback(document_text: str) -> dict:
    models = [
        ("gpt-4", 0.95),       # Primary — higher confidence weight
        ("gpt-35-turbo", 0.80), # Fallback — lower confidence weight
    ]

    for deployment, confidence_multiplier in models:
        try:
            llm = AzureChatOpenAI(
                azure_deployment=deployment,
                temperature=0,
                request_timeout=30,
            )
            result = await classifier_chain.ainvoke({
                "document_text": document_text,
                "format_instructions": parser.get_format_instructions(),
            })
            result["confidence_score"] *= confidence_multiplier
            result["model_used"] = deployment
            return result
        except Exception as e:
            logger.warning(f"Model {deployment} failed: {e}. Trying next...")
            continue

    raise Exception("All models failed")

理想数据集的CI/CD回归测试

将“黄金数据集 CI CD 阶段”视为可度量的对象来管理，效果最佳。在扩大范围之前，需记录一份标准流程文档、一个故障案例以及回滚说明。 配置应与应用程序代码分开存放。环境文件、密钥存储和功能开关应集中于一处，以便操作人员无需查看整个架构就能进行审计。 保持架构状态的简洁性与类型化。嵌套的数据结构会掩盖哪个节点修改了哪个字段的信息，还会在中断后导致无法继续处理。

# golden_dataset.py
GOLDEN_DOCUMENTS = [
    {
        "id": "GD001",
        "text": "Employee: John Smith\nDate: 15-Mar-2026\nGross Pay: $4,500\nNet Pay: $3,200...",
        "expected_type": "PAYSLIP",
        "expected_fields": {
            "employee_name": "John Smith",
            "date": "2026-03-15",
            "gross_amount": 4500.0,
        },
        "min_confidence": 0.85,
    },
    # ... 200 more golden examples across all document types
]

async def run_regression_suite():
    passed = 0
    failed = 0
    failures = []

    for golden_doc in GOLDEN_DOCUMENTS:
        result = await processor.process_document(
            golden_doc["id"],
            golden_doc["text"]
        )

        # Check type
        if result["document_type"] != golden_doc["expected_type"]:
            failures.append({
                "id": golden_doc["id"],
                "expected_type": golden_doc["expected_type"],
                "actual_type": result["document_type"],
                "failure": "wrong_classification",
            })
            failed += 1
            continue

        # Check confidence
        if result["confidence_score"] < golden_doc["min_confidence"]:
            failures.append({
                "id": golden_doc["id"],
                "expected_confidence": golden_doc["min_confidence"],
                "actual_confidence": result["confidence_score"],
                "failure": "low_confidence",
            })
            failed += 1
            continue

        passed += 1

    pass_rate = passed / len(GOLDEN_DOCUMENTS)

    if pass_rate < 0.95:  # 95% pass rate required
        raise Exception(
            f"Regression suite failed: {pass_rate:.1%} pass rate "
            f"({failed} failures). Blocking deployment."
        )

    return {"passed": passed, "failed": failed, "pass_rate": pass_rate}

第 4 层：可观测性

将“第 4 层可观测性阶段”视为可度量的对象来管理，效果最佳。在扩大范围之前，需记录一份标准流程文档、一个故障案例以及回滚说明。 需同时记录正常运行路径和恢复路径。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。

每次 AI 调用都进行 OpenTelemetry 追踪

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from azure.monitor.opentelemetry.exporter import AzureMonitorTraceExporter

tracer_provider = TracerProvider()
tracer_provider.add_span_processor(
    BatchSpanProcessor(
        AzureMonitorTraceExporter(
            connection_string=os.getenv("APPLICATIONINSIGHTS_CONNECTION_STRING")
        )
    )
)
trace.set_tracer_provider(tracer_provider)
tracer = trace.get_tracer(__name__)async def classify_document_traced(
    state: DocumentProcessingState,
) -> DocumentProcessingState:
    with tracer.start_as_current_span("classify_document") as span:
        span.set_attribute("document.id", state["document_id"])
        span.set_attribute("document.length", len(state["document_text"]))

        start_time = time.time()
        result = await classify_document(state)
        latency_ms = (time.time() - start_time) * 1000

        span.set_attribute("llm.document_type", result["document_type"])
        span.set_attribute("llm.confidence_score", result["confidence_score"])
        span.set_attribute("llm.latency_ms", latency_ms)
        span.set_attribute("llm.model", "gpt-4")
        span.set_attribute("llm.requires_human_review", result["requires_human_review"])

        return result

需跟踪的关键指标

METRICS_TO_TRACK = {
    # Reliability
    "classification_accuracy": "% correct vs golden dataset",
    "confidence_score_distribution": "histogram — drift detection",
    "human_review_rate": "% documents escalated — should stay < 5%",

    # Performance
    "end_to_end_latency_p50": "median processing time per document",
    "end_to_end_latency_p99": "tail latency — SLA violations",
    "llm_call_latency": "broken down by model and node",

    # Cost
    "tokens_per_document": "input + output tokens",
    "cache_hit_rate": "Redis embedding cache effectiveness",
    "monthly_api_spend": "trend by document type",

    # Business
    "documents_processed_per_hour": "throughput",
    "analyst_hours_saved_weekly": "business impact",
    "error_rate_by_document_type": "which types fail most",
}

结果分析

主要经验总结

结论

操作检查清单