网站域名是指什么意思,高级搜索引擎技巧,开源企业建站系统哪个好,苏州网站建设永阳网络本篇文章将带你一步步构建一个智能火车票查询 Agent#xff1a;你只需要输入自然语言指令#xff0c;例如#xff1a; “帮我查一下6月15号从上海到南京的火车票” Agent就能自动理解你的需求并使用 Playwright 打开 12306 官网查询前 10 条车次信息#xff0c;然后汇总结果…本篇文章将带你一步步构建一个智能火车票查询 Agent你只需要输入自然语言指令例如 “帮我查一下6月15号从上海到南京的火车票” Agent就能自动理解你的需求并使用 Playwright 打开 12306 官网查询前 10 条车次信息然后汇总结果。
通过这个完整示例希望可以帮助大家入门AI Agent开发掌握如何结合大语言模型、LangChain 工具调用能力以及Playwright打造一个可以执行任务的智能Agent。那我们开始吧
项目初始化
现在开始进行具体项目搭建项目整体结构如下
train_ticket_agent
├── core/ # 核心逻辑模块MyAgent类封装
│ └── agent.py
├── main.py # 入口程序运行 Agent
├── prompts/ # 存放提示词模板
│ ├── final_prompt.txt
│ └── task_prompt.txt
├── requirements.txt # 依赖列表
├── tools/ # 工具模块供 Agent 调用
│ ├── finish.py # Finish 工具占位结束
│ └── train_ticket_query.py # 火车票查询工具调用 Playwright 查询 12306
└── utils/ # 通用工具代码└── ticket_query_scraper.py # Playwright 查询 12306 官网封装成可复用方法core/ → 封装 MyAgent 核心智能体逻辑prompts/ → 任务提示词task_prompt 完成提示词final_prompttools/ → 所有可调用工具火车票查询 / 结束任务utils/ ticket_query_scraper.py → Playwright爬取12306封装main.py → 主入口requirements.txt → 项目依赖管理
安装运行环境
1 . 创建虚拟环境
python -m venv .venv
source .venv/bin/activate # Mac/Linux
# 或
.venv\\Scripts\\activate # Windows2 . 安装依赖
requirements.txt内容如下
langchain0.3.25
python-dotenv~1.1.0
langchain-experimental0.3.4
pydantic~2.10.3
playwright~1.52.0
pypinyin~0.54.0安装依赖包
pip install -r requirementst.txt安装Playwright
playwright install3 . 设置openai的api key
在这个示例中使用的大模型是gpt-3.5需要在项目中配置API Key当然大家也可以使用其他大模型
在项目根目录下创建一个 .env 文件若尚未存在添加以下内容
OPENAI_API_KEYsk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx工具Tools开发
自动查询火车票工具
我们首先的第一个任务是接收用户的自然语言输入比如 “帮我查一下 6 月 15 号从上海到南京的火车票”然后将用户的需求解析为结构化的输入出发地、目的地、日期、时间段以便工具可以使用Playwright实时访问12306查询页面提取前 10 条火车票信息整理成结构化的 JSON 结果返回给用户。
utils/train_ticket_scraper.py
import asyncio
from typing import Listfrom playwright.async_api import async_playwright
from pypinyin import lazy_pinyin, Styleasync def select_city(page, selector: str, city_name: str):initials get_pinyin(city_name)await page.click(selector)for c in initials:await page.keyboard.press(c)await page.wait_for_timeout(100)await page.wait_for_timeout(500)await page.keyboard.press(Enter)async def extract_train_data(page):rows await page.query_selector_all(#queryLeftTable tr.bgc)results []for row in rows[:10]: # 只取前10条train_info {}# 车次编号train_number_el await row.query_selector(div.train a.number)train_info[train_number] (await train_number_el.text_content()).strip() if train_number_el else -# 出发地与到达地station_els await row.query_selector_all(div.cdz strong)from_station_el station_els[0] if len(station_els) 0 else Noneto_station_el station_els[1] if len(station_els) 1 else Nonetrain_info[origin] (await from_station_el.text_content()).strip() if from_station_el else -train_info[destination] (await to_station_el.text_content()).strip() if to_station_el else -# 出发时间与到达时间departure_time_el await row.query_selector(div.cds .start-t)arrival_time_el await row.query_selector(div.cds .color999)train_info[departure_time] (await departure_time_el.text_content()).strip() if departure_time_el else -train_info[arrival_time] (await arrival_time_el.text_content()).strip() if arrival_time_el else -# 历时duration_el await row.query_selector(div.ls strong)train_info[duration] (await duration_el.text_content()).strip() if duration_el else -# 各座位类型seat_cells await row.query_selector_all(td)try:train_info[business_seat] (await seat_cells[1].inner_text()).strip()train_info[first_class_seat] (await seat_cells[3].inner_text()).strip()train_info[second_class_seat] (await seat_cells[4].inner_text()).strip()except IndexError:train_info[business_seat] -train_info[first_class_seat] -train_info[second_class_seat] -results.append(train_info)return resultsdef get_pinyin(text: str) - str:将中文字符串转换为拼音return .join(lazy_pinyin(text, styleStyle.NORMAL))async def extract_train_data_with_browser(origin: str, destination: str, date: str) - List[dict]:async with async_playwright() as p:browser await p.chromium.launch(headlessFalse) # 设置为 True 可无头运行context await browser.new_context()page await context.new_page()# 打开 12306 首页await page.goto(https://www.12306.cn/index/)# 输入查询条件await select_city(page, #fromStationText, origin)await select_city(page, #toStationText, destination)# 填写出发日期注意必须是未来的日期格式YYYY-MM-DDawait page.fill(#train_date, date)# 等待新页面打开async with context.expect_page() as new_page_info:await page.click(#search_one)result_page await new_page_info.value # 获取新打开的 tabawait result_page.wait_for_load_state(domcontentloaded)await result_page.wait_for_selector(#queryLeftTable, timeout10000)result await extract_train_data(result_page)print(查询结果)for train in result:print(train)print(查询完成)await browser.close()return {message: 查询成功,results: result}✅ 通过Playwright从12306爬取真实的火车票信息
extract_train_data_with_browser启动浏览器输入查询条件提取结果。extract_train_data 负责从结果页面中提取前10条火车票数据整理成JSON格式。
tools/train_ticket_query.py
from typing import List
from langchain_core.tools import StructuredTool
import asyncio
from utils.ticket_query_scraper import extract_train_data_with_browser # 改造你的 Playwright 脚本成一个可复用函数def search_train_ticket(origin: str,destination: str,date: str,
) - List[dict]:按条件查询火车票async def _run():return await extract_train_data_with_browser(origin, destination, date)# 用 asyncio 运行异步逻辑result asyncio.run(_run())return resultsearch_train_ticket_tool StructuredTool.from_function(funcsearch_train_ticket,name查询火车票,description调用12306官网真实查询火车票
)✅ 将playwright工具封装到LangChain Tool中
search_train_ticket_tool使用 StructuredTool.from_function封装 Python函数供Agent调用。LangChain Agent调用这个工具时能自动传入origin / destination / date参数调用封装了playwright的函数并获取火车票结果。
完成任务工具 tools/finish.py
from langchain_core.tools import StructuredTooldef finish_placeholder():用于表示任务完成的占位符工具return Nonefinish_tool StructuredTool.from_function(funcfinish_placeholder,nameFINISH,description表示任务完成
)Prompt提示词设计
现在编写提示词让大模型可以根据任务内容和上下文记忆自己去选择使用什么工具需要两个prompt
任务提示词模板task_prompt: 用于指导大模型按格式输出任务完成提示词final_prompt: Agent任务完成后调用此提示词生成最终回复
任务提示词模板task_prompt.txt
你是强大的AI火车票助手可以使用工具与指令查询并购买火车票。你的任务是
{task_description}你可以使用以下工具或指令它们又称为动作Actions
{tools}当前的任务执行记录如下
{memory}请根据任务描述和历史记录思考你下一步的行动。请按照以下格式输出任务你收到的需要执行的任务
思考你如何理解这个任务下一步该怎么做
Action: 要执行的工具名称必须是上面列出的工具名之一
Action Input: 调用该工具所需的参数
{format_instructions}示例格式
{{name: 查询火车票,args: {{origin: 北京,destination: 上海,date: 2024-10-30}}
}}⚠️ 特别说明- 如果你调用工具后观察到的结果中包含以下字段{{message: 查询成功}}说明任务已经成功完成请在下一步输出以下内容表示任务完成{{name: FINISH,args: {{}}}}- 请确保你的输出是符合JSON格式的结构化内容不能包含自然语言。这个prompt将接收以下的参数
变量作用{task_description}当前用户请求如“帮我查一下 6 月 15 号从上海到南京的火车票”{tools}传入工具列表以便大模型可以选择这些就是之前我们开发的工具{memory}上下文记忆思考 工具执行记录{format_instructions}用于约束输出为合法 JSON否则 Pydantic 会报错 **调试建议**在调试时模型经常会不听话输出非Json的文本导致解析失败如 OutputParserException: Invalid json output 报错。使用 {format_instructions} 可强制模型生成结构化 JSON 输出是解决这类问题的关键。 任务完成提示词模板final_prompt.txt
你的任务是
{task_description}以下是你之前的思考过程和使用工具与外部资源交互的结果
{memory}你已经完成了任务。现在请根据上述交互结果总结出本次任务的最终答案。请遵循以下规则输出结果
- 请优先参考 Observation工具的返回结果来组织信息不需要分析思考内容。
- 如果任务是火车票查询请汇总返回的车次列表、出发/到达站、时间、座位情况整理成清晰可读的文本。
- 遍历所有results列表中的项目提取有用信息。完整罗列出来不要省略、不仅仅选前几个结果。在完成查询后让大模型帮忙总结并汇总出车次结果 MyAgent 类实现
MyAgent 是智能火车票助手的核心类它主要的功能包括
✅ 管理大模型调用✅ 管理工具调用✅ 维护上下文记忆✅ 实现推理主流程
先上完整代码
# core/agent.pyimport json
import sys
from typing import Optional, Tuple, Dict, Any
from uuid import UUIDfrom pydantic import ValidationError, BaseModel, Field
from langchain.memory import ConversationTokenBufferMemory
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import PydanticOutputParser, StrOutputParser
from langchain_core.language_models import BaseChatModel
from langchain_core.outputs import GenerationChunk, ChatGenerationChunk, LLMResult
from langchain_core.callbacks import BaseCallbackHandlerfrom langchain.tools.render import render_text_descriptionclass ActionModel(BaseModel):name: str Field(description工具或指令名称)args: Optional[Dict[str, Any]] Field(description工具或指令参数由参数名称和参数值组成)class MyPrintHandler(BaseCallbackHandler):自定义 CallbackHandler用于打印 LLM 推理过程def on_llm_new_token(self,token: str,*,chunk: Optional[GenerationChunk] None,run_id: UUID,parent_run_id: Optional[UUID] None,**kwargs: Any,) - Any:sys.stdout.write(token)sys.stdout.flush()def on_llm_end(self, response: LLMResult, **kwargs: Any) - Any:sys.stdout.write(\\n)sys.stdout.flush()return responseclass MyAgent:def __init__(self,llm: BaseChatModel,tools: list,prompt: PromptTemplate,final_prompt: str,max_thought_steps: Optional[int] 3,):self.llm llm# Convert tool list to dict for fast lookup by nameself.tools {tool.name: tool for tool in tools}self.max_thought_steps max_thought_stepsself.output_parser PydanticOutputParser(pydantic_objectActionModel)self.final_prompt PromptTemplate.from_template(final_prompt)self.llm_chain prompt | self.llm | StrOutputParser()self.verbose_printer MyPrintHandler()self.agent_memory self.init_memory()def init_memory(self):memory ConversationTokenBufferMemory(llmself.llm, max_token_limit4000)memory.save_context({input: \\ninit}, {output: \\n开始})return memorydef run(self, task_description: str) - str:print(开始执行任务...)thought_step_count 0agent_memory self.agent_memorywhile thought_step_count self.max_thought_steps:print(f思考步骤 {thought_step_count 1})action, response self.__step(task_description, agent_memory)# 如果 Action 是 FINISH则结束if action.name FINISH:final_chain self.final_prompt | self.llm | StrOutputParser()reply final_chain.invoke({task_description: task_description,memory: agent_memory})print(f----\\n最终回复:\\n{reply})return reply# 执行动作action_result self.__exec_action(action)# 更新记忆self.update_memory(response, action_result)thought_step_count 1if thought_step_count self.max_thought_steps:# 如果思考步数达到上限返回错误信息print(任务未完成)return 任务未完成def __step(self, task_description, memory) - Tuple[ActionModel, str]:response for s in self.llm_chain.stream({task_description: task_description,memory: memory}, config{callbacks: [self.verbose_printer]}):response sprint(f----\\nResponse:\\n{response})action self.output_parser.parse(response)return action, responsedef __exec_action(self, action: ActionModel) - str:if not action or not action.name:print(未提供有效的动作或工具名称)return 未提供有效的动作或工具名称tool self.tools.get(action.name)if not tool:print(f未找到名称为 {action.name} 的工具)return f未找到名称为 {action.name} 的工具try:return tool.run(action.args)except ValidationError as e:return f参数校验错误: {str(e)}, 参数: {action.args}except Exception as e:return f执行出错: {str(e)}, 类型: {type(e).__name__}, 参数: {action.args}def update_memory(self, response, observation):self.agent_memory.save_context({input: response},{output: \\n返回结果:\\n str(observation)})
初始化init方法介绍
def __init__(self,llm: BaseChatModel,tools: list,prompt: PromptTemplate,final_prompt: str,max_thought_steps: Optional[int] 3,
):self.llm llm# 将工具列表转为 dict 方便按 name 快速查找self.tools {tool.name: tool for tool in tools}self.max_thought_steps max_thought_stepsself.output_parser PydanticOutputParser(pydantic_objectActionModel)self.final_prompt PromptTemplate.from_template(final_prompt)self.llm_chain prompt | self.llm | StrOutputParser()self.verbose_printer MyPrintHandler()self.agent_memory self.init_memory()init方法的参数和说明如下
参数说明llm大语言模型实例表示需要使用大模型接口tools可调用的工具列表需为StructuredTool 对象max_thought_steps智能体最多思考几轮避免死循环output_parser通过ActionModel将LLM 输出结构化为一个 Action(name..., args...) 对象self.llm_chainLangChain中的Chain管道式写法的表示将prompt调用大模型后再将respone内容使用StrOutputParser处理输出final_prompt完成任务时的提示词verbose_printerMyPrintHandler 是一个自定义的 CallbackHandler用于实时输出 LLM 的推理过程agent_memory初始化智能体Agent的记忆上下文
初始化记忆
Agent 需要具备“上下文记忆”能力以便在多轮推理过程中保留每一步的思考与执行记录。这里使用ConversationTokenBufferMemory它能够根据token限制保留最新的上下文信息。
def init_memory(self):memory ConversationTokenBufferMemory(llmself.llm, max_token_limit4000)memory.save_context({input: \\ninit}, {output: \\n开始})return memoryAgent推理主流程 - run
run是Agent的核心方法执行任务完整的思考和工具调用的过程主要步骤包括 获取智能体Agent的上下文记忆agent_memory 执行推理思考的循环 Agent会在限定的思考轮次内不断尝试解决任务直到完成或达到最大步数为止。在每一轮的思考中的步骤如下 调用__step() 把 task描述和上下文记忆memory传入prompt大模型根据记忆和任务描述返回下一步需要执行的Action调用__exec_action函数根据Action执行对应的工具将工具返回的结果更新到记忆中重复进入下一轮思考 生成最终回复 如果Agent 成功完成任务或达到最大轮次后会执行finish的工具并以比较友好的自然语言回复给用户。 运行整体流程
前面我们已经完成以下部分
✅ 工具开发查询、完成✅ 编写Prompttask_prompt、final_prompt✅ 编写MyAgent类
现在需要验证整体流程是否串联成功。main.py示例代码
import jsonfrom dotenv import load_dotenv
from langchain_community.chat_models import ChatOpenAI
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.tools import StructuredTool, render_text_description
from core.agent import MyAgent, ActionModel
from tools.train_ticket_query import search_train_ticket_tool
from tools.finish import finish_toolload_dotenv()if __name__ __main__:tools [search_train_ticket_tool, finish_tool]with open(prompts/task_prompt.txt, r, encodingutf-8) as f:prompt_text f.read()with open(prompts/final_prompt.txt, r, encodingutf-8) as f:final_prompt_text f.read()# 构建提示词模板PromptTemplate ← 你在 main.py 中做这件事parser PydanticOutputParser(pydantic_objectActionModel)prompt PromptTemplate.from_template(prompt_text).partial(toolsrender_text_description(tools),format_instructionsjson.dumps(parser.get_format_instructions(), ensure_asciiFalse))my_agent MyAgent(llmChatOpenAI(modelgpt-3.5-turbo, temperature0),toolstools,promptprompt,final_promptfinal_prompt_text,)task 帮我买25年6月10日早上去南京的火车票reply my_agent.run(task)运行结果示意
运行main.py 后可以看到类似下面这样的流程打印
开始执行任务...
思考步骤 1
{name: 查询火车票,args: {origin: 上海,destination: 南京,date: 2025-06-10}
}
----
Response:
{name: 查询火车票,args: {origin: 上海,destination: 南京,date: 2025-06-10}
}
查询结果
{train_number: G7070, origin: 上海, destination: 南京南, departure_time: 20:46, arrival_time: 22:48, duration: 02:02, business_seat: 无, first_class_seat: 12, second_class_seat: 有}
{train_number: G7098, origin: 上海, destination: 南京, departure_time: 21:05, arrival_time: 22:59, duration: 01:54, business_seat: --, first_class_seat: 18, second_class_seat: 有}
{train_number: D182, origin: 上海松江, destination: 南京, departure_time: 21:22, arrival_time: 00:31, duration: 03:09, business_seat: --, first_class_seat: --, second_class_seat: 候补}
{train_number: G7112, origin: 上海虹桥, destination: 南京, departure_time: 21:35, arrival_time: 23:15, duration: 01:40, business_seat: --, first_class_seat: 有, second_class_seat: 有}
{train_number: G7068, origin: 上海, destination: 南京, departure_time: 21:50, arrival_time: 23:23, duration: 01:33, business_seat: --, first_class_seat: 20, second_class_seat: 有}
{train_number: K8482, origin: 上海, destination: 南京, departure_time: 22:10, arrival_time: 01:27, duration: 03:17, business_seat: --, first_class_seat: --, second_class_seat: --}
{train_number: K1048, origin: 上海, destination: 南京, departure_time: 22:23, arrival_time: 02:08, duration: 03:45, business_seat: --, first_class_seat: --, second_class_seat: --}
{train_number: K850, origin: 上海, destination: 南京, departure_time: 23:21, arrival_time: 04:34, duration: 05:13, business_seat: --, first_class_seat: --, second_class_seat: --}
{train_number: K1506, origin: 上海, destination: 南京, departure_time: 23:40, arrival_time: 03:26, duration: 03:46, business_seat: --, first_class_seat: --, second_class_seat: --}
查询完成
思考步骤 2
{name: FINISH,args: {}
}
----
Response:
{name: FINISH,args: {}
}
----
最终回复:
根据查询结果2025年6月10日去南京的火车票如下
1. 列车编号G7070- 出发站上海- 到达站南京南- 出发时间20:46- 到达时间22:48- 历时02小时02分钟- 商务座无- 一等座12张- 二等座有2. 列车编号G7098- 出发站上海- 到达站南京- 出发时间21:05- 到达时间22:59- 历时01小时54分钟- 商务座--- 一等座18张- 二等座有3. 列车编号D182- 出发站上海松江- 到达站南京- 出发时间21:22- 到达时间00:31- 历时03小时09分钟- 商务座--- 一等座--- 二等座候补4. 列车编号G7112- 出发站上海虹桥- 到达站南京- 出发时间21:35- 到达时间23:15- 历时01小时40分钟- 商务座--- 一等座有- 二等座有5. 列车编号G7068- 出发站上海- 到达站南京- 出发时间21:50- 到达时间23:23- 历时01小时33分钟- 商务座--- 一等座20张- 二等座有6. 列车编号K8482- 出发站上海- 到达站南京- 出发时间22:10- 到达时间01:27- 历时03小时17分钟- 商务座--- 一等座--- 二等座--7. 列车编号K1048- 出发站上海- 到达站南京- 出发时间22:23- 到达时间02:08- 历时03小时45分钟- 商务座--- 一等座--- 二等座--8. 列车编号K850- 出发站上海- 到达站南京- 出发时间23:21- 到达时间04:34- 历时05小时13分钟- 商务座--- 一等座--- 二等座--9. 列车编号K1506- 出发站上海- 到达站南京- 出发时间23:40- 到达时间03:26- 历时03小时46分钟- 商务座--- 一等座--- 二等座--小结
通过上面我们完成了一个完整的 LangChain ReAct 智能体实践案例具备以下能力
✅ 能理解用户自然语言请求
✅ 能通过 Prompt 引导大模型选择合适工具
✅ 能自动完成工具调用、记忆更新、迭代推理
✅ 最终输出结果反馈给用户 Github仓库地址
https://github.com/bridgeshi85/train-ticket-agent
