LangGraph实战：用Python从零构建一个具备自我纠错能力的Agent系统
·
LangGraph实战：用Python从零构建一个具备自我纠错能力的Agent系统
引言
痛点引入
作为一名在AI开发领域摸爬滚打五年的老兵，我最近被团队新来的实习生问了一个扎心但又非常普遍的问题：
“为什么我用LangChain或者AutoGPT写的智能助手，看起来能调用API、能查文档，但一遇到稍微复杂一点的真实业务场景（比如帮公司算报销、给客户整理包含歧义要求的需求文档），就经常‘掉链子’？要么算错数字，要么理解错需求，要么调用API的参数写错，连个报错修正的机会都不给，直接输出结果拉胯。”
当时我翻遍了我们组的知识库，给她看了一堆Prompt Engineering的Trick、Retry Chain的写法，甚至给她看了一些公司内部自研的Agent状态机，但她还是说“太零散了，要么Prompt写得我怀疑人生，要么Retry的次数不够可控，要么状态机的代码复杂到我看不懂怎么扩展”。
这个问题其实不是个例——90%以上的初版Agent项目，都会卡在“不可靠性”这个致命瓶颈上。LLM（大语言模型）虽然强大，但本质上是一个“概率性生成器”：
- 逻辑推理容易跳步或出错：尤其是涉及多步计算、多条件判断、需要严格遵循格式的任务时。
- 幻觉（Hallucination）频发：编造不存在的API、不存在的业务规则、不存在的文档内容。
- 对外部反馈利用不足：很多Agent只是简单地把API调用结果或人类输入“塞”回LLM，没有结构化的反思和修正流程。
- 流程不可控：用简单的Sequential Chain或者AgentExecutor的话，整个流程像“黑盒子”，很难知道Agent现在在哪一步、为什么错了、怎么去干预。
而LangGraph的出现，恰好给我们提供了一把破解这个“不可靠性”黑盒子的钥匙——它是LangChain团队在2024年初推出的一个专门用于构建有状态、多角色、多节点、循环可控的Agent系统的框架，最核心的特点就是：
- 把Agent的流程建模成有向图（Directed Graph）：节点是LLM调用、工具执行、状态更新这些具体的“动作”，边是“动作之间的转移逻辑”（比如“上一步成功了吗？如果成功就结束，失败了就去反思修正”）。
- 内置了状态管理机制：整个流程的所有中间信息（用户输入、之前的思考、工具调用结果、修正历史）都存在一个“全局状态字典”里，任何节点都可以读取和修改。
- 支持循环和条件分支：这是构建“自我纠错”能力的关键——你可以让Agent在同一个“思考-执行-检查-修正”的循环里反复跑，直到满足预设的“成功条件”为止，而且循环次数完全可控。
解决方案概述
在这篇实战教程里，我们完全从零开始（不会依赖LangChain的预定义Agent，只会用到LangGraph的核心API和OpenAI的GPT-4o-mini作为基础LLM——因为便宜且够用），构建一个具备完整自我纠错能力的“业务报销助手”Agent。
这个报销助手会具备以下核心功能：
- 需求理解与结构化提取：从用户的自然语言输入（比如“上周我去北京出差三天，花了高铁票856，酒店1200，打车234，餐费180，请帮我算一下总共能报多少钱，生成一份符合公司格式的报销单草稿”）中，结构化地提取出差日期、出差城市、各项费用的类型和金额。
- 工具调用与业务规则验证：调用我们模拟的“公司报销规则API”（比如“高铁二等座实报实销，一等座只能报二等座的1.2倍；酒店每天上限350；打车每天上限100；餐费每天上限80”），验证各项费用的合规性，并计算出合规的总报销金额。
- 自我反思与多轮修正：
- 如果LLM结构化提取的信息有缺失或错误（比如没写餐费的天数、把酒店金额写成了“12000”），会触发反思节点，让LLM检查自己的提取结果，然后返回给用户追问或者自己修正。
- 如果调用报销规则API时出错（比如参数格式不对、日期不在有效期内），会触发错误处理节点，让LLM分析错误原因，修改参数后重新调用API。
- 如果最终生成的报销单草稿不符合格式要求（比如缺少必填项、金额对齐不对），会触发格式检查节点，让LLM对照模板修正草稿。
- 流程可视化与状态追踪：我们会用到LangGraph的StateGraph打印功能和第三方的Graphviz库，把整个Agent的有向图结构可视化出来，同时在每一步输出时打印当前的全局状态，让你清清楚楚地看到Agent在“想什么”、“做什么”、“为什么这么做”。
最终效果展示（先睹为快）
在正式开始实战之前，我们先看一下这个自我纠错报销助手的完整运行示例，感受一下它的强大之处：
示例1：结构化提取缺失→自我反思→追问用户→重新提取→验证合规→生成正确草稿
👤 用户输入：上周我去北京出差，花了高铁票856，酒店1200，打车234，请帮我算一下报销
🤖 结构化提取节点（第一轮）：
- 提取结果：
{
"business_trip": {
"city": "北京",
"start_date": "202X-XX-XX（缺失）",
"end_date": "202X-XX-XX（缺失）",
"days": "未知（缺失）"
},
"expenses": [
{"type": "high_speed_rail", "amount": 856.0, "details": "二等座？未知（缺失）"},
{"type": "hotel", "amount": 1200.0, "details": "天数未知（缺失），星级未知（默认按三星及以下，每天350上限）"},
{"type": "taxi", "amount": 234.0, "details": "天数未知（缺失）"}
],
"user_needs": "计算报销金额，生成草稿（未明确是否需要符合格式，但默认需要）"
}
🤖 反思节点（第一轮）：
- 检查结论：提取结果存在3个关键缺失项：出差日期、天数、各项费用的关联天数/舱位/酒店星级
- 修正方案：生成追问话术，向用户确认这些信息
🤖 追问节点：
为了帮您准确计算报销并生成合规的草稿，请您补充以下信息：
1. 出差的具体开始日期和结束日期（比如2024-05-20到2024-05-22）
2. 高铁票的舱位（一等座/二等座/商务座）
3. 酒店的住宿天数和是否为三星及以下
4. 打车费是几天的总费用
👤 用户补充：哦对，是2024-05-20到2024-05-22，三天，高铁二等座，酒店三星住了三天，打车是三天的
🤖 结构化提取节点（第二轮）：
- 提取结果（修正后）：
{
"business_trip": {
"city": "北京",
"start_date": "2024-05-20",
"end_date": "2024-05-22",
"days": 3
},
"expenses": [
{"type": "high_speed_rail", "amount": 856.0, "details": "二等座，实报实销"},
{"type": "hotel", "amount": 1200.0, "details": "三星，住3天，每天上限350"},
{"type": "taxi", "amount": 234.0, "details": "3天总费用，每天上限100"},
{"type": "meal", "amount": 0.0, "details": "用户未提及，默认为0（如有需要请补充）"}
],
"user_needs": "计算报销金额，生成符合公司格式的草稿"
}
🤖 合规检查与API调用节点（第一轮）：
- 调用模拟的公司报销规则API，参数格式正确
- API返回结果：
{
"status": "success",
"total_reimbursement": 2256.0,
"expense_details": [
{"type": "high_speed_rail", "requested": 856.0, "approved": 856.0, "reason": "二等座实报实销"},
{"type": "hotel", "requested": 1200.0, "approved": 1050.0, "reason": "三星酒店3天，每天350上限，1200>1050"},
{"type": "taxi", "requested": 234.0, "approved": 300.0？不对，等下，3天每天100是300，但用户只花了234，实报实销，哦对！API返回修正后的approved是234，reason是“3天总费用234<300上限，实报实销”；哦还有餐费，用户没提，0
],
"format_template": "（省略具体模板，后面会定义）"
}
🤖 报销单生成节点（第一轮）：
- 生成符合格式的草稿
🤖 格式检查节点（第一轮）：
- 检查结论：格式完全符合要求
🤖 最终输出：
# 公司业务报销单草稿
------------------------------
申请人：（未提及，请补充）
部门：（未提及，请补充）
出差日期：2024-05-20 至 2024-05-22
出差城市：北京
出差天数：3天
------------------------------
费用明细：
| 费用类型 | 申请金额（元） | 核定金额（元） | 备注 |
|---------|---------------|---------------|------|
| 高铁票 | 856.00 | 856.00 | 二等座，实报实销 |
| 住宿费 | 1200.00 | 1050.00 | 三星酒店3天，每天350上限 |
| 打车费 | 234.00 | 234.00 | 3天总费用，低于300上限 |
| 餐费 | 0.00 | 0.00 | 未提及 |
------------------------------
总申请金额：2290.00元
总核定金额：2140.00元
------------------------------
备注：请补充申请人、部门信息，如有餐费请补充金额
示例2：工具调用参数错误→错误处理→自我修正参数→重新调用→成功
这个示例我们后面实战到“错误处理节点”的时候再详细展示，现在先让我们进入正式的准备工作！
准备工作
环境/工具
在开始写代码之前，我们需要先配置好以下环境和工具：
- Python版本：≥3.10（因为LangGraph最新版只支持3.10及以上的Python版本，主要是用到了Python 3.10的
match-case语法和typing.Annotated等类型注解的新特性） - 依赖库：
langgraph：核心框架，我们用最新稳定版（截至2024年7月，是0.2.14）langchain-openai：用来调用OpenAI的GPT-4o-mini（或者你也可以用langchain-anthropic调用Claude，或者用Ollama的本地模型，原理都是一样的）python-dotenv：用来管理环境变量（比如你的OpenAI API Key）pydantic：用来定义结构化的数据模型（比普通的Python字典更安全，因为可以做类型检查和数据验证）graphviz：用来可视化LangGraph的有向图结构（可选，但强烈推荐安装，因为能帮你快速理解Agent的流程）
- OpenAI API Key：你需要去OpenAI的官网（https://platform.openai.com/account/api-keys）注册一个账号并生成一个API Key（注意：API Key生成后只会显示一次，一定要保存好；如果你在国内，可能需要用代理或者找一个API中转服务）
环境安装步骤
我们一步一步来配置环境：
- 创建一个新的Python虚拟环境（强烈推荐，避免依赖冲突）：
# 用conda创建（如果你用Anaconda/Miniconda的话） conda create -n self-correcting-agent python=3.11 -y conda activate self-correcting-agent # 或者用venv创建（如果你用纯Python的话） python3.11 -m venv self-correcting-agent # Windows激活 self-correcting-agent\Scripts\activate # Mac/Linux激活 source self-correcting-agent/bin/activate - 安装所有依赖库：
# 安装核心依赖 pip install langgraph==0.2.14 langchain-openai==0.1.20 python-dotenv==1.0.1 pydantic==2.8.2 # 安装可视化依赖（graphviz需要先安装系统级的库，不能只pip install） # Windows系统：去https://graphviz.org/download/下载安装包，安装时记得勾选“Add Graphviz to PATH” # Mac系统：brew install graphviz # Linux系统（Debian/Ubuntu）：sudo apt-get install graphviz # 安装完系统级的库之后，再pip install Python的绑定库 pip install graphviz==0.20.3 - 配置环境变量：
在项目根目录下创建一个名为.env的文件，内容如下：
注意：# OpenAI API配置 OPENAI_API_KEY="你的OpenAI API Key（或者中转服务的Key）" OPENAI_API_BASE="你的OpenAI API Base（如果用中转服务的话，比如https://api.openai-hub.com/v1；如果用官方的话，可以留空或者写https://api.openai.com/v1）" OPENAI_MODEL_NAME="gpt-4o-mini" # 我们用这个模型，因为便宜且推理能力足够强，速度也快 # LangSmith配置（可选，但强烈推荐，用来调试和追踪Agent的运行） LANGCHAIN_TRACING_V2="true" LANGCHAIN_API_KEY="你的LangSmith API Key（去https://smith.langchain.com/注册生成）" LANGCHAIN_PROJECT="self-correcting-reimbursement-agent" # 随便起一个项目名.env文件里包含了你的API Key，千万不要把它提交到Git仓库里！所以你需要在项目根目录下创建一个名为.gitignore的文件，内容如下：.env __pycache__/ *.pyc *.pyo *.pyd *.so *.dll .DS_Store venv/ env/ .venv/
基础知识
在正式开始写代码之前，我们需要先掌握以下几个LangGraph的核心基础概念（这些概念是整个框架的基石，如果你之前没用过LangGraph，一定要认真看这一部分）：
- State（状态）：
状态是LangGraph中最核心的概念——它是一个全局共享的字典（或者Pydantic模型，我们后面会用Pydantic模型，因为更安全），用来存储Agent整个运行过程中的所有中间信息，比如：- 用户的初始输入和后续的补充输入
- LLM的每一轮思考和输出
- 工具调用的参数和结果
- 修正历史
- 当前的循环次数
- 是否满足结束条件的标志
任何节点（Node）都可以从状态中读取信息，也可以修改状态中的信息——而且LangGraph会自动把修改后的状态传递给下一个节点。
- Node（节点）：
节点是Agent流程中的具体“动作单元”——每个节点都是一个Python函数，接收当前的状态作为输入，返回一个字典（或者Pydantic模型的更新）作为输出，用来修改全局状态。
常见的节点类型有：- LLM调用节点：调用大语言模型，从状态中读取上下文信息，生成思考、工具调用参数、或者最终输出。
- 工具执行节点：调用外部工具（比如API、数据库、本地文件系统），从状态中读取工具调用参数，执行工具，把结果写回状态。
- 状态更新节点：不调用LLM或工具，只是单纯地修改状态中的信息（比如增加循环次数、设置结束标志）。
- 反思节点：调用LLM，检查之前的输出或工具调用结果是否有问题，生成反思内容和修正方案。
- 条件判断节点（其实是Edge的一种，后面会讲）：根据当前的状态，决定下一步要走哪个边，去哪个节点。
- Edge（边）：
边是连接节点的**“转移逻辑”**——它决定了Agent在执行完一个节点之后，下一步要去哪个节点。
边分为两种类型：- Normal Edge（普通边）：无条件转移——执行完节点A之后，一定会去节点B。
- Conditional Edge（条件边）：有条件转移——执行完节点A之后，根据当前的状态，决定去节点B、节点C、还是结束流程。
条件边是构建“自我纠错”循环的关键！
- StateGraph（状态图）：
状态图是LangGraph中用来定义整个Agent流程的类——你可以把它想象成一个“画布”，你在上面添加节点、添加边、设置入口节点（第一个要执行的节点）、设置出口节点（流程结束的标志）。
定义好状态图之后，你需要调用它的compile()方法，把它编译成一个可执行的应用（App），然后就可以调用这个App的invoke()、stream()、astream()等方法来运行Agent了。 - Graphviz可视化：
编译好的状态图App有一个get_graph()方法，返回一个Graph对象，你可以调用这个对象的draw_mermaid_png()、draw_png()等方法，把状态图可视化成图片——这对调试和理解Agent的流程非常有帮助！
核心步骤（从零开始构建）
好的，现在我们终于可以开始写代码了！我们会把整个构建过程拆解成以下8个清晰的核心步骤，每个步骤都会有详细的代码解释、原理解释，以及中间结果的展示：
步骤1：定义全局状态（State）
首先，我们需要用Pydantic 2.x来定义Agent的全局状态——为什么用Pydantic而不用普通的Python字典呢？因为：
- 类型安全：Pydantic会自动检查每个字段的类型，如果类型不对，会抛出明确的错误，而不是像普通字典那样，直到运行到某个节点才会出错，而且错误信息很难调试。
- 数据验证：Pydantic可以让你自定义数据验证规则（比如“出差天数必须≥1”、“费用金额必须≥0”），如果验证失败，也会抛出明确的错误。
- 自动序列化/反序列化：Pydantic模型可以自动转换成JSON字典，也可以从JSON字典转换回来，这对LangGraph的状态传递非常友好。
- 更好的IDE支持：IDE（比如PyCharm、VS Code）可以根据Pydantic模型的字段定义，提供自动补全和类型提示，大大提高开发效率。
代码实现
在项目根目录下创建一个名为state.py的文件，内容如下：
from typing import List, Optional, Literal, Annotated
from pydantic import BaseModel, Field, field_validator
from langgraph.graph import add_messages # 这个是LangGraph提供的一个辅助函数，用来自动合并消息列表（我们后面会用到，虽然这个示例主要用结构化状态，但消息列表也可以用来存对话历史）
# ------------------------------------------------------------------------------
# 1. 定义子模型：业务出差信息
# ------------------------------------------------------------------------------
class BusinessTripInfo(BaseModel):
"""业务出差的结构化信息"""
city: str = Field(..., description="出差的城市名称，比如北京、上海、深圳")
start_date: str = Field(..., description="出差的开始日期，格式必须是YYYY-MM-DD，比如2024-05-20")
end_date: str = Field(..., description="出差的结束日期，格式必须是YYYY-MM-DD，比如2024-05-22")
days: int = Field(..., description="出差的天数，必须≥1")
# 自定义数据验证器：检查start_date和end_date的格式是否正确，以及end_date是否≥start_date
@field_validator("start_date", "end_date")
def check_date_format(cls, v):
import re
if not re.match(r"\d{4}-\d{2}-\d{2}", v):
raise ValueError(f"日期格式必须是YYYY-MM-DD，比如2024-05-20，当前输入是：{v}")
return v
@field_validator("end_date")
def check_end_date_after_start_date(cls, v, info):
# info.data是当前模型已经验证过的字段的字典（注意：只有在start_date验证通过之后，info.data里才会有start_date）
if "start_date" in info.data and v < info.data["start_date"]:
raise ValueError(f"结束日期必须≥开始日期，当前开始日期是：{info.data['start_date']}，结束日期是：{v}")
return v
@field_validator("days")
def check_days_positive(cls, v):
if v < 1:
raise ValueError(f"出差天数必须≥1，当前输入是：{v}")
return v
# ------------------------------------------------------------------------------
# 2. 定义子模型：费用明细
# ------------------------------------------------------------------------------
class ExpenseItem(BaseModel):
"""每一项费用的结构化信息"""
type: Literal["high_speed_rail", "flight", "hotel", "taxi", "meal", "other"] = Field(..., description="费用类型，必须是以下几种之一：high_speed_rail（高铁票）、flight（机票）、hotel（住宿费）、taxi（打车费）、meal（餐费）、other（其他费用）")
amount: float = Field(..., description="费用金额，必须≥0")
details: Optional[str] = Field(None, description="费用的详细信息，比如高铁的舱位、酒店的星级和天数、机票的舱位等")
@field_validator("amount")
def check_amount_non_negative(cls, v):
if v < 0:
raise ValueError(f"费用金额必须≥0，当前输入是：{v}")
return v
# ------------------------------------------------------------------------------
# 3. 定义子模型：反思结果
# ------------------------------------------------------------------------------
class ReflectionResult(BaseModel):
"""LLM反思后的结构化结果"""
has_error: bool = Field(..., description="是否发现错误，True表示发现错误，False表示没有发现错误")
error_type: Optional[Literal["missing_info", "incorrect_info", "invalid_format", "tool_call_error", "other_error"]] = Field(None, description="错误类型，必须是以下几种之一：missing_info（信息缺失）、incorrect_info（信息错误）、invalid_format（格式无效）、tool_call_error（工具调用错误）、other_error（其他错误）")
error_description: Optional[str] = Field(None, description="错误的详细描述")
correction_suggestion: Optional[str] = Field(None, description="修正建议，如果是信息缺失，建议生成追问话术；如果是信息错误或格式无效，建议直接修正；如果是工具调用错误，建议修改参数后重新调用")
corrected_data: Optional[dict] = Field(None, description="修正后的结构化数据（如果适用的话）")
# ------------------------------------------------------------------------------
# 4. 定义子模型：报销规则API返回结果
# ------------------------------------------------------------------------------
class ReimbursementAPIResult(BaseModel):
"""模拟的公司报销规则API返回的结构化结果"""
status: Literal["success", "error"] = Field(..., description="API调用状态，success表示成功，error表示失败")
error_message: Optional[str] = Field(None, description="如果status是error，这里是错误的详细描述")
total_requested_amount: Optional[float] = Field(None, description="总申请金额")
total_approved_amount: Optional[float] = Field(None, description="总核定报销金额")
approved_expenses: Optional[List[ExpenseItem]] = Field(None, description="核定后的费用明细列表")
approval_reasons: Optional[List[str]] = Field(None, description="每项费用的核定原因列表，顺序和approved_expenses一致")
format_template: Optional[str] = Field(None, description="公司要求的报销单格式模板（Markdown格式）")
# ------------------------------------------------------------------------------
# 5. 定义全局状态：SelfCorrectingAgentState
# ------------------------------------------------------------------------------
class SelfCorrectingAgentState(BaseModel):
"""
自我纠错报销助手的全局状态
注意：所有字段都必须是Optional的吗？不一定，但LangGraph推荐用Optional，
因为初始状态可能只有部分字段有值，其他字段会在后续的节点中逐步填充
"""
# ------------------------------
# 用户输入相关的字段
# ------------------------------
user_initial_input: str = Field(..., description="用户的初始输入（必填）")
user_supplementary_inputs: List[str] = Field(default_factory=list, description="用户的后续补充输入列表")
# ------------------------------
# 结构化提取相关的字段
# ------------------------------
extracted_business_trip_info: Optional[BusinessTripInfo] = Field(None, description="结构化提取的出差信息")
extracted_expense_items: Optional[List[ExpenseItem]] = Field(None, description="结构化提取的费用明细列表")
extraction_reflection: Optional[ReflectionResult] = Field(None, description="结构化提取后的反思结果")
# ------------------------------
# 工具调用与合规检查相关的字段
# ------------------------------
reimbursement_api_call_count: int = Field(default=0, description="报销规则API的调用次数（用来控制循环次数，避免无限循环）")
max_api_call_count: int = Field(default=3, description="报销规则API的最大调用次数（超过这个次数就结束流程，返回错误）")
reimbursement_api_result: Optional[ReimbursementAPIResult] = Field(None, description="报销规则API的返回结果")
api_call_reflection: Optional[ReflectionResult] = Field(None, description="报销规则API调用后的反思结果")
# ------------------------------
# 报销单生成与格式检查相关的字段
# ------------------------------
draft_reimbursement_form: Optional[str] = Field(None, description="生成的报销单草稿（Markdown格式）")
form_format_check_count: int = Field(default=0, description="报销单格式检查的次数（用来控制循环次数）")
max_form_format_check_count: int = Field(default=3, description="报销单格式检查的最大调用次数")
form_format_reflection: Optional[ReflectionResult] = Field(None, description="报销单格式检查后的反思结果")
# ------------------------------
# 对话历史与Agent输出相关的字段
# ------------------------------
# 这里的messages字段用了LangGraph的add_messages辅助函数，它的作用是：
# 1. 自动合并同一个角色的连续消息（可选）
# 2. 自动处理消息的添加（比如追加而不是覆盖）
# 注意：Annotated是Python 3.9+的类型注解语法，用来给字段添加元数据
messages: Annotated[List, add_messages] = Field(default_factory=list, description="对话历史消息列表（包含用户消息、Agent消息、工具调用消息等）")
agent_final_output: Optional[str] = Field(None, description="Agent的最终输出（如果流程成功结束的话）")
agent_error_output: Optional[str] = Field(None, description="Agent的错误输出（如果流程失败结束的话）")
# ------------------------------
# 流程控制相关的字段
# ------------------------------
current_step: Literal["initialization", "structured_extraction", "extraction_reflection", "ask_user", "call_reimbursement_api", "api_call_reflection", "generate_draft_form", "form_format_reflection", "final_output", "error_output"] = Field(default="initialization", description="Agent当前所处的步骤（用来辅助调试和条件判断）")
should_continue: bool = Field(default=True, description="是否继续执行流程（True表示继续，False表示结束）")
原理解释
让我们来解释一下这个全局状态的几个关键点：
- 子模型的分层设计：我们把全局状态拆成了多个小的子模型（BusinessTripInfo、ExpenseItem、ReflectionResult、ReimbursementAPIResult），每个子模型只负责存储一类相关的信息——这样的设计符合“单一职责原则”，让代码更清晰、更易维护、更易扩展。
- Pydantic的Field装饰器：我们用
Field(..., description="...")给每个字段添加了描述——这些描述非常重要！因为后面我们会把这些Pydantic模型的JSON Schema传给LLM，让LLM按照这个Schema来生成结构化的输出，而description就是LLM理解每个字段含义的关键。 - Pydantic的field_validator装饰器：我们用
@field_validator给一些关键字段添加了自定义的数据验证规则——比如日期格式的验证、结束日期必须≥开始日期的验证、费用金额必须≥0的验证——这些验证规则可以在LLM生成结构化输出之后，立刻检查输出是否合法，如果不合法，就可以触发反思节点，让LLM修正，而不是等到调用工具的时候才出错。 - LangGraph的add_messages辅助函数：我们给
messages字段添加了Annotated[List, add_messages]的元数据——这是LangGraph的一个核心特性，它的作用是让LangGraph自动处理消息列表的添加：比如你在一个节点里返回{"messages": [HumanMessage(content="用户输入")]}，LangGraph会自动把这个消息追加到全局状态的messages列表里，而不是覆盖整个列表——这对存储对话历史非常有用！ - 流程控制相关的字段：我们添加了
current_step、should_continue、reimbursement_api_call_count、max_api_call_count等流程控制相关的字段——这些字段是构建“可控循环”的关键：比如我们可以在反思节点里检查reimbursement_api_call_count是否超过了max_api_call_count，如果超过了，就设置should_continue=False，结束流程，返回错误，避免无限循环。
步骤2：初始化LLM和模拟工具
接下来，我们需要初始化两个东西：
- LLM（大语言模型）：我们用OpenAI的GPT-4o-mini，通过LangChain的
ChatOpenAI类来调用——因为ChatOpenAI类已经封装好了OpenAI的API，而且可以直接和LangGraph配合使用。 - 模拟工具：我们需要模拟一个“公司报销规则API”——因为如果我们用真实的API的话，教程的可复现性会很差，而且可能会产生额外的费用。这个模拟API会接收结构化的出差信息和费用明细，验证合规性，计算核定金额，返回结构化的结果。
代码实现
在项目根目录下创建一个名为utils.py的文件，内容如下：
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from state import BusinessTripInfo, ExpenseItem, ReimbursementAPIResult
# ------------------------------------------------------------------------------
# 1. 加载环境变量
# ------------------------------------------------------------------------------
load_dotenv() # 加载.env文件里的环境变量
# ------------------------------------------------------------------------------
# 2. 初始化LLM
# ------------------------------------------------------------------------------
def init_llm() -> ChatOpenAI:
"""
初始化并返回一个ChatOpenAI实例
:return: ChatOpenAI实例
"""
llm = ChatOpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
base_url=os.getenv("OPENAI_API_BASE", "https://api.openai.com/v1"),
model=os.getenv("OPENAI_MODEL_NAME", "gpt-4o-mini"),
temperature=0.0, # 温度设置为0，让LLM的输出尽可能确定、尽可能少幻觉
max_tokens=4096, # 最大输出token数，足够我们生成结构化数据和报销单
timeout=60, # 超时时间设置为60秒
max_retries=2, # 最大重试次数设置为2次（注意：这个是OpenAI API调用的重试次数，不是我们Agent自我纠错的循环次数）
)
return llm
# ------------------------------------------------------------------------------
# 3. 模拟公司报销规则API
# ------------------------------------------------------------------------------
def simulate_reimbursement_api(
business_trip_info: BusinessTripInfo,
expense_items: List[ExpenseItem]
) -> ReimbursementAPIResult:
"""
模拟公司报销规则API
公司报销规则（示例，你可以根据自己的需求修改）：
1. 高铁票：二等座实报实销，一等座只能报二等座的1.2倍，商务座只能报二等座的2倍（假设二等座的单价是每公里0.5元，但这里为了简化，我们不管公里数，直接根据舱位给一个固定的上限？不对，原计划的示例里是不管公里数，直接看用户的舱位：如果是二等座，实报实销；如果是一等座，假设用户申请的金额超过了“二等座的1.2倍”，但这里我们没有公里数，所以我们可以简化规则：
简化后的公司报销规则（为了教程的可复现性）：
1. 高铁票（high_speed_rail）：
- 二等座：实报实销
- 一等座：最多报销申请金额的80%（或者你可以改成固定上限，比如1000元）
- 商务座：最多报销申请金额的50%
- 如果details里没有提到舱位，默认按二等座处理
2. 机票（flight）：
- 经济舱：实报实销
- 公务舱：最多报销申请金额的70%
- 头等舱：最多报销申请金额的40%
- 如果details里没有提到舱位，默认按经济舱处理
3. 住宿费（hotel）：
- 三星及以下：每天上限350元
- 四星：每天上限500元
- 五星及以上：每天上限800元
- 如果details里没有提到星级，默认按三星及以下处理
- 如果details里没有提到天数，默认按出差天数处理
4. 打车费（taxi）：
- 每天上限100元
- 如果details里没有提到天数，默认按出差天数处理
5. 餐费（meal）：
- 每天上限80元
- 如果details里没有提到天数，默认按出差天数处理
6. 其他费用（other）：
- 必须在details里说明用途，否则最多报销50%
- 单次最多报销1000元
:param business_trip_info: 结构化的出差信息
:param expense_items: 结构化的费用明细列表
:return: 结构化的API返回结果
"""
try:
# 初始化返回结果
total_requested_amount = 0.0
total_approved_amount = 0.0
approved_expenses = []
approval_reasons = []
# 遍历每一项费用，验证合规性并计算核定金额
for expense in expense_items:
total_requested_amount += expense.amount
approved_amount = expense.amount
reason = ""
# 根据费用类型处理
if expense.type == "high_speed_rail":
# 高铁票处理
details = expense.details.lower() if expense.details else ""
if "商务座" in details or "first class" in details or "business class" in details:
# 商务座
approved_amount = min(expense.amount * 0.5, expense.amount) # 其实就是expense.amount*0.5，不过加个min更安全
reason = "商务座，最多报销申请金额的50%"
elif "一等座" in details or "premium economy" in details:
# 一等座
approved_amount = min(expense.amount * 0.8, expense.amount)
reason = "一等座，最多报销申请金额的80%"
else:
# 二等座（默认）
reason = "二等座（默认），实报实销"
elif expense.type == "flight":
# 机票处理
details = expense.details.lower() if expense.details else ""
if "头等舱" in details or "first class" in details:
# 头等舱
approved_amount = expense.amount * 0.4
reason = "头等舱，最多报销申请金额的40%"
elif "公务舱" in details or "business class" in details:
# 公务舱
approved_amount = expense.amount * 0.7
reason = "公务舱，最多报销申请金额的70%"
else:
# 经济舱（默认）
reason = "经济舱（默认），实报实销"
elif expense.type == "hotel":
# 住宿费处理
details = expense.details.lower() if expense.details else ""
# 确定星级对应的每天上限
daily_limit = 350.0 # 三星及以下（默认）
if "五星" in details or "five-star" in details:
daily_limit = 800.0
reason_prefix = "五星及以上酒店"
elif "四星" in details or "four-star" in details:
daily_limit = 500.0
reason_prefix = "四星酒店"
else:
reason_prefix = "三星及以下酒店（默认）"
# 确定住宿天数
days = business_trip_info.days # 默认按出差天数处理
# 尝试从details里提取天数
import re
day_match = re.search(r"(\d+)\s*天", expense.details) if expense.details else None
if day_match:
days = int(day_match.group(1))
# 计算总上限和核定金额
total_limit = daily_limit * days
approved_amount = min(expense.amount, total_limit)
reason = f"{reason_prefix}，住宿{days}天，每天上限{daily_limit}元，总上限{total_limit}元，"
if expense.amount <= total_limit:
reason += "实报实销"
else:
reason += f"申请金额{expense.amount}元超过上限，按上限报销"
elif expense.type == "taxi":
# 打车费处理
daily_limit = 100.0
# 确定天数
days = business_trip_info.days
import re
day_match = re.search(r"(\d+)\s*天", expense.details) if expense.details else None
if day_match:
days = int(day_match.group(1))
# 计算总上限和核定金额
total_limit = daily_limit * days
approved_amount = min(expense.amount, total_limit)
reason = f"打车费，{days}天总上限{total_limit}元，"
if expense.amount <= total_limit:
reason += "实报实销"
else:
reason += f"申请金额{expense.amount}元超过上限，按上限报销"
elif expense.type == "meal":
# 餐费处理
daily_limit = 80.0
# 确定天数
days = business_trip_info.days
import re
day_match = re.search(r"(\d+)\s*天", expense.details) if expense.details else None
if day_match:
days = int(day_match.group(1))
# 计算总上限和核定金额
total_limit = daily_limit * days
approved_amount = min(expense.amount, total_limit)
reason = f"餐费，{days}天总上限{total_limit}元，"
if expense.amount <= total_limit:
reason += "实报实销"
else:
reason += f"申请金额{expense.amount}元超过上限，按上限报销"
elif expense.type == "other":
# 其他费用处理
single_limit = 1000.0
if expense.details and len(expense.details.strip()) > 0:
# 有说明用途
approved_amount = min(expense.amount, single_limit)
reason = f"其他费用，有说明用途，单次上限{single_limit}元，"
else:
# 没有说明用途
approved_amount = min(expense.amount * 0.5, single_limit)
reason = f"其他费用，没有说明用途，最多报销申请金额的50%，单次上限{single_limit}元，"
if expense.amount <= approved_amount:
reason += "实报实销"
else:
reason += f"申请金额{expense.amount}元超过可报销金额，按可报销金额报销"
# 保留两位小数
approved_amount = round(approved_amount, 2)
total_approved_amount += approved_amount
# 添加到核定费用列表和原因列表
approved_expense = ExpenseItem(
type=expense.type,
amount=approved_amount,
details=expense.details
)
approved_expenses.append(approved_expense)
approval_reasons.append(reason)
# 生成公司要求的报销单格式模板（Markdown格式）
format_template = """
# 公司业务报销单草稿
------------------------------
申请人：[请补充申请人姓名]
部门：[请补充部门名称]
出差日期：[YYYY-MM-DD] 至 [YYYY-MM-DD]
出差城市：[城市名称]
出差天数：[X]天
------------------------------
费用明细（必须包含以下列：费用类型、申请金额（元）、核定金额（元）、备注）：
| 费用类型 | 申请金额（元） | 核定金额（元） | 备注 |
|---------|---------------|---------------|------|
| [类型1] | [金额1.00] | [金额1.00] | [备注
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
0
0- 0
已为社区贡献103条内容
所有评论(0)