ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

从 0 到 1:用 LangGraph 与 RAG 搭建旅行规划 Agent Harness 的 TaoToken 实践

从 0 到 1:用 LangGraph 与 RAG 搭建旅行规划 Agent Harness 的 TaoToken 实践

1. 旅行规划 Agent Harness 到底解决什么问题

旅行规划这件事,表面看是“排个行程”,实际做起来是一堆互相牵制的约束在打架。你要同时考虑出发和返程时间、每天能承受的游玩强度、老人小孩的体力、预算上限、景点是否临时关闭、高铁票能不能抢到、民宿有没有电梯。任何一个条件变了,整条链路都得跟着重算。我试过用纯提示词让大模型直接吐一份七天行程,结果它把玉龙雪山大索道排给了带 65 岁老人和 5 岁小孩的家庭,还信誓旦旦说“适合全家”。这就是没有 Harness 的典型症状:模型有认知能力,但没有约束校验、没有实时数据、没有失败重试。

所谓 Agent Harness,你可以把它理解成“智能体团队的调度中枢加基础设施”。它不负责某一个具体决策,而是负责把多个 Agent 串起来,给它们统一的状态、统一的工具入口、统一的错误处理。旅行规划场景里,至少需要这几类角色:负责和用户对话并抽取偏好的交互 Agent、负责从目的地知识库里检索的 RAG 检索 Agent、负责生成初步行程的规划 Agent、负责按多目标做优化的优化 Agent、负责把行程拆成可执行任务的调度 Agent。Harness 要保证这些角色之间的状态传递不丢、格式不崩、失败可降级。

为什么用 LangGraph 而不是普通 Chain?因为旅行规划天然是带循环和条件分支的。用户说“把普达措换成松赞林寺”,系统要回到检索节点重新拉数据,再走一遍规划和优化,而不是线性往下走。LangGraph 的状态图模型正好描述这种“可回退、可分支、可持久化”的流程。而 RAG 解决的是模型知识过时和幻觉问题:目的地开放时间、门票、交通接驳这些信息必须从外部知识库注入,不能靠模型记忆。

这套东西适合谁?如果你是大模型应用开发者,想从“玩具级 Agent”跨到“能跑通完整链路的生产级骨架”,这篇可以直接照着搭。如果你是旅游行业的技术同学,想做一个差异化的行程规划服务,这里的目录结构和节点配置可以当模板。普通用户也能看懂整体思路,知道未来的旅行规划工具应该长什么样。

TaoToken 在这里的角色是统一模型通道。你不需要为每个模型单独维护一套 Key 和 Base URL,通过一个兼容接口就能切换不同模型,Harness 里的模型调用层因此可以保持稳定。下面从目录结构开始,一步步把可运行的骨架搭出来。

2. TaoToken 前置准备与 Harness 目录结构

在写第一行 LangGraph 代码之前,先把模型通道和项目骨架定下来。模型通道用 TaoToken 统一接入,这样后面无论你换哪个模型做规划或做检索摘要,Harness 里的调用代码都不用改。你需要先拿到 API Key,入口在控制台的 API Keys 页面:https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite 。拿到 Key 之后,模型调用的 Base URL 统一用 https://taotoken.net/api ,注意这个地址后面不加任何查询参数。

Harness 的目录结构建议按“关注点分离”来切,不要把所有逻辑塞进一个文件。下面是我实测下来比较顺手的一套结构,你可以直接复制:

travel-agent-harness/ ├── .env # 存放 TAOTOKEN_API_KEY 等敏感配置 ├── requirements.txt ├── config/ │ ├── settings.py # 读取环境变量,集中管理模型与检索参数 │ └── graph.toml # LangGraph 节点与边的声明式配置 ├── harness/ │ ├── __init__.py │ ├── state.py # 定义全局 State(TypedDict) │ ├── graph_builder.py # 组装 LangGraph 状态图 │ ├── llm_client.py # 封装 TaoToken 模型调用 │ └── nodes/ │ ├── extract_preferences.py │ ├── retrieve_knowledge.py │ ├── plan_itinerary.py │ ├── optimize_itinerary.py │ └── human_review.py ├── rag/ │ ├── ingest.py # 目的地知识入库 │ ├── vector_store.py # 向量库封装 │ └── retriever.py # 检索链路 ├── data/ │ └── destinations/ # 目的地原始知识文件(md/json) └── run_local.py # 本地运行入口

这个结构里,harness/是核心,rag/是知识注入链路,config/放配置。state.py定义的状态对象是整个 Harness 的“记忆”,所有节点都读写它。我建议状态里至少包含这些字段:用户原始输入、抽取出的偏好字典、检索到的知识片段列表、初步行程、优化后行程、用户反馈、当前重试次数。用 TypedDict 定义,LangGraph 会帮你做类型提示。

llm_client.py里封装 TaoToken 调用,核心是把 Base URL 和 Key 从环境变量读进来,然后暴露一个统一的chat()方法。这样规划节点和检索摘要节点都调同一个方法,换模型只改配置。graph.toml用来声明节点和边的连接关系,把流程配置和代码逻辑分开,后面调整分支条件不用动 Python 文件。

依赖方面,requirements.txt至少要有:langgraph、langchain-core、openai(因为 TaoToken 兼容 OpenAI 接口)、chromadb或faiss-cpu做向量库、python-dotenv读环境变量。装完之后,.env里写一行TAOTOKEN_API_KEY=你的Key,settings.py用os.getenv读出来。这一步做完,模型通道和项目骨架就齐了,接下来进入可复制的配置。

3. 可复制配置:LangGraph 节点与 RAG 检索链路

这一节给的是能直接粘贴运行的配置片段。先看config/settings.py,它负责把 TaoToken 的通道参数集中管理:

import os from dotenv import load_dotenv load_dotenv() class Settings: TAOTOKEN_API_KEY = os.getenv("TAOTOKEN_API_KEY") TAOTOKEN_BASE_URL = "https://taotoken.net/api" PLANNER_MODEL = "claude-3-5-sonnet" # 规划节点用的模型 ID RETRIEVER_MODEL = "gpt-4o-mini" # 检索摘要用的模型 ID VECTOR_STORE_PATH = "./data/vector_store" TOP_K = 5 settings = Settings()

注意TAOTOKEN_BASE_URL写死为https://taotoken.net/api,不要加斜杠后缀。模型 ID 按你实际在控制台看到的填,这里只是示例。接下来是harness/llm_client.py,它把 TaoToken 封装成统一调用:

from openai import OpenAI from config.settings import settings client = OpenAI( api_key=settings.TAOTOKEN_API_KEY, base_url=settings.TAOTOKEN_BASE_URL, ) def chat(messages, model=None, temperature=0.3): resp = client.chat.completions.create( model=model or settings.PLANNER_MODEL, messages=messages, temperature=temperature, ) return resp.choices[0].message.content

这段代码里base_url指向 TaoToken 的 API 地址,api_key从环境变量来。三件套就是 Base URL、Key、Model ID,缺一不可。如果你后面要换成别的模型,只改model参数即可,客户端不用动。

然后是 LangGraph 的状态定义和节点配置。harness/state.py:

from typing import TypedDict, List, Dict, Any class TravelState(TypedDict): user_input: str preferences: Dict[str, Any] retrieved_docs: List[str] draft_itinerary: str optimized_itinerary: str feedback: str retry_count: int

harness/graph_builder.py里组装状态图,核心是定义节点和条件边:

from langgraph.graph import StateGraph, END from harness.state import TravelState from harness.nodes.extract_preferences import extract_preferences from harness.nodes.retrieve_knowledge import retrieve_knowledge from harness.nodes.plan_itinerary import plan_itinerary from harness.nodes.optimize_itinerary import optimize_itinerary from harness.nodes.human_review import human_review def build_graph(): g = StateGraph(TravelState) g.add_node("extract", extract_preferences) g.add_node("retrieve", retrieve_knowledge) g.add_node("plan", plan_itinerary) g.add_node("optimize", optimize_itinerary) g.add_node("review", human_review) g.set_entry_point("extract") g.add_edge("extract", "retrieve") g.add_edge("retrieve", "plan") g.add_edge("plan", "optimize") g.add_edge("optimize", "review") g.add_conditional_edges( "review", lambda s: "plan" if s["feedback"] == "reject" else END, {"plan": "plan", END: END}, ) return g.compile()

这里的关键是review节点后面的条件边:如果用户反馈是reject,就回到plan重新生成,形成循环;否则结束。这就是 LangGraph 相比线性 Chain 的优势。RAG 检索链路在rag/retriever.py:

from rag.vector_store import get_collection from config.settings import settings def retrieve(query: str, top_k: int = None): collection = get_collection() results = collection.query( query_texts=[query], n_results=top_k or settings.TOP_K, ) return results["documents"][0]

retrieve_knowledge节点里调用这个函数,把检索到的文档拼进提示词。plan_itinerary节点则把偏好和检索结果一起发给 TaoToken 的模型。这套配置跑起来后,状态会在节点间流转,检索结果会注入到规划提示词里。配置部分到此完整,下面验证请求。

4. 验证请求与成功结果

配置写完后,先做一次最小验证,确认 TaoToken 通道能通、LangGraph 能跑。第一步单独测模型调用,在项目根目录执行:

from harness.llm_client import chat resp = chat([{"role": "user", "content": "用一句话说明什么是旅行规划 Agent Harness"}]) print(resp)

如果返回一段正常的中文说明,说明 Base URL、Key、Model ID 三件套都对了。如果报 401,说明 Key 没读到或写错了;如果报连接错误,检查base_url是不是https://taotoken.net/api,不要多写路径。

第二步跑完整图。run_local.py里这样写:

from harness.graph_builder import build_graph graph = build_graph() initial_state = { "user_input": "带65岁老人和5岁小孩去云南7天,预算2万,低强度,喜欢自然风光和人文历史", "preferences": {}, "retrieved_docs": [], "draft_itinerary": "", "optimized_itinerary": "", "feedback": "", "retry_count": 0, } result = graph.invoke(initial_state) print(result["optimized_itinerary"])

运行python run_local.py,你会看到终端先输出检索到的知识片段数量,然后输出一份优化后的行程。成功结果的特征是:行程里每天的景点数量不超过 2 个(低强度约束生效)、玉龙雪山部分没有出现大索道(老人小孩约束生效)、总预算在 2 万以内。如果这些约束都体现在输出里,说明 Harness 的约束传递链路是通的。

再验证一次循环分支:把feedback初始值改成reject,重新跑,观察日志里plan节点是否被执行了两次。如果第二次执行后feedback仍为reject,图会一直循环,所以实际使用时要加retry_count上限,比如超过 3 次就强制结束并提示用户。这个上限在human_review节点里判断。

验证通过后,你可以把retrieve_knowledge节点里的检索 query 打印出来,确认它确实把“云南 7 天 老人 小孩 低强度”这些关键词带进去了。RAG 链路是否生效,看检索结果里有没有出现目的地知识文件里的具体条目,比如某个景点的开放时间或交通方式。如果检索结果为空,检查data/destinations/下有没有放知识文件,以及ingest.py有没有跑过入库。

5. 本篇常见错排查

第一个高频错误是 401 未授权。报错长这样:openai.AuthenticationError: Error code: 401 - {'error': {'message': 'Invalid API key'}}。原因通常是.env文件没被加载,或者 Key 复制时带了空格。排查动作:在settings.py里加一行print(settings.TAOTOKEN_API_KEY[:8]),确认前 8 位和你控制台看到的一致。如果打印出来是None,说明load_dotenv()没找到.env,检查文件是不是在项目根目录。

第二个错误是local proxy failed或连接超时。这类报错通常出现在base_url写错的情况下,比如写成了https://taotoken.net/api/v1或者带了多余斜杠。正确写法就是https://taotoken.net/api。另外检查你的运行环境有没有设置全局代理变量,如果有,把它清掉再跑。这个错误和网络环境有关,但不需要任何额外工具,只要地址对、环境干净就能通。

第三个错误是reading 'choices'相关的 KeyError,报错类似KeyError: 'choices'或TypeError: 'NoneType' object is not subscriptable。这通常发生在模型返回体结构和你预期不一致时。排查动作:在chat()方法里把resp完整打印出来,看返回的 JSON 里有没有choices字段。如果返回的是错误信息对象,说明请求本身失败了,先解决请求问题。如果返回正常但没有choices,检查你用的模型 ID 是否在 TaoToken 控制台的可用列表里。

第四个错误是 OAuth 或授权相关报错,比如OAuth token expired。如果你在 Harness 里接了需要 OAuth 的第三方工具(比如日历、地图),token 过期会抛这个错。排查动作:把工具调用的授权刷新逻辑单独抽出来,在调度节点里加一个 token 有效性检查,过期就重新走授权流程。注意不要在 Harness 里硬编码任何 token。

第五个错误是 LangGraph 状态字段丢失,报错KeyError: 'retrieved_docs'。原因是某个节点返回的字典没有包含 State 里定义的全部字段。LangGraph 要求每个节点返回的字典是 State 的子集,但未返回的字段会保留之前的值。如果你在节点里直接返回了一个全新的字典而漏了字段,后续节点读不到就会报错。排查动作:每个节点统一用{**state, "字段名": 新值}的方式返回,确保不丢字段。

第六个错误是 RAG 检索结果为空但没报错。这最隐蔽,因为流程能跑完,只是行程里没有目的地具体信息。排查动作:在retrieve_knowledge节点里打印len(state["retrieved_docs"]),如果是 0,检查向量库是否已入库。跑一次python rag/ingest.py,确认data/vector_store目录下有文件生成。如果入库了但检索为空,把TOP_K调大试试,或者检查 query 文本是否和知识文件的语言一致。

6. 继续接入与验证入口

骨架跑通之后,下一步是把模型对话验证和长期编码分开处理。如果你只是想确认某个模型在旅行规划提示词下的输出质量,可以直接在模型对话页面里粘贴你的提示词做对比测试,入口是 https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite 。这个页面适合快速试提示词,不用改代码。

如果你打算把这套 Harness 长期用于编码和 Agent 开发,比如后面要接更多工具、加更多节点、做多轮迭代,建议走 Coding Plan,入口在 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite 。它更适合持续性的开发场景,Key 和通道管理也更省心。

接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite ,里面有你换模型、换参数时需要对照的接口说明。API Keys 管理还是回到控制台 https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite ,建议给 Harness 单独建一个 Key,方便按项目追踪用量。

最后说一个我踩过的坑:LangGraph 的循环分支一定要设上限,否则用户一直点“不满意”,图会无限跑下去,模型调用次数会失控。在human_review节点里判断retry_count,超过 3 次就返回一个带提示的结束状态,让用户手动改偏好而不是继续自动重规划。这个上限值你可以按自己的预算调,但一定要有。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进