使用FastAPI、LangGraph与清晰架构构建生产级AI智能体

本指南将逐步构建从原始材料到可运行系统的完整流程，内容为：如何使用FastAPI、LangGraph和清晰架构来设计生产级AI代理。重点在于可操作的步骤、明确的检查点，以及可直接放入代码库的代码，无需猜测其用途。 在修改代码之前，应先明确输入参数、该步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 配置信息应与应用程序代码分开存放。环境文件、密钥存储和功能开关应集中管理，以便操作人员无需查看整个系统结构即可进行审核。

@router.post("/suppliers")
async def create_supplier(request: SupplierRequest):
    policies = opensearch.search(
            index="supplier-policies",
            query=request.description,
        )
        response = bedrock.converse(
            modelId=MODEL_ID,
            messages=build_messages(request, policies),
        )
        supplier = Supplier(
            name=request.name,
            tax_id=request.tax_id,
        )
        db.add(supplier)
        db.commit()
        sqs.send_message(
            QueueUrl=SUPPLIER_QUEUE,
            MessageBody=serialize(supplier),
        )
        return {"status": "created"}

FastAPI只是接口，而非应用程序本身

由于FastAPI仅是接口而非应用程序，在修改代码之前应先明确输入参数、各步骤的负责人以及终止条件。操作人员应当能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的功能。 在批量导入数据之前应使用约束条件和索引。通过对业务键设置唯一性，可将后续的合并操作转化为可预测的插入或更新操作，从而避免出现大量重复数据的问题。

@router.post("/suppliers")
async def create_supplier(
    request: CreateSupplierRequest,
    use_case: CreateSupplierUseCase = Depends(
        get_create_supplier_use_case
    ),
):
    command = CreateSupplierCommand(
        name=request.name,
        tax_id=request.tax_id,
        country=request.country,
    )

    result = await use_case.execute(command)
    return CreateSupplierResponse.from_result(result)

HTTP Request
     ↓
   FastAPI
     ↓
 Application
     ↓
   Domain
     ↓
    Ports
     ↓
  Adapters

应用程序应与功能交互，而非技术实现

由于应用关注的是功能而非技术，因此在修改代码之前需明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某个步骤失败时，故障应指向单一责任模块，而非复杂的流程链。 在批量导入数据之前应使用约束条件和索引。通过对业务键设置唯一性，可将后续的合并操作转化为可预测的插入或更新操作，从而避免出现数据重复的问题。

class CreateSupplierUseCase:
    def __init__(
        self,
        repository: SupplierRepository,
        policy_service: SupplierPolicyService,
        event_publisher: EventPublisher,
    ):
        self.repository = repository
        self.policy_service = policy_service
        self.event_publisher = event_publisher
    async def execute(
        self,
        command: CreateSupplierCommand,
    ) -> Supplier:
        existing = await self.repository.find_by_tax_id(
            command.tax_id
        )
        if existing:
            raise SupplierAlreadyExists(command.tax_id)
        policy = await self.policy_service.evaluate(
            country=command.country
        )
        supplier = Supplier.create(
            name=command.name,
            tax_id=command.tax_id,
            country=command.country,
            requires_approval=policy.requires_approval,
        )
        await self.repository.add(supplier)
        await self.event_publisher.publish(
            SupplierCreated(supplier.id)
        )
        return supplier

端口界定边界

在修改代码之前，需为相关流程定义边界、明确输入参数、指定该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，设定成功检测标准，杜绝默许的半完成状态。 在批量导入数据之前应使用约束条件与索引。通过对业务键的唯一性控制，可将后续的合并操作转化为可预测的插入或更新操作，从而避免数据重复问题。 在修改代码之前，需为相关流程定义边界、明确输入参数、指定该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个操作人员可审计的位置，无需阅读整个代码库。

页码。

from typing import Protocol

class SupplierRepository(Protocol):
    async def find_by_tax_id(
        self,
        tax_id: str,
    ) -> Supplier | None:
        ...
    async def add(
        self,
        supplier: Supplier,
    ) -> None:
        ...

class PolicyRetriever(Protocol):
    async def retrieve(
        self,
        query: str,
    ) -> list[PolicyDocument]:
        ...

class LLMProvider(Protocol):
    async def reason(
        self,
        context: AgentContext,
    ) -> AgentDecision:
        ...

class EventPublisher(Protocol):
    async def publish(
        self,
        event: DomainEvent,
    ) -> None:
        ...

SupplierRepository
PolicyRetriever
LLMProvider
EventPublisher

PostgreSQL
Amazon OpenSearch
Amazon Bedrock
AWS SQS

Amazon Bedrock的适用场景

在修改代码之前，需先明确输入内容、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品本身的组成部分，而非后续需要补充的功能。 在第一批数据处理完成后，应检查样本数据集。与端到端的聊天测试相比，Cypher探针能更低成本地检测到标签偏差和属性缺失问题。

class BedrockLLMProvider(LLMProvider):
    def __init__(self, client, model_id: str):
        self.client = client
        self.model_id = model_id
    async def reason(
        self,
        context: AgentContext,
    ) -> AgentDecision:
        response = self.client.converse(
            modelId=self.model_id,
            messages=build_messages(context),
        )
        return map_bedrock_response(response)

Application
     ↓
LLMProvider
     ↑
BedrockLLMProvider

RAG的适用场景

在确定RAG的适用场景时，应在修改代码之前明确输入内容、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比庞大的脚本，更应优先使用小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向单一责任方，而非复杂的流程链。在处理第一批数据后，需检查样本数据集的情况。与端到端的聊天测试相比，Cypher探针能更低成本地检测到标签偏差和属性缺失问题。

FastAPI
   ↓
OpenSearch
   ↓
LLM

class OpenSearchPolicyRetriever(PolicyRetriever):
    def __init__(
        self,
        opensearch_client,
        embedding_provider,
    ):
        self.client = opensearch_client
        self.embedding_provider = embedding_provider
    async def retrieve(
        self,
        query: str,
    ) -> list[PolicyDocument]:
        vector = await self.embedding_provider.embed(query)
        results = self.client.search(
            index="supplier-policies",
            body=build_vector_query(vector),
        )
        return map_documents(results)

RAG提供上下文，但并不决定最终决策。

对于“RAG提供上下文，但并不决定最终结果”这一原则，在修改代码之前需明确输入参数、各步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需猜测隐藏状态。 应将这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，设定成功检测标准，杜绝无声的半完成状态。 在处理第一批数据后需检查样本邻域情况。与端到端聊天测试相比，Cypher探针能更低成本地发现标签偏移和属性缺失问题。 对于“RAG提供上下文，但并不决定最终结果”这一原则，在修改代码之前需明确输入参数、各步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需猜测隐藏状态。 应将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关都应集中存放在操作人员可访问的位置。

无需阅读整个图结构即可进行审计。

RAG
 ↓
LLM
 ↓
"Looks fine"
 ↓
Create Supplier

RAG
 ↓
Relevant Context
 ↓
Agent Reasoning
 ↓
Application
 ↓
Domain Rules
 ↓
Decision

LangGraph的定位

在确定LangGraph的定位时，应在修改代码之前明确输入参数、各步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及错误处理都是产品本身的组成部分，而非后续需要补充的功能。 应为要解决的问题构建模型节点与关联关系，而非针对文档中的所有名词都建立模型。稀疏且有类型的边比密集且难以理解的图结构更佳。

Understand Request
       ↓
Retrieve Policies
       ↓
Evaluate Information
       ↓
Need More Data?
   ↙          ↘
 Yes          No
  ↓            ↓
Ask User   Continue
               ↓
       Approval Required?
          ↙          ↘
        Yes           No
         ↓             ↓
   Human Approval   Continue
          ↘          ↙
        Request Action

graph = StateGraph(AgentState)
graph.add_node(
    "understand_intent",
    understand_intent,
)
graph.add_node(
    "retrieve_policies",
    retrieve_policies,
)
graph.add_node(
    "evaluate",
    evaluate_request,
)
graph.add_node(
    "human_approval",
    request_human_approval,
)
graph.add_node(
    "request_action",
    request_application_action,
)

graph.add_edge(
    "understand_intent",
    "retrieve_policies",
)
graph.add_edge(
    "retrieve_policies",
    "evaluate",
)
graph.add_conditional_edges(
    "evaluate",
    determine_next_step,
    {
        "approval": "human_approval",
        "execute": "request_action",
    },
)

LangGraph不应成为新的单一庞大系统

为避免LangGraph变成新的单一系统，在修改代码之前应明确输入参数、各步骤的负责人以及终止标准。操作人员应当能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比庞大的脚本，更应优先使用小型且可测试的单元。当某个步骤出错时，错误原因应能指向具体的责任主体，而非复杂的流程问题。应为要解决的问题构建模型节点与关联关系，而非针对文档中的所有名词都进行建模；稀疏且有类型的边比密集且难以理解的图结构更佳。

def evaluate_node(state):
    if state.amount > 100_000:
        state.requires_approval = True
    if state.country == "BR":
        ...
    if state.supplier_type == "CRITICAL":
        ...

decision = supplier_policy.evaluate(
    supplier=supplier,
    context=context,
)

LangGraph → orchestrates
Domain → decides

MCP的适用场景

在确定MCP的适用场景时，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将这一阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功判定标准，杜绝默许的半完成状态。 针对要提出的问题来建模节点及其关联关系，而非文档中的所有名词。稀疏且有类型的边比密集且难以理解的图结构更佳。 在确定MCP的适用场景时，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于一个操作员能够审核的位置，无需阅读整个系统结构。

LLM
 ↓
PostgreSQL

LLM
 ↓
SAP

Agent
  ↓
MCP Tool
  ↓
Application API
  ↓
Use Case
  ↓
Domain
  ↓
Infrastructure

create_supplier

execute_sql

PostgreSQL也是一种适配器

在修改代码之前，对于“PostgreSQL也是一种适配器”这一方案，需先明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与故障恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的功能。 在批量导入数据之前应使用约束条件和索引。通过对业务键设置唯一性，可将后续的合并操作转化为可预测的插入或更新操作，从而避免出现数据重复的问题。

class SupplierRepository(Protocol):
    async def add(
        self,
        supplier: Supplier,
    ) -> None:
        ...

class PostgresSupplierRepository(
    SupplierRepository
):
    def __init__(self, session):
        self.session = session
    async def add(
        self,
        supplier: Supplier,
    ) -> None:
        entity = SupplierModel.from_domain(
            supplier
        )
        self.session.add(entity)

FastAPI      → doesn't know PostgreSQL exists

Application  → doesn't know PostgreSQL existsDomain       → doesn't know PostgreSQL existsInfrastructure → does

SAP应位于另一边界之后

对于SAP系统，应将其置于另一边界之后，在修改代码之前需明确输入参数、该步骤的负责人以及退出标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏的状态。相比庞大的脚本，更应采用小型且可测试的单元。当某个步骤失败时，故障原因应能指向单一责任主体，而非复杂的流程链。在批量导入数据之前，应使用约束条件和索引；通过对业务键设置唯一性，可将后续的合并操作转化为可预测的插入或更新操作，从而避免数据重复问题。

CreateSupplierUseCase
        ↓
Persist State
        ↓
SupplierCreated
        ↓
AWS SQS
        ↓
Integration Worker
        ↓
SAP

构建整体架构

在构建架构时，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将这一阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功检测标准，并拒绝默许的半完成状态。 在批量导入数据之前先设置约束条件与索引。通过对业务键的唯一性控制，可将后续的合并操作转化为可预测的插入或更新操作，从而避免数据重复问题。 在构建架构时，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于操作人员可审计的位置，且无需特殊权限即可查看。

对整个图表进行操作。

USER
                           │
                           ▼
                       FastAPI
                           │
                           ▼
                       LangGraph
                           │
              ┌────────────┼────────────┐
              │            │            │
              ▼            ▼            ▼
             RAG          LLM          MCP
              │            │            │
              ▼            ▼            │
         OpenSearch     Bedrock         │
                                        ▼
                                  Application
                                        │
                                        ▼
                                     Domain
                                        │
                              ┌─────────┴─────────┐
                              │                   │
                              ▼                   ▼
                         PostgreSQL            AWS SQS
                                                  │
                                                  ▼
                                               Worker
                                                  │
                                                  ▼
                                                 SAP

两个世界

在修改代码之前，需为“两个世界”明确输入参数、步骤执行者以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及错误消息处理都是产品本身的组成部分，而非后续需要补充的功能。 在第一批数据处理完成后，需检查样本邻域情况。与端到端的聊天测试相比，Cypher探针能更低成本地检测到标签偏移和属性缺失问题。

PROBABILISTICLLM
RAG
Natural Language
Intent Understanding
Agent Reasoning
Tool Selection

DETERMINISTIC
Authorization
Business Rules
Transactions
Persistence
Idempotency
Integration
Auditability

针对该架构的实际测试

对于架构的实用性测试，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比冗长的脚本，更应优先使用小型、可测试的单元。当某个步骤失败时，故障原因应能指向单一责任模块，而非复杂的流程链。在完成第一批处理后，需检查样本数据集的情况。与端到端的聊天测试相比，Cypher探针能更低成本地检测到标签偏差和属性缺失问题。

async def test_supplier_requires_approval():
    repository = FakeSupplierRepository()
    policies = FakePolicyService(
        requires_approval=True
    )
    events = FakeEventPublisher()
    use_case = CreateSupplierUseCase(
        repository=repository,
        policy_service=policies,
        event_publisher=events,
    )
    supplier = await use_case.execute(
        CreateSupplierCommand(
            name="ACME",
            tax_id="123",
            country="BR",
        )
    )
    assert supplier.requires_approval is True

修改模型应当是件枯燥的事

为确保模型更新过程枯燥无趣，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与验证后输出之间的契约。为相关成果命名，定义成功检测标准，并拒绝默许的半完成状态。 在处理第一批数据后检查样本邻域。与端到端聊天测试相比，Cypher探针能更低成本地检测到标签偏移和属性缺失问题。 为确保模型更新过程枯燥无趣，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将配置信息置于应用程序代码之外。环境文件、密钥存储及功能开关应集中存放于操作人员可审计且无需读取的地方。

覆盖整个图结构。

LLMProvider
             ▲
             │
   ┌─────────┴─────────┐
   │                   │
BedrockProvider   AnotherProvider

FastAPI
Application
Domain
Business Rules

整洁架构并非文件夹结构

在修改代码之前，需明确输入内容、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新执行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续的优化工作。 应为需要解答的问题建模节点与关联关系，而非文档中的所有名词。稀疏且有类型的边比密集且难以理解的图结构更佳。

src/
├── api/
├── application/
├── domain/
├── agents/
├── ports/
└── infrastructure/

人工智能让这些边界更为重要

由于人工智能使得这些边界更为重要，因此在修改代码之前必须明确输入内容、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比冗长的脚本，更应选择小型且可测试的单元。当某个步骤失败时，故障原因应能指向单一责任主体，而非错综复杂的流程。应为要提出的问题构建模型节点与关联关系，而非针对文档中的每个名词都这样做；稀疏且有类型的边比密集且难以理解的图结构更佳。

一句话概括架构

对于“用一句话描述架构”的方法，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将这一阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功判定标准，杜绝默许的半完成状态。 针对要提出的问题来建模节点与关联关系，而非文档中的每个名词。稀疏且有类型的边比密集且难以理解的图结构更佳。 对于“用一句话描述架构”的方法，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放于操作人员可审计且无需阅读其内容的地点。

整个图表。

总结

在修改代码之前，需先明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品功能的一部分，而非后续需要补充的内容。 在批量导入数据之前应使用约束条件和索引。通过对业务键设置唯一性，可将后续的合并操作转化为可预测的插入或更新操作，从而避免出现大量重复数据的问题。

Prompt
  ↓
 LLM
  ↓
 Tool
  ↓
Database

操作检查清单