本篇文章將帶你一步步構建一個智能火車票查詢 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_prompt)
- tools/ → 所有可調用工具(火車票查詢 / 結束任務)
- 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 # Windows
2 . 安裝依賴
requirements.txt內容如下
langchain==0.3.25
python-dotenv~=1.1.0
langchain-experimental==0.3.4
pydantic~=2.10.3
playwright~=1.52.0
pypinyin~=0.54.0
安裝依賴包
pip install -r requirementst.txt
安裝Playwright:
playwright install
3 . 設置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 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, style=Style.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(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)# 等待新頁面打開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", 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 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(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 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_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 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_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 memory
Agent推理主流程 - run
run是Agent的核心方法,執行任務完整的思考和工具調用的過程,主要步驟包括:
-
獲取智能體Agent的上下文記憶agent_memory
-
執行推理思考的循環
Agent會在限定的思考輪次內不斷嘗試解決任務,直到完成或達到最大步數為止。在每一輪的思考中的步驟如下:
- 調用__step(), 把 task描述和上下文記憶memory傳入prompt,大模型根據記憶和任務描述返回下一步需要執行的Action
- 調用__exec_action函數,根據Action執行對應的工具
- 將工具返回的結果更新到記憶中
- 重復進入下一輪思考
-
生成最終回復
如果Agent 成功完成任務或達到最大輪次后會執行finish的工具,并以比較友好的自然語言回復給用戶。
運行整體流程
前面我們已經完成以下部分:
- ? 工具開發(查詢、完成)
- ? 編寫Prompt(task_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", 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分鐘- 商務座:--- 一等座:--- 二等座:--
小結
通過上面我們完成了一個完整的 LangChain + ReAct 智能體實踐案例,具備以下能力:
? 能理解用戶自然語言請求
? 能通過 Prompt 引導大模型選擇合適工具
? 能自動完成工具調用、記憶更新、迭代推理
? 最終輸出結果反饋給用戶
Github倉庫地址
https://github.com/bridgeshi85/train-ticket-agent