利用LangGraph AI助手在FastAPI中实现交易记录的自动化录入。

任何尝试通过网页表单记录日常开支的人都知道这有多繁琐。购买一杯5美元的咖啡本不必填写多个字段，但在许多使用FastAPI和PostgreSQL开发的财务追踪应用中，无论金额多小，每笔交易都必须手动输入。

试想开发这样的应用——无论交易简单还是复杂，都得通过表单来完成。偶尔记录几笔交易还好，但一旦需要一次性记录多笔交易，就会变得极其耗时。

这时自然会产生一个疑问：如果整个流程能够自动化该多好？如果不必填写表单，只需告诉助手“我今天在餐厅花了5美元买咖啡”，让系统处理其余部分，那该多方便？

这就是LangGraph发挥作用的地方。

使用 LangGraph，你可以构建一个人工智能助手，它能够接收用通俗语言描述的事件，并为你将其转换为格式规范的交易记录。

简介

本文将介绍 LangGraph 及其相关生态系统的核心概念，随后深入讲解如何利用 LangGraph 为 FastAPI 应用添加人工智能助手以实现交易记录的自动化处理。

LangGraph

LangGraph 由 LangChain 的开发团队打造，是一款开源工具包，可通过图结构来构建和管理人工智能智能体工作流。借助它，你可以将某个流程描述为由“节点”和“边”组成的结构，从而使复杂的智能体行为更加有序、可扩展且易于控制。

在深入了解 LangGraph 之前，先理解 LangChain 是很有帮助的，因为 LangGraph 是建立在 LangChain 之上的。

LangChain

LangChain 也是一个开源工具包，用于构建基于大型语言模型的应用程序。它的核心作用是为开发者搭建起大型语言模型与外部资源——如数据源、工具以及工作流步骤——之间的桥梁，从而使系统能够进行多步推理和自动化任务处理，而不仅仅是简单的提示-响应交互。

用途：它专为需要串联多个步骤来构建 AI 应用程序而设计——例如处理用户输入、检索相关信息并生成响应。

结构：LangChain依赖于“链”，即按顺序排列的操作序列，每一步的输出会成为下一步的输入。这有助于将复杂的逻辑分解为更小、更易管理的部分。

应用场景：常见的应用包括聊天机器人、多步骤推理任务、文档检索与总结，以及将大型语言模型与外部工具或API相连。

LangGraph（续）

简而言之，LangGraph将大型语言模型的调用组织成图形化的工作流，从而支持灵活甚至并行的多步骤推理，而非严格的线性顺序。

目的：它能够实现逻辑可以分支、循环或并行执行步骤的AI应用，超越了简单顺序链所能表达的能力。

结构：LangGraph将操作表示为“节点”，而节点之间的数据流则表现为“边”。单个节点的输出可以供给多个下游节点，从而实现动态的决策路径。

应用场景：LangGraph非常适合协调多个智能体、构建复杂的决策流程、自动化处理涉及条件逻辑的任务，以及同时管理多个大型语言模型或工具。

LangGraph中的图是什么？

一般来说，图是一种非线性数据结构，由“顶点”（节点）和“边”（节点之间的连接）组成，用于体现对象之间的关系。

在 LangGraph 中，这种图结构被用于构建具有状态且呈循环性的工作流——在这种工作流中，AI 可以根据中间结果做出决策、回退到早期步骤或分支到不同的路径。

LangChain 与 LangGraph 的对比

项目架构

（FastAPI 接口 + LangGraph + 事务创建与持久化）

存在的问题

在将 LangGraph 引入金融应用之前，创建事务意味着直接调用 /transactions/add 接口，其请求数据格式如下：

{
    title: "Coffee for Rosy",
    type: "expense",
    amount: 5,
    note: "Paid $5 to Rosy for coffee",
    category_id: "5bc22126-5982-4500-9e74-71c9c089f0c8",
    payment_option_id: "07c5d180-fa4d-4435-aa04-b54ef436eca1"
}

要达到那个状态，需要先进行两次API调用——一次用于获取categories列表，另一次用于获取payment_options——仅仅是为了获得构造请求数据所需的ID。换句话说，创建一笔交易实际上是一个分为三步的过程，而且速度还很慢。

解决方案

解决办法是让AI助手处理全部三个步骤，而用户只需用通俗的语言描述自己如何使用资金即可。基于这一目标，以下是该系统的实现结构。

三步架构

1. FastAPI接口
2. LangGraph调度器
3. PostgreSQL数据库

1. FastAPI接口

用户向 FastAPI 接口 /assistance/transaction-entry 发送请求，请求体中包含描述该交易的消息。

{
  message: "Sent $5 to Rosy for Coffee through cash."
}

2. LangGraph 调度器

该调度器以图的形式构建，其中每个节点代表一个操作，每条边代表操作之间的数据流。

第一个节点即分析型大语言模型，它会接收用户的消息，并检查其中是否包含记录交易所需的所有信息——无论是收入还是支出、涉及金额、交易目的、使用的支付方式等等。

如果消息已包含所有必要细节，大语言模型会将其转换为结构化数据，例如：

{
    title: "Coffee",
    transaction_type: "expense",
    amount: 5.0,
    note: "Sent $5 to Rosy for coffee through cash",
    category: "coffee",
    payment_option: "cash",
    payment_type: "Cash",
    is_complete: True,  // Flag
    missing_info_message: None  // Flag
}

这些结构化数据在移除两个标志字段（is_complete和missing_info_message）后会传递到数据库写入节点。该节点会调用create_transaction()方法，从而在数据库中为用户记录相关交易信息。

但如果消息缺少某些细节会怎样呢？以这样的消息为例：

{
  message: "Sent $5 to Rosy for Coffee." // payment mode is not specified
}

此处未指定支付方式。在这种情况下，大型语言模型提取的数据将包含已设置的标志值，例如：

{
    title: "Coffee",
    transaction_type: "expense",
    amount: 5.0,
    note: "Sent $5 to Rosy for coffee",
    category: "coffee",
    payment_option: None,
    payment_type: None,
    is_complete: False,  // Flag
    missing_info_message: "Please enter the payment mode used for this expense." // Flag
}

由于这里的is_complete标志值为False，因此missing_info_message会被转发到与分析型大型语言模型相连的另一个节点——澄清节点。只有当is_complete的值为False时才会触发这条路径。

Clarification节点接收到missing_info_message后，会调用ask_again()方法，该方法会将该消息作为响应返回给最初的FastAPI请求。这就标志着此次运行中图表的执行结束——用户收到的结果仅是一个提示，要求补充缺失的详细信息，在此例中即为支付方式。

假设用户随后回复了缺失的信息，例如：

{
  message: "UPI"
}

这样的回复会导致协调器图表被重新初始化，然后按照与之前相同的步骤顺序执行。

第二次处理的关键区别在于之前的所有信息都不会丢失——每次大语言模型提取数据时都会保留对话历史（本文后面会更详细地介绍这种持久化机制）。由于现在有了payment_option，并且可以将其与之前获取的值结合，is_complete的值就会变为True，最终经过筛选的数据会被传递给数据库写入节点，其形式如下：

{
    title: "Coffee",
    transaction_type: "expense",
    amount: 5.0,
    note: "Sent $5 to Rosy for coffee through cash",
    category: "coffee",
    payment_option: "UPI",
    payment_type: "Digital"
}
// Flags removed.

数据库写入节点随后会使用这些经过过滤的数据来继续处理。需要记住，当手动创建交易记录时，必须先进行两次额外的API调用——一次用于查询categories，另一次用于查询payment_options——才能在创建交易记录之前获取对应的ID。这里也存在同样的问题：过滤后的数据中包含的是类别和支付选项的实际文本值，而非它们的数据库ID，而数据库不会接受这些字段的原始值。

为了解决这个问题，数据库写入节点必须根据过滤数据中的值查询数据库，以找到对应的类别和支付选项记录。

由于该项目同时使用了FastAPI和SQLAlchemy，这些查询是通过SQLAlchemy语句来实现的。

关于categories：

from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError

# Run a select query to check if the category in data.category exists or not.
stmt = select(CategoriesModel).where(
  CategoriesModel.user_id == user_id,
  func.lower(getattr(CategoriesModel, name)) == data.category.lower()
)

# Execute the query.
result = await session.execute(stmt)

# If category exists, assign its ID to data.category.
existing_category = resule.scalar_one_or_none()
if existing_category:
  data.category = existing_category.id

# If category doens't exits, create a new category and save it to database.
new_category = CategoriesModel(**{name: data.category, "user_id": user_id})
session.add(new_row)

try:
  await session.flush()
except IntegrityError:
  # In case another concurrent request created it first,
  # we need to roll back and fetch it again.
  await session.rollback()

  result = await session.execute(stmt)
  existing_category = result.scalar_one_or_none()
  if existing_category:
      data.category = existing_category.id
  raise

简而言之，其逻辑如下： 执行查询以检查data.category中引用的类别是否已存在。 如果存在，则用该类别的ID替换data.category中的值。 如果不存在，则创建新的类别记录，并使用其新生成的ID。

payment_options也遵循相同的模式：

from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError

# Run a select query to check if the peyment_option in data.payment_option exists or not.
stmt = select(PaymentOptionsModel).where(
  PaymentOptionsModel.user_id == user_id,
  func.lower(getattr(PaymentOptionsModel, name)) == data.payment_option.lower()
)

# Execute the query.
result = await session.execute(stmt)

# If payment_option exists, assign its ID to data.payment_option.
existing_option = resule.scalar_one_or_none()
if existing_option:
  data.payment_option = existing_option.id

# If payment_option doesn't exits, create a new payment_option and save it to database.
new_option = PaymentOptionsModel(**{name: data.payment_option, "user_id": user_id})
session.add(new_row)

try:
  await session.flush()
except IntegrityError:
  # In case another concurrent request created it first,
  # we need to roll back and fetch it again.
  await session.rollback()

  result = await session.execute(stmt)
  existing_option = result.scalar_one_or_none()
  if existing_option:
      data.payment_option = existing_option.id
  raise

一旦类别ID和支付选项ID都确定后，数据对象就会完成更新并准备好被插入，其结构如下：

{
    title: "Coffee",
    type: "expense",
    amount: 5.0,
    note: "Sent $5 to Rosy for coffee through cash",
    category_id: "5bc22126-5982-4500-9e74-71c9c089f0c8",
    payment_option_id: "07c5d180-fa4d-4435-aa04-b54ef436eca1"
}

使用这些已确定的 数据后，数据库写入节点会调用create_transaction()方法，该方法会将事务记录永久保存到数据库中。

3. PostgreSQL数据库

这代表了该架构的最终阶段，此时由数据库写入节点处理的最终数据会被写入transactions表中。

transactions表的最终结构如下：

实现方式

在介绍了架构之后，现在就来详细讲解如何使用LangGraph构建这个协调器。需要注意的是，这里的实现顺序与上述架构介绍的顺序并不完全一致，实际的实现结构如下：

1. LangGraph协调器
2. PostgreSQL数据库
3. FastAPI接口

1. LangGraph协调器

该调度器本身位于 src/assistance/graph.py 文件中。此文件负责初始化大语言模型，定义图中的节点，连接这些节点之间的关系，最终将所有内容编译成一个可运行的图结构。

如前所述，这个调度器由三个节点组成：分析器大语言模型、数据库写入器以及澄清节点。

分析器大语言模型节点（Groq）

该节点实际上是一个语言模型，其任务是理解用户的意图，并确认消息中是否包含所有必要且正确的信息。该项目没有从零开始构建自定义模型，而是依赖 Groq 来处理这些复杂任务。

什么是 Groq？

Groq 是一个专为处理图结构数据而设计的开源 Python 框架。它为开发者提供了强大的方式来查询、过滤和聚合以图形式存储的信息，非常适合处理大型图数据集，比如社交网络、知识图谱或推荐系统，正如 GeekForGeeks 关于 Groq API 的文章中所描述的那样。

通过 Groq 的托管 API，你可以向广泛使用的开源模型发送提示词——本项目中使用的就是 openai/gpt-oss-120b——并且能获得比其他提供类似模型的服务更快得多的响应速度。

为何选择 Groq？

选择 Groq 而非 ChatOpenAI 或 ChatAnthropic 等替代方案有以下几个原因：

• 速度：Groq 采用专为语言处理设计的硬件——LPU（语言处理单元），而非其他大多数服务提供商所依赖的 GPU，因此推理速度更快。
• 实用的免费套餐：Groq 的免费套餐十分充裕，足以支持个人或学习型项目，在实验过程中不会产生过高的 API 费用。
• 通过 LangChain 实现无缝兼容：langchain_groq.ChatGroq 类能够像 ChatOpenAI 或 ChatAnthropic 一样与 LangChain 和 LangGraph 相集成。这意味着日后更换服务提供商时无需重新设计图结构逻辑，只需替换客户端即可。

如何获取 Groq API 密钥

Groq 允许用户免费生成用于开发的 API 密钥。获取方法如下：

1. 访问 https://console.groq.com，然后登录或注册。
2. 从导航栏中选择 API 密钥选项。
3. 点击“创建 API 密钥”。
4. 系统会显示一个表单，要求输入密钥名称（本项目使用的是 transaction-assistant）以及密钥的有效期。填写完成后提交表单。
5. 密钥仅在创建后立即显示一次，因此请务必立即复制它。

创建完成后，所有密钥都会显示在该页面的主列表中。

在 FastAPI 代码中使用 Groq API 密钥

将 Groq API 密钥与其他环境变量一起添加到项目根目录下的 .env 文件中：

GROQ_API_KEY = "gsk_***************************************DyxM"

有几种将环境变量加载到需要它们的模块中的方法。该项目使用了一个专门的设置类：

在src/utils/settings.py中定义一个Settings类：

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    # Configure connection with the .env file
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    # ... Other Variables ...
    GROQ_API_KEY: str


settings = Settings()

然后在需要使用该设置类的地方导入它：

from src.utils.settings import settings

# After importing, the object settings can be used as
# "settings.GROQ_API_KEY" to access the environment variable for Groq API Key.

LLM配置

在配置LLM之前，先安装LangGraph、LangChain以及Groq集成模块： pip install -U langgraph langchain langchain-groq

接着创建一个Groq客户端实例，并为其配置特定的模型：

from langchain_groq import ChatGroq
from src.assitance.schema import ExtractedTransactionSchema
from src.utils.settings import settings

assistance_llm = ChatGroq(model="openai/gpt-oss-120b", temperature=0.2,
                          api_key=settings.GROQ_API_KEY)

structured_llm = assistance_llm.with_structured_output(
    ExtractedTransactionSchema)

在这里，ChatGroq充当了LangChain对Groq聊天模型的封装，使得可以通过LangChain的标准接口与这些模型交互，而无需手动构造HTTP请求。

assistance_llm = ChatGroq(model="openai/gpt-oss-120b", temperature=0.2,
                          api_key=settings.GROQ_API_KEY)

这段代码构建了上文提到的Groq客户端实例，该实例配置了选定的模型及较低的温度值，并使用从环境设置中获取的API密钥进行身份验证。

温度是一个参数，通常取值范围在0到1之间，用于控制模型响应的随机性或创造性。较高的数值（如0.8）会使输出结果更加多样且富有创意，而较低的数值（如0.2）则会让响应更为固定、更易预测。本项目将temperature = 0.2。

structured_llm = assistance_llm.with_structured_output(
    ExtractedTransactionSchema)

这段代码对大语言模型进行了封装，使其不再返回纯文本，而是生成一个完全符合ExtractedTransactionSchema标准的Python对象。其实现方式是让模型生成符合该架构的输出，然后自动解析并验证该输出——从而无需手动解读模型生成的原始文本。

ExtractedTransactionSchema本身定义在src/assistance/schema.py中：

from typing import Optional
from pydantic import BaseModel, Field


class ExtractedTransactionSchema(BaseModel):
    is_complete: bool = Field(
        description="True only if title, type, amount, category, and payment method were all found.")
    missing_info_message: Optional[str] = Field(
        default=None, description="A polite clarifying question listing listing exactly what's missing. Must be null if is_complete is True")
    title: str
    transaction_type: str = Field(description="'income' or 'expense'")
    amount: float
    category: str
    payment_option: str = Field(
        description="e.g. 'UPI', 'Cash', 'HDFC Credit Card'")
    payment_type: str = Field(
        description="Broad classification of the payment_option, one of: 'Cash', 'Card', 'Digital', 'Bank Transfer', 'Other'"
    )
    note: str

请注意，此时大语言模型实际上尚未被调用——这一步仅用于规定在模型被调用后输出应具备的格式。

图状态

图状态代表了在图中流动并不断更新的数据结构。可以将其视为指挥家的工作记忆：在执行每一步时，它都会存储图所跟踪和修改的所有信息。对于这个事务助手而言，图状态的定义如下：

from pydantic import BaseModel, Field
from typing import Annotated, List, Optional
import operator
from src.assitance.schema import ExtractedTransactionSchema

class GraphState(BaseModel):
    user_input: str = Field(description="The user input to the graph.")
    conversation_history: Annotated[List[str], operator.add] = []
    extracted: Optional[ExtractedTransactionSchema] = None
    final_response: Optional[str] = None

让我们来详细分析这段代码的实际功能：

from pydantic import BaseModel, Field

这里使用的是数据验证库Pydantic。BaseModel是你在定义类似GraphState这样的结构化、类型检查过的对象时需要继承的父类。Field则允许你为每个属性添加元数据——如描述、默认值等。

from typing import Annotated, List, Optional
import operator

这些都是 Python 的类型标注工具。Optional 表示某个字段可能为空并包含 None 值。List 用于将属性标记为项目列表。Annotated 结合 operator.add 使用，可告知 LangGraph “当节点为该字段返回新值时，应将其追加到现有内容中而非直接替换”。 正是这一机制使得 conversation_history 能在多轮对话中不断积累，而不会因每条新消息的到达而被清空。

class GraphState(BaseModel):
    user_input: str = Field(description="The user input to the graph.")
    conversation_history: Annotated[List[str], operator.add] = []
    extracted: Optional[ExtractedTransactionSchema] = None
    final_response: Optional[str] = None

• user_input：本次调用中用户最新发送的消息。
• conversation_history：所有历史消息的完整记录，按对话轮次逐步积累而非被覆盖。

extracted：在大型语言模型从对话中提取出结构化交易数据后会被填充。由于在运行开始时尚未提取任何内容，该字段初始值为None。

final_response：最终返回给用户的消息——可能是确认交易已记录的回复，也可能是要求提供更多细节的后续问题。

提取提示词

from langchain_core.prompts import PromptTemplate

EXTRACTION_PROMPT = PromptTemplate(
    template="""
        You are a financial assistant extracting transaction details.

        Below is the conversation so far (it may span multiple messages, where later
        messages answer questions raised by earlier ones). Treat it as one combined input.

        Required fields: title, transaction_type (income/expense), amount, category, payment_option.

        If title is missing, add one based on the context of the message.
        If anything required is missing, except title, set is_complete to False and write a short, polite
        clarifying question in missing_info_message asking only for what's missing.

        If everything is present, set is_complete to True, leave missing_info_message null,
        and fill in all fields. Always copy the user's original message into `note`.

        Conversation so far:
        {user_input}
        """,
    input_variables=["user_input"]
)

这是以自然语言形式提供给大语言模型的指令——它明确了需要查找哪些字段、遇到缺失信息时该如何处理，以及响应应如何组织结构。由于structured_llm已在输出层面强制执行架构规范，因此提示语的主要作用是引导模型的推理过程：决定“完整”意味着什么、如何表述澄清性问题等等，而格式化工作则由架构规范负责。

提取器

def extractor(state: GraphState):
    full_conversation = "\n".join(
        state.conversation_history + [state.user_input])

    prompt = EXTRACTION_PROMPT.format(user_input=full_conversation)
    result: ExtractedTransactionSchema = structured_llm.invoke(prompt)

    return {"extracted": result, "conversation_history": [state.user_input]}

extractor函数的功能如下：

• 将所有之前的消息与当前消息合并，以便大语言模型能够看到完整上下文。
• 将合并后的文本传递给大语言模型。
• 接收返回的结构化ExtractedTransactionSchema对象。

返回一个状态更新字典——包括新提取的数据以及当前消息，由于为该字段配置了operator.add行为，LangGraph会自动将这些内容纳入历史记录中。

决策过程

route_after_extraction

def route_after_extraction(state: GraphState):
    return "create_transaction" if state.extracted.is_complete else "ask_again"

该函数并不执行任何实际处理——它的唯一任务就是做出决策。根据大语言模型是否将提取的数据标记为完整，它会返回一个字符串，指示图结构接下来应运行哪个节点。可以将其视为流程图中的分支逻辑：图结构会检查该函数的返回值，并按照相应的路径继续执行，要么前往create_transaction记录交易，要么前往ask_again请求更多信息。

数据库写入节点

create_transaction_node

该节点负责将已完成的事务数据写入数据库，并关联到相应的用户。实现DB Writer节点的create_transaction_node函数如下：

from langchain_core.runnables import RunnableConfig
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession
from src.transaction import controller
from src.transaction.schema import TransactionCreateSchema
from src.utils.db_helper import get_or_create
from src.categories.models import CategoriesModel
from src.categories.controller import get_deterministic_color
from src.payment_options.models import PaymentOptionsModel
from src.utils.db_helper import get_or_create

async def create_transaction_node(state: GraphState, config: RunnableConfig):
    session: AsyncSession = config["configurable"]["session"]
    user = config["configurable"]["user"]
    data = state.extracted

    try:
        category_id = await get_or_create(
            session, CategoriesModel, user.id, data.category,
            extra_defaults={"color": get_deterministic_color(data.category)}
        )
        payment_option_id = await get_or_create(
            session, PaymentOptionsModel, user.id, data.payment_option,
            extra_defaults={"payment_type": data.payment_type}
        )

        payload = TransactionCreateSchema(
            amount=data.amount,
            category_id=category_id,
            payment_option_id=payment_option_id,
            note=data.note,
            title=data.title,
            type=data.transaction_type,
        )

        await controller.create_transaction(payload, session, user)
        await session.commit()

    except SQLAlchemyError as err:
        await session.rollback()
        print(
            f"Error while creating transaction through AI assistance :: {err}")
        return {
            "final_response": "Something went wrong while saving your transaction. Please try again."
        }

    message = f"Added {data.transaction_type} of {data.amount} under '{data.category}' ({data.payment_option})"
    return {"final_response": message}

这些内容一次性看下来有点复杂，我们逐部分来了解。

from langchain_core.runnables import RunnableConfig

这是一种用于表示传递给任何节点的config对象的类型。它仅作为类型提示存在，因此任何看到create_transaction_node函数签名的人都能立刻明白config的格式。

from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession

这些是处理数据库错误以及为用于与PostgreSQL通信的异步数据库会话添加类型标注所需的常规SQLAlchemy导入项。

from src.transaction import controller
from src.transaction.schema import TransactionCreateSchema

这会引入应用中其他地方已使用的交易创建逻辑及其输入架构。通过重用该逻辑，助手可以通过与常规CRUD API完全相同的代码路径来创建交易，而无需在此处重复实现该逻辑。

from src.utils.db_helper import get_or_create
from src.categories.models import CategoriesModel
from src.categories.controller import get_deterministic_color
from src.payment_options.models import PaymentOptionsModel

这些是用于将大语言模型提取的类别和支付选项名称转换为数据库中的实际行和ID的辅助组件，当这些记录尚不存在时还会创建新记录。

注意：由于categories和payment_options这两种模型本质上需要相同的功能，因此其查找/创建逻辑已被合并到一个通用的辅助函数get_or_create中。

async def create_transaction_node(state: GraphState, config: RunnableConfig):

create_transaction_node函数仅在提取的数据被确认完整后才会执行。它被声明为async类型，因为它需要执行实际的数据库操作；同时它会接收config和state参数，以便能够访问当前的数据库会话以及已登录的用户信息。这两个值来自API路由，而非LLM或对话状态，因为它们属于特定的请求，而非正在进行的对话。

    session: AsyncSession = config["configurable"]["session"]
    user = config["configurable"]["user"]
    data = state.extracted

该函数用于获取会话信息、用户信息以及提取出的交易数据。

    try:
        category_id = await get_or_create(...)
        payment_option_id = await get_or_create(...)

由于大语言模型仅提取了类别和支付方式的名称，如“食品杂货”或“UPI”，而非其数据库编号，因此此步骤会检查当前用户是否存在匹配的记录。若不存在，则创建一条新记录；无论哪种情况，都会返回对应的编号。

        payload = TransactionCreateSchema(...)
        await controller.create_transaction(payload, session, user)
        await session.commit()

有效载荷会按照现有交易创建逻辑所期望的格式进行组装，然后传递给同一个控制器函数，从而复用现有的应用程序逻辑而非重新编写。之后会提交数据库事务以持久化这些更改。

    except SQLAlchemyError as err:
        await session.rollback()
        print(...)
        return {"final_response": "Something went wrong..."}

如果在数据库层出现故障，任何未完成的更改都会被回滚，并返回友好的错误信息，而不会导致请求崩溃。这样可以避免出现诸如新建了分类却没有对应事务的情况。

    message = f"Added {data.transaction_type} of {data.amount} under '{data.category}' ({data.payment_option})"
    return {"final_response": message}

操作成功时，会生成一条便于人类理解的确认信息，并将其作为状态更新返回。

说明节点

ask_again

def ask_again_node(state: GraphState):
    return {"final_response": state.extracted.missing_info_message}

这是一个简单的备用路径。如前所述，只有当提取到的数据的is_complete值为False，且missing_info_message中包含有用信息时，此节点才会被执行。

在 ask_again_node 函数内部，它会接收到 state 参数，从而可以访问 state.extracted.is_complete 和 state.extracted.missing_info_message。

简而言之，每当缺少信息时，该节点就会直接转发大语言模型在信息提取过程中已经生成的澄清问题，这样用户就能明确知道接下来需要提供什么内容。

构建图结构

from langgraph.graph import StateGraph, START, END
graph_builder = StateGraph(GraphState)

这会创建一个新的图构建器，并告知它图中的每个节点都将从形如 GraphState 的对象中读取数据并向其写入数据。

graph_builder.add_node("extractor", extractor)
graph_builder.add_node("create_transaction", create_transaction_node)
graph_builder.add_node("ask_again", ask_again_node)

每个函数都会在这里作为带名称的节点被注册，本质上就是图中的某个标记步骤。

graph_builder.add_edge(START, "extractor")

这设置了入口点：图的每次执行都是从 extractor 节点开始的。

graph_builder.add_conditional_edges(
    "extractor",
    route_after_extraction,
    {
        "create_transaction": "create_transaction",
        "ask_again": "ask_again",
    },
)

分支判断就在这里进行。当 extractor 执行完毕后，LangGraph 会调用 route_after_extraction 来确定下一步操作。无论它返回的是 create_transaction 还是 ask_again，都会在对应的映射表中查找，该映射表将每个决策字符串与应跳转到的实际节点关联起来。

graph_builder.add_edge("create_transaction", END)
graph_builder.add_edge("ask_again", END)

这两种可能的分支在执行完成后都会终止图的处理流程，最终都会到达 END 状态。

使用内存进行编译

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()
assistance_graph = graph_builder.compile(checkpointer=memory)

调用compile()可将图定义转换为可运行的程序。传递参数checkpointer=memory会启用之前描述的状态持久化机制，这样使用相同的thread_id再次调用该图时，会从上次中断的地方继续执行，而非重新开始。

最终代码

至此，编排层已经完成。以下是最终的文件内容（src/assistance/graph.py）：

from langgraph.graph import StateGraph, START, END
from langchain_groq import ChatGroq
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableConfig
from pydantic import BaseModel, Field
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Annotated, List, Optional
import operator

from src.utils.settings import settings
from src.assitance.schema import ExtractedTransactionSchema
from src.transaction import controller
from src.transaction.schema import TransactionCreateSchema
from src.utils.db_helper import get_or_create
from src.categories.models import CategoriesModel
from src.categories.controller import get_deterministic_color
from src.payment_options.models import PaymentOptionsModel

assistance_llm = ChatGroq(model="openai/gpt-oss-120b", temperature=0.2,
                          api_key=settings.GROQ_API_KEY)

structured_llm = assistance_llm.with_structured_output(
    ExtractedTransactionSchema)


class GraphState(BaseModel):
    user_input: str = Field(description="The user input to the graph.")
    conversation_history: Annotated[List[str], operator.add] = []
    extracted: Optional[ExtractedTransactionSchema] = None
    final_response: Optional[str] = None


EXTRACTION_PROMPT = PromptTemplate(
    template="""
        You are a financial assistant extracting transaction details.

        Below is the conversation so far (it may span multiple messages, where later
        messages answer questions raised by earlier ones). Treat it as one combined input.

        Required fields: title, transaction_type (income/expense), amount, category, payment_option.

        If title is missing, add one based on the context of the message.
        If anything required is missing, except title, set is_complete to False and write a short, polite
        clarifying question in missing_info_message asking only for what's missing.

        If everything is present, set is_complete to True, leave missing_info_message null,
        and fill in all fields. Always copy the user's original message into `note`.

        Conversation so far:
        {user_input}
        """,
    input_variables=["user_input"]
)


def extractor(state: GraphState):
    full_conversation = "\n".join(
        state.conversation_history + [state.user_input])

    prompt = EXTRACTION_PROMPT.format(user_input=full_conversation)
    result: ExtractedTransactionSchema = structured_llm.invoke(prompt)


    return {"extracted": result, "conversation_history": [state.user_input]}


def route_after_extraction(state: GraphState):
    return "create_transaction" if state.extracted.is_complete else "ask_again"


async def create_transaction_node(state: GraphState, config: RunnableConfig):
    session: AsyncSession = config["configurable"]["session"]
    user = config["configurable"]["user"]
    data = state.extracted


    try:
        category_id = await get_or_create(
            session, CategoriesModel, user.id, data.category,
            extra_defaults={"color": get_deterministic_color(data.category)}
        )
        payment_option_id = await get_or_create(
            session, PaymentOptionsModel, user.id, data.payment_option,
            extra_defaults={"payment_type": data.payment_type}
        )

        payload = TransactionCreateSchema(
            amount=data.amount,
            category_id=category_id,
            payment_option_id=payment_option_id,
            note=data.note,
            title=data.title,
            type=data.transaction_type,
        )

        await controller.create_transaction(payload, session, user)
        await session.commit()

    except SQLAlchemyError as err:
        await session.rollback()
        return {
            "final_response": "Something went wrong while saving your transaction. Please try again."
        }

    message = f"Added {data.transaction_type} of {data.amount} under '{data.category}' ({data.payment_option})"
    return {"final_response": message}


def ask_again_node(state: GraphState):
    return {"final_response": state.extracted.missing_info_message}


graph_builder = StateGraph(GraphState)
graph_builder.add_node("extractor", extractor)
graph_builder.add_node("create_transaction", create_transaction_node)
graph_builder.add_node("ask_again", ask_again_node)

graph_builder.add_edge(START, "extractor")
graph_builder.add_conditional_edges(
    "extractor",
    route_after_extraction,
    {
        "create_transaction": "create_transaction",
        "ask_again": "ask_again",
    },
)
graph_builder.add_edge("create_transaction", END)
graph_builder.add_edge("ask_again", END)

memory = MemorySaver()
assistance_graph = graph_builder.compile(checkpointer=memory)

2. PostgreSQL数据库

此阶段的数据库操作已在之前介绍的create_transaction_node函数中处理，最终的交易数据会被写入transactions表中。

3. FastAPI接口

from fastapi import APIRouter, Depends, status
from sqlalchemy.ext.asyncio import AsyncSession

from src.assitance.schema import UserMessageSchema
from src.assitance.graph import assistance_graph
from src.auth.models import UsersModel
from src.utils.db import get_db
from src.utils.auth.authentication import allow_all


assistance_routes = APIRouter(prefix="/assistance")


@assistance_routes.post("/transaction-entry", status_code=status.HTTP_201_CREATED)
async def run_transaction_assistance(payload: UserMessageSchema, session: AsyncSession = Depends(get_db), user: UsersModel = Depends(allow_all)):
    config = {"configurable": {
              "thread_id": str(user.id),
              "session": session,
              "user": user
              }
              }

    result = await assistance_graph.ainvoke(
        {"user_input": payload.message}, config=config)

    return {"response": result["final_response"]}

客户端会调用此端点（/assistance/transaction-entry），并在请求体中包含一段描述交易内容的消息。

让我们来逐一了解各部分的功能。

@assistance_routes.post("/transaction-entry", status_code=status.HTTP_201_CREATED)

这会在/transaction-entry处设置一个POST路由。将status_code=status.HTTP_201_CREATED设置后，FastAPI就会知道在成功时默认返回哪个状态码。201是表示“已创建新资源”的常规代码，此处很适用，因为成功的调用会生成一条新的交易记录。

组装图结构配置：

    config = {
        "configurable": {
            "thread_id": str(user.id),
            "session": session,
            "user": user
        }
    }

这会构建用于传递给图结构调用的config对象。config中包含那些不应存储在持久化对话状态中的、与请求相关的值。

• "thread_id": str(user.id)：此值是LangGraph的检查点器用来确定应检索和更新谁的对话历史的依据。通过以已认证用户的ID作为键，每个用户都会自动拥有一个独立的、持久化的对话线程，这样某个用户未完成的交易记录就永远不会影响到另一个用户。该值会被转换为字符串，因为检查点器期望thread_id为字符串形式，而user.id通常为UUID。
• "session"和"user"：这些参数会被传递过去，以便在图计算中运行的create_transaction_node能够访问当前的数据库会话以及发起请求的用户信息。

调用图计算：

    result = await assistance_graph.ainvoke(
        {"user_input": payload.message}, config=config)

这就是实际触发执行的行。ainvoke是执行图操作的异步对应函数，若使用同步的invoke则会阻塞事件循环，而这在此处非常重要，因为create_transaction_node在内部会执行异步数据库操作。

• 第一个参数{"user_input": payload.message}表示此次运行的起始状态。只需明确提供user_input即可；其余的GraphState字段（conversation_history、extracted、final_response）要么具有默认值，要么会在执行过程中逐步填充。如果现有的thread_id已保存有历史记录，LangGraph会将新输入整合到该存储状态中，而非从头开始。
• config=config会提供上一步准备好的所有内容：用于定位对应状态的thread_id，以及负责数据库写入操作的节点所需的session和user信息。

await 关键字会暂停此协程，直到图结构完成全部执行，因为 ainvoke 返回的协程在结果可用之前必须先进行等待。

作为 result 返回的值即为最终的 GraphState，它以字典形式呈现，反映了图的执行情况，无论是在 create_transaction 处还是 ask_again 处终止。

返回响应：

    return {"response": result["final_response"]}

该路由通过返回一个仅包含最终响应文本的简单字典来结束执行。FastAPI 会负责将其转换为客户端可用的 JSON 数据，生成类似如下的内容：

{ "response": "Added expense of 450 under 'Groceries' (UPI)" }

这段文字与之前在create_transaction_node或ask_again_node中生成的文本完全相同。该路由本身并不关心实际执行了哪条分支，它只是将最终存储在final_response中的内容转发出去。

结论

通过研究这个交易处理助手，我们可以发现教程往往忽略的一个问题：在部署人工智能功能时，难点并不在于向模型发起请求以获取回复，而在于确保该回复在与真实系统交互时能够安全运行。编写提示语才是容易的部分，真正的工程挑战在于设计能够强制生成结构化输出的架构、用于判断是直接存储数据还是需要进一步澄清的条件逻辑，以及能够在多轮对话中正确保持状态机制的方案。

对于这个项目而言，LangGraph之所以是合适的选择，是因为其工作流程需要真正的决策过程，而非简单的从输入到输出的单一处理步骤。如果你的功能只需要线性流程，那么直接调用大型语言模型或使用LangChain链式结构可能更为简单且合适。但一旦人工智能逻辑需要分支处理、保留记忆，或暂停以收集更多信息后再继续执行，基于图的结构就不再显得是多余的复杂性，反而成为建模该流程最合理的方式。

参考资料

相关阅读

• LangChain与LangGraph：链式结构与有状态图结构的抉择 — 了解LangChain的线性构建模块与LangGraph的有状态、分支式工作流之间的差异，以及如何判断哪种更适合您的AI应用。
• 从零开始构建AI智能体：模式、ReAct与LangGraph — 学习AI智能体的核心概念——规划、工具使用、反思以及ReAct模式——并了解LangChain和LangGraph在手动构建AI智能体中的作用。