使用DSPy、Guardrails AI 和 MLflow

照片由 Jan Huber 来自 Unsplash

在这篇帖子中，我分享了一个使用FastAPI、React、LangGraph、DSPy、Guardrails AI和MLflow构建的AI驱动的注册系统。该系统通过引导用户进行结构化的对话流程，收集注册详情，并利用AI验证来确保输入的准确性。它通过使用AI进行验证循环过程、多步骤工作流程导航和实时数据驱动的调整来实现动态适应能力。虽然该系统自动化了结构化的交互，但它并未完全实现代理行为，保持了AI引导和可预测用户体验之间的平衡。

一个完全自主的系统会进一步自主决定交互顺序，根据更广泛的目调整工作流程，并在无需预设规则的情况下自我修正。不会遵循固定的路径，而是根据情境做出决策，动态调整任务的优先级，并不断优化其方式。

在预测性和自主性之间的权衡：**

• 可预测性确保了合规性、可重复性和用户信任——特别是在像金融和医疗保健这样的受监管的行业中，决策必须是可审计和可解释的。
• 自主性使AI能够在没有严格限制的情况下适应和优化工作流程，这可能提高效率，但代价是减少了监督的力度。
• 例如本系统这样的混合方法平衡了AI驱动的灵活性与人类监督和基于规则的约束，允许动态交互，同时保持治理和控制。

尽管相当简单，这个系统展示了人工智能如何让结构化工作流程更高效，同时保持透明和可靠，展示了一条实用的途径将人工智能融入到用户界面的应用程序中，同时确保易理解性。

• 对话式用户界面：用户通过类似聊天机器人的界面逐步输入注册信息，并可以选择跳过某些步骤。
• FastAPI 后端：处理会话管理、状态转换以及利用 AI 进行验证。
• React 前端：提供一个直观的用户体验，并根据用户的响应动态更新。
• 使用 LangGraph 进行流程控制：管理多步骤对话逻辑，确保注册步骤间的平滑过渡。
• 使用 OpenAI 的 GPT-3.5 Turbo 验证用户输入：系统使用 OpenAI 的 GPT-3.5 Turbo 验证用户输入，通过两种方式实现如下：

1. 使用 OpenAI Python 库：直接调用 OpenAI 的 API 来生成结构化的验证回复。
2. 使用 DSPy 优化提示：利用 DSPy，一个声明式的 AI 框架，优化提示并提高 AI 生成验证的可靠性。
3. 比较：虽然 OpenAI 库提供了直接的 API 调用，但 DSPy 通过学习过往的交互来提高提示效率并优化生成的响应。

• Guardrails AI 结构化验证工具：确保输入符合预期格式要求，修正小错误并请求澄清。
• MLflow 日志功能：跟踪验证表现情况，记录交互过程，并支持通过调试改进模型。

• Python 后端技术 : FastAPI, LangGraph, DSPy, MLflow, Guardrails AI
• 前端框架 : React (Vite), Material UI
• 使用的数据库 : SQLite (用于存储会话信息)
• 部署方式 : Docker, Docker Compose

本文介绍的对话式注册系统将传统的填表流程变成了一个动态引导式对话，从而提高用户参与度并保证数据质量。系统根据用户回答逐步收集信息，而不是展示一个静态的表格。尽管用户界面很简单，但它展示了如何通过强大的实时验证和格式化来创造一个既引导又开放的用户体验。

对话式注册用户使用体验

我在这篇文章中主要关注后端功能和相关的代码，因为它驱动了前端的动态性。我提供了一个链接到完整的代码库，这样你可以自己尝试前端和后端的代码。

在这个示例中，LangGraph 通过定义一个交互图来结构化注册流程，其中每个节点代表一个特定的问题（例如，“您的电子邮件是什么？”），而边则决定了对话如何根据用户的回答进行推进。系统可以动态跳过步骤、请求澄清或根据验证反馈调整流程，从而确保一个上下文感知且自适应的用户体验。通过利用 LangGraph，应用程序可以在交互之间保持状态，并根据需要恢复会话，确保用户顺利地完成注册流程，不会出现冗余或不相关的问题，从而使用户体验更加流畅。定义流程的有向循环图（DCG）如图所示。请注意，用户可以选择跳过ask_address 和 ask_phone 节点（用虚线表示）。

LangGraph DCG 的注册流程。

LangGraph 是一个用于构建结构化、有状态且多步骤的 AI 应用程序的框架，它使用基于图的工作流。不同于提示链或简单的顺序 API，LangGraph 支持灵活的分支、条件逻辑和内存持久化，这使得它非常适合像这个 AI 驱动的注册系统这样的复杂对话应用。需要注意的是，虽然我在这个示例中使用了 SQLite 来存储对话的状态，但实际上也可以选择使用 LangGraph 的内存持久化功能。下面的 Python 类负责构建 DCG，该 DCG 将被编译并在 FastAPI 后端使用。

    from langgraph.graph import END
    from db.sqlite_db import RegistrationState
    from graph.base_graph import BaseGraphManager
    import logging

    class RegistrationGraphManager(BaseGraphManager):
        """具有特定领域的逻辑的专用图管理器。"""

        def __init__(self, name: str, question_map: dict):
            # 我们将RegistrationState作为状态类
            super().__init__(name, question_map, RegistrationState)

        def _build_graph(self):
            """
            覆写基础构建方法以包含可选步骤或领域特定的边。
            """

            def ask_question(state: RegistrationState, question_text: str):
                logging.info(f"[注册] 过渡到：{question_text}")
                return {
                    "collected_data": state.collected_data,
                    "current_question": question_text,
                }

            def create_question_node(question_text):
                return lambda s: ask_question(s, question_text)

            # 添加每个节点
            for key, question_text in self.question_map.items():
                self.graph.add_node(key, create_question_node(question_text))

            # 设置入口点
            self.graph.set_entry_point("ask_email")

            # 前两个节点的普通边
            self.graph.add_edge("ask_email", "ask_name")

            def path_func(state: RegistrationState):
                """确定下一步去哪里："""
                skip_address = state.collected_data.get("skip_ask_address", False)
                skip_phone = state.collected_data.get("skip_ask_phone", False)

                if skip_address and skip_phone:
                    return "ask_username"  # 跳过地址和电话
                elif skip_address:
                    return "ask_phone" .  # 只跳过地址
                else:
                    return "ask_address"  # 默认先询问地址

            self.graph.add_conditional_edges(
                source="ask_name",
                path=path_func,
                path_map={
                    "ask_address": "ask_address",
                    "ask_phone": "ask_phone",
                    "ask_username": "ask_username",
                },
            )

            self.graph.add_edge("ask_address", "ask_phone")
            self.graph.add_conditional_edges(
                source="ask_phone",
                path=lambda state: (
                    "ask_username"
                    if state.collected_data.get("skip_ask_phone", False)
                    else "ask_username"
                ),
                path_map={"ask_username": "ask_username"},
            )
            self.graph.add_edge("ask_username", "ask_password")
            self.graph.add_edge("ask_password", END)

DSPy 是一个用于程序定义和优化 AI 工作流的框架，非常适合结构化验证任务。在这个应用中，DSPy 定义了一套验证规则，它接收用户的回答和当前问题作为输入，并生成结构化的输出信息。

• 状态 (有效，需要澄清 或 错误）：
• 反馈意见： （对任何问题的解释）
• 格式化的回答： （正确格式化的回答）

验证逻辑基于 OpenAI 的 GPT-3.5-turbo。DSPy 通过声明式 dspy.Predict 函数来调用。通过使用 DSPy，注册系统确保一致的响应格式（例如，小写的电子邮件地址、标准化的电话号码、姓名首字母大写），并识别不完整或模糊的输入。

此外，DSPy 与 Guardrails AI 无缝集成，增加了额外的验证层以确保符合预期的响应模式——这在后面的章节中介绍。实现中还包含 MLflow，用于记录每个验证步骤的详细日志，以进行审计和模型监控。以下是展示这种集成的 DSPy 验证代码。

import guardrails as gd  
import dspy  
from validation.base_validator import BaseValidator  
from pydantic import ValidationError  
from validation.validated_response import ValidatedLLMResponse  
from typing import Literal  
import logging  
import json  
import mlflow  
from helpers.config import OPENAI_API_KEY, MLFLOW_ENABLED, MLFLOW_EXPERIMENT_NAME  

# 导入 guardrails 作为 gd
# 从 validation.base_validator 导入 BaseValidator

dspy.settings 配置 lm=dspy.LM(model='gpt-3.5-turbo', api_key=OPENAI_API_KEY)

if MLFLOW_ENABLED:  
    mlflow.dspy.autolog()  

guard = gd.Guard.for_pydantic(ValidatedLLMResponse)  

# 定义DSPy签名
class ValidateUserAnswer(dspy.Signature):  
    """验证并格式化用户回答。应返回 'valid'、'clarify' 或 'error'。"""  
    question: str = dspy.InputField()  
    user_answer: str = dspy.InputField()  

    status: Literal["valid", "clarify", "error"] = dspy.OutputField(  
        desc="验证状态：'valid'、'clarify' 或 'error'。如果输入包含所有必要信息，则应返回 'valid'。"  
    )  
    feedback: str = dspy.OutputField(  
        desc="解释回答不正确或需要详细说明的内容。"  
    )  
    formatted_answer: str = dspy.OutputField(  
        desc="返回格式化后的回答。例如："  
        "- 邮箱：小写（例如 'John@gmail.com' → 'john@gmail.com'）。"  
        "- 姓名：首字母大写（例如 'john doe' → 'John Doe'）。"  
        "- 地址：首字大写并确保完整信息（例如 '123 main st,newyork,ny' → '123 Main St, New York, NY 10001'）。"  
        "- 电话号码：格式为 (XXX) XXX-XXXX（例如 '1234567890' → '(123) 456-7890'）。"  
        "地址必须包括：街道号码、街道名称、城市、州和邮政编码。"  
        "不满足此格式的回答应返回 'clarify'。"  
        "如果格式化回答失败，则返回原始回答。"  
    )  

定义 run_llm_validation 为 dspy.Predict(ValidateUserAnswer)  

class DSPyValidator(BaseValidator):  
    """使用DSPy和Guardrails AI进行结构化验证。"""  
    def validate(self, question: str, user_answer: str):  
        """验证用户回答，应用Guardrails，并使用MLflow记录。"""  
        try:  
            raw_result = run_llm_validation(question=question, user_answer=user_answer)  
            structured_validation_output = guard.parse(json.dumps(raw_result.toDict()))  
            validated_dict = dict(structured_validation_output.validated_output)  

            if MLFLOW_ENABLED:  
                mlflow 设置实验为 MLFLOW_EXPERIMENT_NAME  
                with mlflow.start_run(nested=True):  
                    mlflow.log_param("validation_engine", "DSPy + Guardrails AI")  
                    mlflow.log_param("question", question)  
                    mlflow.log_param("input_answer", user_answer)  
                    mlflow.log_param("status", validated_dict["status"])  
                    mlflow 记录参数 'formatted_answer' 为 validated_dict['formatted_answer']  

            return {  
                "status": validated_dict["status"],  
                "feedback": validated_dict["feedback"],  
                "formatted_answer": validated_dict["formatted_answer"],  
            }  

        except ValidationError as ve:  
            logging.error(f"验证错误：{str(ve)}")  
            return {  
                "status": "error",  
                "feedback": "输出验证失败。",  
                "formatted_answer": user_answer,  
            }  
        except Exception as e:  
            logging.error(f"验证错误：{str(e)}")  
            return {  
                "status": "error",  
                "feedback": "验证过程中发生错误。",  
                "formatted_answer": user_answer,  
            }  

The ValidateUserAnswer 类在上述代码片段中定义了一个 DSPy 签名规范，该规范作为声明性约束规范，用于定义输入和输出，确保结构化 AI 工作流中的 LLM 遵循预定义的结构，从而使验证更为可预测和可解释。在该用例中，签名如下所示：

问题, 用户的回复 → 状态码, 反馈, 回复

此结构化的定义被传递给 dspy.Predict，这样DSPy可以直接将其应用于用户的输入。MLflow捕获并记录DSPy发送给 gpt-3.5-turbo 的确切请求，以及模型的结构化回复，从而保证透明度和可重复性。

特别地，DSPy 遵循系统提示、用户提示和助手提示的结构，这在与 OpenAI 的 GPT 模型交互时是一种标准方法。这种格式有助于保持对话的上下文，确保 LLM 的验证逻辑保持一致。

    系统  

    您的输入字段为：  
    question (str)  
    user_answer (str)  
    您的输出字段为：  
    status (typing.Literal['valid', 'clarify', 'error'])：如果输入包含了所有必填信息，则返回 'valid'。  
    feedback (str)：如果响应不正确或需要详细信息，则返回解释。  
    formatted_answer (str)：返回格式正确的响应。例如：- 邮件：全部小写（例如 'John@gmail.com' → 'john@gmail.com'）。- 姓名：首字母大写（例如 'john doe' → 'John Doe'）。- 地址：确保信息完整并正确大写（例如 '123 main st,newyork,ny' → '123 Main St, New York, NY 10001'）。- 电话号码：格式为 (XXX) XXX-XXXX（例如 '1234567890' → '(123) 456-7890'）。地址必须包括：街道号码、街道名称、城市、州和邮政编码。如果响应未能符合此格式要求，则返回 'clarify' 状态。如果响应无法格式化，则返回原始答案。  
    所有交互将按照以下结构进行，其中适当填充相应值。  
    [[ ## question ## ]] {question}  
    [[ ## user_answer ## ]] {user_answer}  
    [[ ## status ## ]] {status} # 注意：您生成的值必须完全匹配（不能有额外字符）：valid；clarify；error  
    [[ ## feedback ## ]] {feedback}  
    [[ ## formatted_answer ## ]] {formatted_answer}  
    [[ ## completed ## ]]  
    遵循此结构，您的目标是验证和格式化用户响应。应返回 'valid'、'clarify' 或 'error'。

    用户  

    [[ ## question ## ]] 你住在哪儿？  
    [[ ## user_answer ## ]] 洛杉矶加州90210主街112号  
    请按以下格式回复，首先是字段 [[ ## 状态 ## ]] （必须格式化为有效的 Python 类型字面量 'typing.Literal["有效", "澄清", "错误"]'），然后是 [[ ## 反馈 ## ]]，接着是 [[ ## 格式化答案 ## ]]，最后以 [[ ## 完成 ## ]] 标记结束。

    助手  

    [[ ## status ## ]] 有效状态  
    [[ ## feedback ## ]] 无反馈  
    [[ ## formatted_answer ## ]] 112 Main St, Los Angeles, CA 90210  
    [[ ## completed ## ]] 完成

你可以看到，模型验证了输入并确认它是正确的，并返回了一个格式良好的地址，在显示给用户之前确保了格式的一致性。这种结构化的验证有助于通过强制执行标准化格式来保持数据的一致性和完整性。

现在，我们来看看没有使用DSPy的OpenAI Python客户端的情况。下面是直接调用OpenAI的API的验证器代码，绕过了DSPy的声明式结构。

该应用程序允许通过调整环境变量在基于DSPy的验证器和平滑切换到未经过DSPy处理的原始OpenAI Python实现之间进行切换。这使得实验变得简单，您可以轻松地进行尝试，观察DSPy的结构化方法如何影响验证、格式化以及整体预测能力。

    import openai  
    import guardrails as gd  
    import json  
    from typing import Dict  
    from validation.base_validator import BaseValidator  
    from validation.validated_response import ValidatedLLMResponse  
    from helpers.config import OPENAI_API_KEY, MLFLOW_ENABLED, MLFLOW_EXPERIMENT_NAME  
    import mlflow  

    if MLFLOW_ENABLED:  
        mlflow.openai.autolog()  

    guard = gd.Guard.for_pydantic(ValidatedLLMResponse)  

    class ChatGPTValidator(BaseValidator):  
        """基于ChatGPT的验证策略实现。"""  

        def validate(self, question: str, user_answer: str) -> Dict[str, str]:  
            """使用OpenAI的ChatGPT来验证回答。"""  
            client = openai.OpenAI(api_key=OPENAI_API_KEY)  

            response = client.chat.completions.create(  
                model="gpt-3.5-turbo",  
                messages=[  
                    {  
                        "role": "system",  
                        "content": (  
                            "您是乐于助人的助手，负责验证用户的回答。"  
                            "您必须用JSON格式回复，明确验证状态。"  
                            "如果回答有效，返回 {'status': 'valid', 'feedback': '<反馈信息>', 'formatted_answer': '<格式化后的回答>'}，"  
                            "如果回答需要澄清，返回 {'status': 'clarify', 'feedback': '<澄清信息>', 'formatted_answer': '<原始回答>'}，"  
                            "确保正确的格式化：电子邮件地址小写，姓名首字母大写，电话号码和地址标准化。"  
                        ),  
                    },  
                    {  
                        "role": "user",  
                        "content": f"问题: {question}\n用户回答: {user_answer}\n请验证此回答。",  
                    },  
                ],  
                response_format={"type": "json_object"},  
            )  

            try:  
                validation_str = response.choices[0].message.content.strip()  
                validation_result = json.loads(validation_str)  
                print(validation_result)  

                # 应用Guardrails AI  
                validated_result = guard.parse(json.dumps(validation_result))  
                validated_dict = validated_result.validated_output  
                print(validated_dict)  
            except (json.JSONDecodeError, KeyError):  
                validation_result = {  
                    "status": "error",  
                    "feedback": "处理验证响应时出现错误。",  
                    "formatted_answer": user_answer,  
                }  

            if MLFLOW_ENABLED:  
                mlflow.set_experiment(MLFLOW_EXPERIMENT_NAME)  
                with mlflow.start_run(nested=True) as run:  
                    mlflow.log_param("validation_engine", "ChatGPT")  
                    mlflow.log_param("question", question)  
                    mlflow.log_param("input_answer", user_answer)  
                    mlflow.log_param("status", validated_dict.get("status", "error"))  
                    mlflow.log_param(  
                        "feedback", validated_dict.get("feedback", "没有反馈")  
                    )  
                    mlflow.log_param(  
                        "formatted_answer",  
                        validated_dict.get("formatted_answer", user_answer),  
                    )  

            return validated_dict

与使用声明式的签名进行验证的DSPy实现不同，与之相反，OpenAI的Python客户端需要在API调用中明确指定系统和用户角色。然而，它缺少可以在Python代码中直接定义验证逻辑的结构化方式。

如果没有DSPy，验证逻辑必须手动编写在提示中，而不是通过输入输出字段来程序化定义。这使得保持一致、调试问题更困难，并且集成额外的验证层，如Guardrails AI也更加困难。

为了展示差异，这里是使用OpenAI的Python客户端发送给gpt-3.5-turbo模型的实际发送的内容。请注意，提示结构是手工设计的，而不是像在DSPy中那样通过编程自动实现。

    [  
      {  
        "role": "system",  
        "content": "你是一个乐于助人的助手，帮助验证用户的回答。你必须以JSON格式回复，明确验证状态。如果回答有效，返回：{'status': 'valid', 'feedback': '<反馈消息>', 'formatted_answer': '<格式化的回答>'}。如果需要澄清，返回：{'status': 'clarify', 'feedback': '<澄清信息>', 'formatted_answer': '<原始回答>'}。确保电子邮件小写，姓名首字母大写，电话号码和地址格式正确。"  
      },  
      {  
        "role": "user",  
        "content": "问题：你的地址是什么？\n用户回答：112 Main Street, Los Angeles, CA 90210\n验证这个回答。"  
      }  
    ]

    [  
      {  
        "finish_reason": "stop",  
        "index": 0,  
        "logprobs": null,  
        "message": {  
          "content": "{\n\t\"状态\": \"有效\",\n\t\"反馈\": \"地址有效。\",\n\t\"格式化答案\": \"112 Main St, 洛杉矶, 加利福尼亚州 90210\"\n}",  
          "拒绝": null,  
          "角色": "助手",  
          "音频": null,  
          "函数调用": null,  
          "工具调用": null  
        }  
      }  
    ]

在这个例子里面，两个验证器都返回了相同的响应：输入被验证为有效，且正确格式的地址是"112 Main St, Los Angeles, CA 90210"。这是因为第二个验证器的提示经过多次调整才变得结构良好。

虽然在这种简单的情况下，OpenAI 的 Python 客户端同样工作得很好，但在更复杂的场景中，DSPy 的结构化方法则更为突出。它允许开发人员明确地定义期望的输入输出结构，确保在多个请求中的结构一致。此外，如果模型的输出不符合预期或模棱两可，DSPy 可以自动重试请求或应用内置的启发式方法来优化响应以使其更加明确 —— 从而减少手动后处理的需要。

想了解更多内容，请访问DSPy文档或查看启发了这项工作的研究论文。

护栏AI对GPT-3.5-turbo模型生成的输出进行结构化验证和安全约束，确保响应在到达应用程序之前符合预定义的格式、限制和业务逻辑。在此系统中，护栏AI会检查关键输出字段，包括状态、反馈和格式化回答，以确保这些关键字段符合要求。

• 电子邮件地址应全部小写
• 电话号码应采用标准格式（XXX） XXXX-XXXX
• 地址应包含所有必要信息（街道、城市、省份/州、邮政编码）

这是验证层的关键所在，特别是在高风险的应用场合中，错误的、不完整的或格式不正确的信息可能会导致数据的完整性受损。

• 监管和合规风险
• 操作效率低下（例如：交易失败、错误记录）
• 用户体验不佳和失去信任

通过将Guardrails AI与DSPy结合，系统可以同时受益于结构化验证和智能重试机制。这确保了错误可以被早期发现，域特定规则得到执行，虚构或格式错误的输出被阻止进一步传播——从而增强了AI驱动决策的整体可靠性和可信度。

以下是一个由Guardrails AI用来执行输出验证的Pydantic类。

    from pydantic import BaseModel, Field, field_validator  
    import re  

    class ValidatedLLMResponse(BaseModel):  
        """使用Guardrails AI和Pydantic来验证并格式化用户响应。"""  

        status: str = Field(..., pattern="^(valid|clarify|error)$")  
        feedback: str  
        formatted_answer: str  

        @field_validator("formatted_answer", mode="before")  
        @classmethod  
        def validate_and_format(cls, value, values):  
            """根据问题类型格式化和验证响应内容."""  

            if values.get("status") == "error":  
                return value  # 对错误跳过验证步骤  

            question = values.get("question", "").lower()  

            # 验证电子邮件格式  
            if "email" in question:  
                return cls.validate_email(value)  

            # 将姓和名的首字母大写  
            if "name" in question:  
                return cls.validate_name(value)  

            # 验证并格式化电话号码为(XXX) XXX-XXXX  
            if "phone" in question:  
                return cls.validate_phone(value)  

            # 确保地址包含街道、城市、州和邮政编码，并正确格式化地址  
            if "address" in question:  
                return cls.validate_address(value)  

            return value  # 默认情况下，保持原样返回  

        @staticmethod  
        def validate_email(email: str) -> str:  
            """验证电子邮件格式并将其转换为小写."""  
            return (  
                email.lower() if re.match(r"^[\w\.-]+@[\w\.-]+\.\w+$", email) else "clarify"  
            )  

        @staticmethod  
        def validate_name(name: str) -> str:  
            """将姓和名的首字母大写."""  
            return " ".join(word.capitalize() for word in name.split())  

        @staticmethod  
        def validate_phone(phone: str) -> str:  
            """验证并格式化电话号码为(XXX) XXX-XXXX，移除非数字的字符."""  
            digits = re.sub(r"\D", "", phone)  # 移除非数字的字符  
            return (  
                f"({digits[:3]}) {digits[3:6]}-{digits[6:]}"  
                if len(digits) == 10  
                else "clarify"  
            )  

        @staticmethod  
        def validate_address(address: str) -> str:  
            """确保地址包含街道、城市、州和邮政编码，并正确格式化地址."""  
            components = address.split(",")  
            if len(components) < 3:  
                return "地址信息不完整"  # 地址信息不完整  

            formatted_address = ", ".join(comp.strip().title() for comp in components)  
            return (  
                formatted_address if re.search(r"\d{5}", formatted_address) else "clarify"  
            )

这个 FastAPI 后端作为系统的协调中心，负责处理会话管理、用户输入验证和工作流进展。它暴露了三个主要端点： / start_registration、 / submit_response 和 / edit_field，每个端点都在引导用户完成注册流程中起着关键作用。

/start_registration 端点开始一个新的会话，分配一个唯一的 session_id 用于会话，并开始注册流程。它使用 LangGraph 来确定工作流程中的第一个问题，并将初始状态存储在 SQLite 数据库中。由 RegistrationGraphManager 定义的注册图确定问题的顺序，确保工作流程按逻辑顺序推进。

当用户通过 /submit_response 端点提交响应时，系统从数据库中检索会话状态，使用 validate_user_input 验证输入，并更新存储的响应。验证使用 DSPy 和 Guardrails AI 进行，确保输入符合预期格式和业务规则。如果输入不完整或无效，系统会提示用户进行澄清，而不是直接跳到下一步。工作流随后根据之前的响应动态确定下一个问题，允许自适应流程——例如，根据之前的回答跳过可选步骤。一旦注册完成，系统会返回所收集回答的结构化摘要。

/edit_field端点允许用户修改之前提交的信息。它会获取会话状态，重新验证更新后的信息。如果新值符合要求，它会更新数据库。这样，用户的更正会和最初输入数据时一样，通过相同的验证和格式化逻辑来进行处理。

    from fastapi import FastAPI  
    from fastapi.middleware.cors import CORSMiddleware  
    import uuid  
    import logging  
    from validation.factory import validate_user_input  
    from db.sqlite_db import fetch_session_from_db, upsert_session_to_db, RegistrationState  
    from graph.registration_graph import RegistrationGraphManager  

    app = FastAPI()  

    app.add_middleware(  
        CORSMiddleware,  
        allow_origins=[  
            "http://localhost",  
            "http://localhost:8080",  
            "http://localhost:5173",  
        ],  
        allow_credentials=True,  
        allow_methods=["*"],  
        allow_headers=["*"],  
    )  

    registration_questions = {  
        "ask_email": "请输入您的电子邮件地址？",  
        "ask_name": "请输入您的全名？",  
        "ask_address": "请输入您的地址？",  
        "ask_phone": "请输入您的电话号码？",  
        "ask_username": "请输入一个用户名。",  
        "ask_password": "请输入一个强密码。",  
    }  

    registration_graph = RegistrationGraphManager("registration", registration_questions)  
    registration_graph.generate_mermaid_diagram()  

    @app.post("/start_registration")  
    def start_registration():  
        session_id = str(uuid.uuid4())  

        # 初始状态  
        initial_state: RegistrationState = {  
            "collected_data": {},  
            "current_question": "",  
            "current_node": "ask_email",  
            "session_id": session_id,  
        }  

        # 开始图的执行并获取第一个节点  
        execution = registration_graph.compiled_graph.stream(initial_state)  
        try:  
            steps = list(execution)  # 完全遍历生成器  
            if not steps:  
                raise RuntimeError("生成器在生成任何状态之前退出。")  
            first_step = steps[0]  
        except GeneratorExit:  
            raise RuntimeError("在第一次转换之前检测到生成器退出")  

        # 从第一个节点提取状态  
        first_node_key = list(first_step.keys())[0]  
        first_node_state = first_step[first_node_key]  
        first_node_state["current_node"] = first_node_key  
        first_node_state["session_id"] = session_id  

        # 保存会话  
        upsert_session_to_db(  
            session_id,  
            first_node_state["collected_data"],  
            first_node_state["current_question"],  
            first_node_state["current_node"],  
        )  

        return {  
            "session_id": session_id,  
            "message": first_node_state["current_question"],  
            "state": first_node_state,  
        }  

    @app.post("/submit_response")  
    def submit_response(response: dict):  
        session_id = response.get("session_id")  
        if not session_id:  
            return {"error": "缺少session_id"}  

        current_state = fetch_session_from_db(session_id)  
        if not current_state:  
            return {"error": "会话未找到，请重新开始注册"}  

        skip_steps = response.get("skip_steps", [])  
        for node_key in skip_steps:  
            logging.info(f"跳过 {node_key}")  
            current_state[f"skip_{node_key}"] = True  

        user_answer = response.get("answer", "")  
        current_question = current_state["current_question"]  
        current_node = current_state["current_node"]  

        # 使用dspy对用户的回答进行验证  
        if current_node in skip_steps:  
            # 如果用户跳过了该问题，创建一个假的验证结果  
            validation_result = {  
                "status": "valid",  
                "feedback": "跳过了这个问题",  
                "formatted_answer": "-",  
            }  
            logging.info(f"跳过验证 {current_node}")  
        else:  
            # 正常验证  
            validation_result = validate_user_input(current_question, user_answer)  

            # 如果需要澄清或出错  
            if validation_result["status"] in ("clarify", "error"):  
                return {  
                    "next_question": current_question,  
                    "validation_feedback": validation_result["feedback"],  
                    "user_answer": user_answer,  
                    "formatted_answer": validation_result["formatted_answer"],  
                    "state": current_state,  
                }  

        current_state["collected_data"][current_state["current_node"]] = validation_result[  
            "formatted_answer"  
        ]  

        if "current_node" not in current_state or not current_state.get("collected_data"):  
            return {"error": "会话状态已损坏，请重新开始注册流程"}  

        next_step = registration_graph.resume_and_step_graph(current_state)  

        if not next_step or next_step == {}:  
            # 表示我们到达了END节点或没有更多步骤  
            return {  
                "message": "注册成功完成！",  
                "validation_feedback": validation_result["feedback"],  
                "user_answer": user_answer,  
                "formatted_answer": validation_result["formatted_answer"],  
                "state": current_state,  
                "summary": current_state["collected_data"],  
            }  

        next_node_key = list(next_step.keys())[0]  
        next_node_state = next_step[next_node_key]  
        next_node_state["current_node"] = next_node_key  

        upsert_session_to_db(  
            session_id,  
            current_state["collected_data"],  
            next_node_state["current_question"],  
            next_node_state["current_node"],  
        )  

        return {  
            "next_question": next_node_state["current_question"],  
            "validation_feedback": validation_result["feedback"],  
            "user_answer": user_answer,  
            "formatted_answer": validation_result["formatted_answer"],  
            "state": next_node_state,  
            "summary": current_state["collected_data"],  
        }  

    @app.post("/edit_field")  
    def edit_field(request: dict):  
        session_id = request.get("session_id")  
        if not session_id:  
            return {"error": "缺少session_id"}  

        field_to_edit = request.get("field_to_edit")  
        new_value = request.get("new_value")  

        current_state = fetch_session_from_db(session_id)  
        if not current_state:  
            logging.error("会话未找到，请重新开始注册")  
            return {"error": "找不到会话，请重新开始注册"}  

        question_text = registration_questions.get(field_to_edit)  
        if not question_text:  
            logging.error(f"无效的field_to_edit: {field_to_edit}")  
            return {"error": f"无效的field_to_edit: {field_to_edit}"}  

        validation_result = validate_user_input(  
            question=question_text, user_answer=new_value  
        )  

        if validation_result["status"] == "clarify":  
            return {  
                "message": "需要进一步说明",  
                "validation_feedback": validation_result["feedback"],  
                "raw_answer": new_value,  
                "formatted_answer": validation_result["formatted_answer"],  
            }  

        current_state["collected_data"][field_to_edit] = validation_result[  
            "formatted_answer"  
        ]  

        upsert_session_to_db(  
            session_id,  
            current_state["collected_data"],  
            current_state["current_question"],  
            current_state["current_node"],  
        )  

        return {  
            "message": "字段更新成功！",  
            "validation_feedback": validation_result["feedback"],  
            "raw_answer": new_value,  
            "formatted_answer": validation_result["formatted_answer"],  
            "summary": current_state["collected_data"],  
        }

在整个过程中，MLflow 记录模型的输入和输出，以便监控和复现。通过结合工作流管理、结构化验证、会话持久性和用户纠错功能，这个 FastAPI 实现为增强型结构化对话注册系统提供了基础。

在这篇帖子中，我使用FastAPI、React、LangGraph、DSPy、Guardrails AI和MLflow实现了一个AI驱动的注册系统。该系统引导用户进行结构化的会话，实时验证他们的回复，同时确保数据的准确性和一致性。通过利用AI驱动的验证，应用程序根据用户的输入动态调整，从而提供更流畅、更智能的注册过程体验。

• DSPy 通过定义结构化的输入和输出签名，支持声明式的验证方法。它应用内置的格式化启发式算法，并进行必要的重试，使 AI 的验证既稳健又优雅。
• Guardrails AI 作为一个额外的安全保障层，强制执行模型输出的约束，以防止生成畸形或不完整的响应。通过集成 Pydantic 验证，它确保符合预设的业务规则，从而使 AI 生成的响应既可信又结构化。
• LangGraph 提供灵活的工作流管理，允许根据用户输入动态调整问题顺序，这种灵活的编排确保在必要时可以跳过可选步骤，并始终向用户展示最相关的下一步。

这篇文章展示了一种结构化但又灵活的对话体验。有很多方法可以扩展和增强系统，从而提供更丰富的用户体验，包括设定可量化的标准，系统地评估其可靠性和有效性，以确保其性能。

1. 自动化测试套件和A/B测试： 开发涵盖各种边缘案例的测试用例，并实时比较不同的验证策略，例如DSPy与OpenAI原生验证。
2. 增强个性化： 通过结合用户历史和上下文，系统可以预填答案并提出个性化的后续问题，以提高参与度。
3. 扩展模型能力： 实验检索增强生成（RAG）可以提高响应的准确性并确保符合特定领域的规范要求。
4. 多模态能力： 添加对语音输入的支持可以将注册流程扩展到文本交互之外。
5. 实时分析和监控： 更广泛地利用MLflow等模型监控工具可以提供有关模型漂移、用户行为和验证失败的见解，从而帮助优化系统性能。

通过整合结构化的AI验证、工作流程调度和实时监控，这个系统为可靠且可扩展的AI驱动的登记和数据收集打下了基础。随着AI驱动的对话体验不断进步，完善这些技术对于在金融、医疗和企业工作流等高风险领域平衡自动化、准确性以及用户信任来说非常重要。

希望你会喜欢这篇文章，这里是代码库，在这个链接里。