企业官网建设流程全解析

Python + 官方 LangChain + LangGraph
目的：熟悉从环境配置到搭建简单agent的全过程
环境：Mac + IDEA

项目概述

场景：一个可以暂停、恢复、记住偏好的采购助手

架构图

接收采购需求 ↓ 检查需求是否完整 ├── 缺信息 → 暂停，等待用户补充 ↓ 查询三家供应商 ├── A：正常返回 ├── B：模拟一次失败，再重试 └── C：正常返回 ↓ 比较报价，生成方案 ↓ 暂停，等待用户审批 ├── 拒绝 → 修改要求，重新生成 └── 批准 → 创建模拟订单

LangGraph 能力对应

过程

环境配置

创建 python 3.12 的虚拟环境

python3.12 -m venv myEnv3.12

• python3.12 虚拟环境的python版本
• myEnv3.12 虚拟环境名称

启动虚拟环境

source myEnv3.12/bin/activate

安装 langgraph == 1.2.12

python -m pip install "langgraph==1.2.12"

核心对象

• State：任务共享的数据。节点读取它，并返回需要更新的字段。
• Node：普通python函数，负责完成一步工作。

状态字段

• TypedDict 定义数据的结构，真正的采购数据存放在运行时的字典里
  例如：

  class baojiaState(TypedDict): unit_price: int

  表达的是：这种字典应该有一个 unit_price 字段，其值是整数。它没有保存某次报价，所以

  baojiaState.unit_price

  不能取得报价。
  字段值应该从某个具体的 state 字典中读取，而不是类型定义上读取。

定义一份共享 State

class PurchaseState(TypedDict): quantity: int budget: int unit_price: int total_price: int within_budget: bool summary: str

不是每个节点各拥有一份完全独立的状态，而是三个节点逐步补充同一份任务状态。

三个节点

计算规则：

• 总价 = 数量 * 单价
• 总价小于或等于预算，表示符合预算。
• 说明文字应根据计算结果生成。

每个节点优先返回自己更新的字段。

在每个节点中打印：

• 当前节点名称；
• 这个节点读取的关键数据。

使用的 API

写法

LangGraph 如何调用函数，以及如何接收函数的结果

普通python函数，调用者需要分别传入三个参数

def calculate_quote(quantity, unit_price, budget):

LangGraph节点形式

def calculate_quote(state: PurchaseState):

LangGraph 会把当前状态传给 state，再从这个字典里读取需要的数据。

执行过程：

LangGraph 持有当前状态 ↓ 把当前状态传给 calculate_quote 的 state 参数 ↓ 函数从 state 中读取数量、单价、预算 ↓ 函数完成计算，返回状态更新

字典需要通过键访问。
不能写成PurchaseState.quantity，而是 传参时用state:PurchaseState，使用时用state["quantity]

完整代码示例

fromtypingimportTypedDictfromlanggraph.graphimportSTART,END,StateGraphclassPurchaseState(TypedDict,total=False):quantity:intbudget:intunit_price:inttotal_price:intwithin_budget:boolsummary:strdefload_quote(state:PurchaseState):print("load_quote")return{"unit_price":7300}defcalculate_quote(state:PurchaseState):print("calculate_quote")print("读取的关键数据是：",state["quantity"],state["unit_price"],state["budget"])total_price=state["quantity"]*state["unit_price"]iftotal_price<=state["budget"]:within_budget=Trueelse:within_budget=Falsereturn{"total_price":total_price,"within_budget":within_budget}defmake_summary(state:PurchaseState):print("make_summary")print("读取的关键数据是：",state["quantity"],state["total_price"])ifstate["within_budget"]:summary=f"购买 {state["quantity"]} 件，总价 {state["total_price"]}，在预算内"else:summary=f"购买 {state["quantity"]} 件，总价 {state["total_price"]}，超出预算"return{"summary":summary}builder=StateGraph(PurchaseState)builder.add_node("load_quote",load_quote)builder.add_node("calculate_quote",calculate_quote)builder.add_node("make_summary",make_summary)builder.add_edge(START,"load_quote")builder.add_edge("load_quote","calculate_quote")builder.add_edge("calculate_quote","make_summary")builder.add_edge("make_summary",END)graph=builder.compile()result=graph.invoke({"quantity":30,"budget":240000})print("最终结果为",result)

终端执行

python 文件名.py

输出结果为

((myEnv3.12) ) xuxiaorui@U-C47PDXMX-0020 purchase-lab % python purchase_graph.py load_quote calculate_quote 读取的关键数据是： 30 7300 240000 make_summary 读取的关键数据是： 30 219000 最终结果为 {'quantity': 30, 'budget': 240000, 'unit_price': 7300, 'total_price': 219000, 'within_budget': True, 'summary': '购买 30 件，总价 219000，在预算内'}

回顾

节点之间没有直接调用

函数负责业务计算，图负责执行顺序

比如load_quote函数里没有调用calculate_qupte，而是因为加了这条边

builder.add_edge("load_quote","calculate_quote")

节点返回的字典只更新对应字段

接入 API

这里使用 deepseek 的 API

安装接入包

python -m pip install langchain-deepseek

• 安装包名：langchain-deepseek 用连字符
• python 导入名：langchain_deepseek，用下划线

配置 API Key

read -rs "DEEPSEEK_API_KEY?请输入 DeepSeek API Key: "

粘贴密钥后回车。
执行

export DEEPSEEK_API_KEY

从这个终端启动的 Python 程序就能读取该环境变量。ChatDeepSeek 默认会从 DEEPSEEK_API_KEY 获取密钥，你不用把它写进 Python 文件。

在 model_probe.py 中创建模型对象

导入

from langchain_deepseek import ChatDeepSeek

创建 ChatDeepSeek 对象

创建一个ChatDeepSeek对象，赋给变量model。

创建对象是在本地配置一个模型客户端。 模型运行在 DeepSeek 的服务器上，调用invoke时才会发送生成请求。

组织输入消息

创建一个列表变量messages，放入两条消息。

LangChain 中：
每条消息是一个二元组，第一项是角色，第二项是文字内容。
示例：

("system","填写任务规则")

两条消息分别负责：

发起调用

模型调用使用

response=model.invoke(messages)

response是一个AIMessage消息对象，

• 用type(response)查看返回对象的类型
• 用response.content读取生成的正文

完整代码示例

fromlangchain_deepseekimportChatDeepSeek# 创建 ChatDeepSeek 对象model=ChatDeepSeek(model="deepseek-flash",timeout=30,max_retries=0,)# 创建 输入消息messages=[("system","你是一个采购助手，你需要忠实使用提供的数据，不补充未知事实"),("human","数量：30；单价：7300 元；总价：219000 元；预算：240000 元；预算判断：符合预算")]# 模型调用response=model.invoke(messages)print("response的类型为：",type(response))print(response.content)

输出：

((myEnv3.12))xuxiaorui@U-C47PDXMX-0020 purchase-lab % python model_probe.py response的类型为：<class'langchain_core.messages.ai.AIMessage'>已核对采购数据： - 数量：30 - 单价：7300 元 - 总价：219000 元（30 ×7300=219000元） - 预算：240000 元 - 预算判断：符合预算（219000 元 ≤240000元） 结论：该采购总价未超出预算，符合预算要求。

讲解

ChatDeepSeek(...) 创建本地客户端对象，记录模型名、超时等配置 ↓ 你构造 messages 明确告诉模型任务要求和采购数据 ↓ model.invoke(messages) 将消息发给 DeepSeek 服务，等待模型响应 ↓ LangChain 将响应转换成 AIMessage ↓ response.content 取出生成的正文

模型只能看到你发给它的消息，不会自动看到 python 中的PurchaseState、变量或函数。
接入图之后，需要在节点里把 State 中的数据组织成消息。

模型接入 make_summary

让节点完成

读取 State ↓ 构造消息 ↓ 调用模型，拿到正文 ↓ 返回 summary 的更新

节点的外部接口保持

defmake_summary(state:PurchaseState):

接入代码示例

fromtypingimportTypedDictfromlanggraph.graphimportSTART,END,StateGraphfromlangchain_deepseekimportChatDeepSeekclassPurchaseState(TypedDict,total=False):quantity:intbudget:intunit_price:inttotal_price:intwithin_budget:boolsummary:strdefload_quote(state:PurchaseState):print("load_quote")return{"unit_price":7300}defcalculate_quote(state:PurchaseState):print("calculate_quote")print("读取的关键数据是：",state["quantity"],state["unit_price"],state["budget"])total_price=state["quantity"]*state["unit_price"]iftotal_price<=state["budget"]:within_budget=Trueelse:within_budget=Falsereturn{"total_price":total_price,"within_budget":within_budget}# 加入 ChatDeepSeek 的导入和模型初始化。# 可以复用你刚刚验证成功的配置。把模型对象放在节点函数外面，供 make_summary 使用。## 在 make_summary 内构造 messages。# 系统消息描述规则；用户消息通过 f-string 从 state 读取数据，不能继续固定写成“数量 30”。## 调用模型，并接住响应。# 使用刚刚学会的 model.invoke(messages)。## 把正文作为 summary 返回。# 返回值仍然是状态更新字典，其中 summary 对应的值来自 response.content。# 创建 ChatDeepSeek 对象model=ChatDeepSeek(model="deepseek-flash",timeout=30,max_retries=0,)defmake_summary(state:PurchaseState):print("make_summary")print("读取的关键数据是：",state["quantity"],state["total_price"])messages=[("system","你是一个采购助手，你需要忠实使用提供的数据，不补充未知事实"),("human",f"数量：{state['quantity']}；单价：{state['unit_price']}元；总价：{state['total_price']}元；预算：{state['budget']}元；预算判断：{state['within_budget']}")]response=model.invoke(messages)summary=response.contentreturn{"summary":summary}builder=StateGraph(PurchaseState)builder.add_node("load_quote",load_quote)builder.add_node("calculate_quote",calculate_quote)builder.add_node("make_summary",make_summary)builder.add_edge(START,"load_quote")builder.add_edge("load_quote","calculate_quote")builder.add_edge("calculate_quote","make_summary")builder.add_edge("make_summary",END)graph=builder.compile()result=graph.invoke({"quantity":30,"budget":240000})print("最终结果为",result["summary"])

输出为：

((myEnv3.12))xuxiaorui@U-C47PDXMX-0020 purchase-lab % python purchase_graph.py load_quote calculate_quote 读取的关键数据是：307300240000make_summary 读取的关键数据是：30219000最终结果为 已确认： - 数量：30 - 单价：7300 元 - 总价：219000 元 - 预算：240000 元 - 预算判断：True 即总价219000元未超过预算240000元。

工具调用

当前，程序先取得报价，再把数据交给模型。
接下来实现：

用户提出采购问题 ↓ 模型判断需要查询报价 ↓ 模型返回“调用哪个工具、传什么参数” ↓ 你的 Python 程序执行报价工具 ↓ 把查询结果交回模型 ↓ 模型根据结果回答

模型负责提出调用请求，python程序负责执行工具。

写一个可以查询供应商报价的工具

fromlangchain_core.toolsimporttoolfromlangchain_deepseekimportChatDeepSeekfromlangchain_core.messagesimportToolMessageimportjson# | 项目 | 要求 |# |---|---|# | 函数名 | `get_supplier_quote` |# | 输入参数 | `supplier_name`，类型为 `str` |# | 返回值 | 一个字典 |# | 函数说明 | 用 docstring 说明它查询什么，以及支持哪些供应商 |# | 装饰器 | 在函数定义上一行添加 `@tool` |model=ChatDeepSeek(model="deepseek-flash",timeout=30,max_retries=0,)@tooldefget_supplier_quote(supplier_name:str):""" 查询供应商报价。supplier_name 支持 "A" 和 "B"；其他名称返回 error。 :param supplier_name: :return: """ifsupplier_name=="A":return{"supplier_name":supplier_name,"unit_price":7300,"delivery_days":10}elifsupplier_name=="B":return{"supplier_name":supplier_name,"unit_price":7000,"delivery_days":18}else:return{"error":"未知供应商"}model_with_tools=model.bind_tools([get_supplier_quote])messages=[("system","你是一个采购助手，你需要使用提供的工具进行供应商报价的查询后再作答"),("user",f"请查询供应商B的报价")]responses=model_with_tools.invoke(messages)# 把模型的工具调用请求加入对话messages=messages+[responses]print("responses.content:",responses.content)print("responses.tool_calls",responses.tool_calls)# 取出调用请求，并执行工具tool_call=responses.tool_calls[0]tool_result=get_supplier_quote.invoke(tool_call["args"])# 把工具结果作为 ToolMessage 加入对话tool_message=ToolMessage(content=json.dumps(tool_result,ensure_ascii=False),tool_call_id=tool_call["id"],)messages=messages+[tool_message]# 让模型根据工具生成最终答复final_res=model_with_tools.invoke(messages)print("final_res.content",final_res.content)# 对话顺序# system 消息# user 消息# AIMessage：请求调用 get_supplier_quote# ToolMessage：工具返回 B 的报价，并带上对应的调用编号# AIMessage：根据报价生成的最终答复# 工具结果是字典，放入 ToolMessage.content 前可以用 json.dumps(..., ensure_ascii=False) 转成 JSON 字符串。这样模型收到的是可读的结构化报价内容。

输出为：

((myEnv3.12))xuxiaorui@U-C47PDXMX-0020 purchase-lab % python tool_probe.py responses.content: responses.tool_calls[{'name':'get_supplier_quote','args':{'supplier_name':'B'},'id':'call_00_AGSAw61sXXAr8loGDefJ5063','type':'tool_call'}]final_res.content 供应商B的报价信息如下： - **供应商名称**：B - **单价**：7,000 - **交货周期**：18 天 如需对比其他供应商报价或进一步分析，请告诉我。