本篇文章将带你一步步构建一个智能火车票查询 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/ → 任务提示词tools/ → 所有可调用工具（火车票查询 / 结束任务）utils/ ticket_query_scraper.py → Playwright爬取12306封装main.py → 主入口requirements.txt → 项目依赖管理
安装运行环境
1 . 创建虚拟环境
python -m venv .venvsource .venv/bin/activate # Mac/Linux# 或.venv\Scripts\activate # Windows2 . 安装依赖
requirements.txt内容如下
langchain==0.3.25python-dotenv~=1.1.0langchain-experimental==0.3.4pydantic~=2.10.3playwright~=1.52.0pypinyin~=0.54.0安装依赖包
pip install -r requirementst.txt安装Playwright：
playwright install3 . 设置openai的api key
在这个示例中使用的大模型是gpt-3.5，需要在项目中配置API Key**，当然大家也可以使用大模型
在项目根目录下创建一个.env文件，添加以下内容：
OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx工具Tools开发
自动查询火车票工具
我们首先的第一个任务是接收用户的自然语言输入比如 “帮我查一下 6 月 15 号从上海到南京的火车票”，然后将用户的需求解析为结构化的输入（出发地、目的地、日期、时间段），以便工具可以使用Playwright实时访问12306查询页面，提取前 10 条火车票信息，整理成结构化的 JSON 结果返回给用户。
utils/train_ticket_scraper.py
import asynciofrom typing import Listfrom playwright.async_api import async_playwrightfrom pypinyin import lazy_pinyin, Styleasyncdef 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")asyncdef 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) > 0elseNone to_station_el = station_els[1] if len(station_els) > 1elseNone train_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, style=Style.NORMAL))asyncdef extract_train_data_with_browser(origin: str, destination: str, date: str) -> List[dict]: asyncwith async_playwright() as p: browser = await p.chromium.launch(headless=False) # 设置为 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-DD） await page.fill('#train_date', date) # 等待新页面打开 asyncwith context.expect_page() as new_page_info: await page.click('#search_one') result_page = await new_page_info.value # 获取新打开的 tab await result_page.wait_for_load_state('domcontentloaded') await result_page.wait_for_selector("#queryLeftTable", timeout=10000) 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 Listfrom langchain_core.tools import StructuredToolimport asynciofrom utils.ticket_query_scraper import extract_train_data_with_browser # 改造你的 Playwright 脚本成一个可复用函数def search_train_ticket( origin: str, destination: str, date: str,) -> List[dict]: """按条件查询火车票""" asyncdef _run(): returnawait extract_train_data_with_browser(origin, destination, date) # 用 asyncio 运行异步逻辑 result = asyncio.run(_run()) return resultsearch_train_ticket_tool = StructuredTool.from_function( func=search_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( func=finish_placeholder, name="FINISH", 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 jsonimport sysfrom typing import Optional, Tuple, Dict, Anyfrom uuid import UUIDfrom pydantic import ValidationError, BaseModel, Fieldfrom langchain.memory import ConversationTokenBufferMemoryfrom langchain_core.prompts import PromptTemplatefrom langchain_core.output_parsers import PydanticOutputParser, StrOutputParserfrom langchain_core.language_models import BaseChatModelfrom langchain_core.outputs import GenerationChunk, ChatGenerationChunk, LLMResultfrom 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 name self.tools = {tool.name: tool for tool in tools} self.max_thought_steps = max_thought_steps self.output_parser = PydanticOutputParser(pydantic_object=ActionModel) 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(llm=self.llm, max_token_limit=4000) memory.save_context({"input": "\ninit"}, {"output": "\n开始"}) return memory def run(self, task_description: str) -> str: print("开始执行任务...") thought_step_count = 0 agent_memory = self.agent_memory while 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 += 1 if 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 += s print(f"----\nResponse:\n{response}") action = self.output_parser.parse(response) return action, response def __exec_action(self, action: ActionModel) -> str: ifnot action ornot action.name: print("未提供有效的动作或工具名称") return"未提供有效的动作或工具名称" tool = self.tools.get(action.name) ifnot tool: print(f"未找到名称为 {action.name} 的工具") returnf"未找到名称为 {action.name} 的工具" try: return tool.run(action.args) except ValidationError as e: returnf"参数校验错误: {str(e)}, 参数: {action.args}" except Exception as e: returnf"执行出错: {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_steps self.output_parser = PydanticOutputParser(pydantic_object=ActionModel) 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_chain | LangChain中的Chain管道式写法的，表示将prompt调用大模型后再将respone内容使用StrOutputParser处理输出 |
| final_prompt | 完成任务时的提示词 |
| verbose_printer | MyPrintHandler 是一个自定义的 CallbackHandler，用于实时输出 LLM 的推理过程 |
| agent_memory | 初始化智能体Agent的记忆上下文 |
初始化记忆
Agent 需要具备“上下文记忆”能力，以便在多轮推理过程中保留每一步的思考与执行记录。这里使用ConversationTokenBufferMemory，它能够根据token限制保留最新的上下文信息。
def init_memory(self): memory = ConversationTokenBufferMemory(llm=self.llm, max_token_limit=4000) memory.save_context({"input": "\ninit"}, {"output": "\n开始"}) return memoryAgent推理主流程 - run
run是Agent的核心方法，执行任务完整的思考和工具调用的过程，主要步骤包括：
- 获取智能体Agent的上下文记忆agent_memory执行推理思考的循环Agent会在限定的思考轮次内不断尝试解决任务，直到完成或达到最大步数为止。在每一轮的思考中的步骤如下：
- 调用__step()， 把 task描述和上下文记忆memory传入prompt，大模型根据记忆和任务描述返回下一步需要执行的Action调用__exec_action函数，根据Action执行对应的工具将工具返回的结果更新到记忆中重复进入下一轮思考
运行整体流程
前面我们已经完成以下部分：
- ✅ 工具开发（查询、完成）✅ 编写Prompt（task_prompt、final_prompt）✅ 编写MyAgent类
现在需要验证整体流程是否串联成功。main.py示例代码：
import jsonfrom dotenv import load_dotenvfrom langchain_community.chat_models import ChatOpenAIfrom langchain_core.output_parsers import PydanticOutputParserfrom langchain_core.prompts import PromptTemplatefrom langchain_core.tools import StructuredTool, render_text_descriptionfrom core.agent import MyAgent, ActionModelfrom tools.train_ticket_query import search_train_ticket_toolfrom tools.finish import finish_toolload_dotenv()if __name__ == "__main__": tools = [search_train_ticket_tool, finish_tool] with open("prompts/task_prompt.txt", "r", encoding="utf-8") as f: prompt_text = f.read() with open("prompts/final_prompt.txt", "r", encoding="utf-8") as f: final_prompt_text = f.read() # 构建提示词模板（PromptTemplate） ← 你在 main.py 中做这件事 parser = PydanticOutputParser(pydantic_object=ActionModel) prompt = PromptTemplate.from_template(prompt_text).partial( tools=render_text_description(tools), format_instructions=json.dumps( parser.get_format_instructions(), ensure_ascii=False ) ) my_agent = MyAgent( llm=ChatOpenAI(model="gpt-3.5-turbo", temperature=0), tools=tools, prompt=prompt, final_prompt=final_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分钟 - 商务座：-- - 一等座：-- - 二等座：--Github仓库地址
