From 779c956647bc1fb5483775d988e3b605f9c27018 Mon Sep 17 00:00:00 2001 From: "wuchengan.2003" Date: Tue, 28 Jul 2026 15:46:23 +0800 Subject: [PATCH] feat: add langgraph migration sample --- .../migration/langgraph/README.md | 136 +++++++++ .../migration/langgraph/README_EN.md | 137 +++++++++ .../migration/langgraph/agent.py | 265 ++++++++++++++++++ .../migration/langgraph/requirements.txt | 6 + 4 files changed, 544 insertions(+) create mode 100644 python/03-integrations/migration/langgraph/README.md create mode 100644 python/03-integrations/migration/langgraph/README_EN.md create mode 100644 python/03-integrations/migration/langgraph/agent.py create mode 100644 python/03-integrations/migration/langgraph/requirements.txt diff --git a/python/03-integrations/migration/langgraph/README.md b/python/03-integrations/migration/langgraph/README.md new file mode 100644 index 00000000..885645c8 --- /dev/null +++ b/python/03-integrations/migration/langgraph/README.md @@ -0,0 +1,136 @@ +# LangGraph 项目适配 AgentKit Runtime 示例 + +本示例将演示如何将 LangGraph 项目适配到 AgentKit Runtime 上。 + +示例项目模拟一个用户已有的 LangGraph 旅行规划项目。该项目的业务入口是 `agent.py:agent`，类型是已编译的 `StateGraph`。它接收用户的旅行问题后，会通过 LangGraph 的图编排能力，把一次旅行规划拆成多个节点执行：先解析需求，再检索旅行上下文和预算信息，最后汇总成每天的景点、美食和交通建议。 + +示例中的工具用于模拟真实 LangGraph 项目中的 tool use： + +- `search_travel_web`：模拟依赖外部知识检索的工具，内部调用 `veadk.tools.builtin_tools.web_search` +- `estimate_trip_budget`：模拟本地业务计算工具，根据城市、天数和预算生成预算判断 + +`agent.py` 是一个基于 LangGraph 构建的 Agent，重点模拟用户使用 LangGraph 搭建 agent 的真实使用场景： + +- `StateGraph(TravelState)`：定义旅行规划的共享状态，并编译为 `agent.py:agent` +- `parse_request` 节点：解析用户问题中的城市、天数、预算、同行人和偏好 +- `search_travel_context` 节点：调用 `search_travel_web` 和 `estimate_trip_budget`，把外部知识和本地预算判断写回 graph state +- `build_final_answer` 节点：读取前面节点写入的 state，汇总搜索上下文、预算判断和行程安排 +- `add_edge`：声明节点执行顺序，让请求沿着 `START -> parse_request -> search_travel_context -> build_final_answer -> END` 流转 +- `InMemorySaver`：保留同一个 `thread_id` 下的会话状态，模拟真实 LangGraph workflow 的状态延续 + +适配到 AgentKit Runtime 时，不需要改写 `agent.py` 的业务逻辑。`agentkit migrate` 会生成 `agentkit_app.py` 和 `.agentkit/` 配置；生成的 Runtime 应用通过 `LangGraphAgentkitBridge(input_key="question")` 调用原始 `agent.py:agent`。 + +## 适配后的图编排调用链路 + +适配前，用户可以直接调用 `agent.py:agent`。适配后，AgentKit Runtime 会通过生成的 `agentkit_app.py` 调用同一个入口；进入 `agent.py:agent` 后，执行逻辑仍然由 LangGraph 的节点和边驱动： + +```text +用户问题 + ↓ +AgentKit Runtime + ↓ +agentkit_app.py + ↓ +LangGraphAgentkitBridge(input_key="question") + ↓ +agent.py:agent # compiled StateGraph + ├── parse_request + ├── search_travel_context + │ ├── search_travel_web + │ │ └── veadk.tools.builtin_tools.web_search + │ └── estimate_trip_budget + └── build_final_answer +``` + +## 目录结构 + +```bash +langgraph/ +├── README.md +├── agent.py # 原生 LangGraph graph、节点和 tools +├── requirements.txt # Python 依赖 +└── tests # 本地行为测试和迁移链路回归测试 +``` + +## 本地运行 + +安装依赖： + +```bash +pip install -r requirements.txt +``` + +直接运行原生 Graph： + +```bash +python agent.py +``` + +运行测试： + +```bash +python -m unittest discover -s tests -v +``` + +测试会直接覆盖 `search_travel_web` 的真实工具调用链路。 + +## 搜索配置 + +`search_travel_web` 直接使用 `veadk.tools.builtin_tools.web_search`。本地或云端运行时，请参考其它 samples 的通用方式，先在 [AgentKit 控制台授权页面](https://console.volcengine.com/agentkit/region:agentkit+cn-beijing/auth?projectName=default) 完成依赖服务授权，并配置火山引擎 AK/SK： + +```bash +VOLCENGINE_ACCESS_KEY= +VOLCENGINE_SECRET_KEY= +``` + +如果环境没有搜索权限，工具会返回搜索失败说明，Graph 仍会按示例逻辑生成可读结果。 + +## 执行迁移 + +在当前目录执行： + +```bash +agentkit migrate . \ + --framework langgraph \ + --entry agent.py:agent \ + --name migration-langgraph-travel \ + --input-key question \ + --verify +``` + +参数含义： + +- `--framework langgraph`：按 LangGraph compiled graph 方式迁移 +- `--entry agent.py:agent`：指定原生 Graph 入口 +- `--input-key question`：把 Runtime 输入写入 `question` 字段 +- `--verify`：生成后执行基础校验 + +迁移会生成： + +```bash +langgraph/ +├── agentkit_app.py +├── .agentkit/ +│ ├── agentkit.yaml +│ ├── Dockerfile +│ └── migration-plan.json +└── requirements.txt +``` + +迁移命令不会改写 `agent.py`。生成的 Runtime 应用会通过 `LangGraphAgentkitBridge(input_key="question")` 调用原始 `agent.py:agent`，并将 AgentKit 会话映射到 LangGraph `thread_id`。 + +## 部署到 AgentKit Runtime + +确认 `.agentkit/agentkit.yaml` 后执行： + +```bash +agentkit deploy +``` + +部署后，Runtime 入口是 `agentkit_app.py`，业务逻辑仍由 `agent.py:agent` 中的 LangGraph 节点、checkpointer 和原有 tools 执行。 + +## 示例问题 + +```text +我想带父母去北京玩3天，总预算3000元，喜欢历史文化、胡同和老北京美食，行程轻松一点。请帮我规划每天的景点、美食和交通建议。 +``` diff --git a/python/03-integrations/migration/langgraph/README_EN.md b/python/03-integrations/migration/langgraph/README_EN.md new file mode 100644 index 00000000..9268a7a4 --- /dev/null +++ b/python/03-integrations/migration/langgraph/README_EN.md @@ -0,0 +1,137 @@ +# LangGraph Project Adaptation to AgentKit Runtime Sample + +This sample shows how to adapt a LangGraph project to AgentKit Runtime. + +The sample project represents an existing LangGraph travel-planning project that a user already has. Its business entry point is `agent.py:agent`, implemented as a compiled `StateGraph`. After receiving a user's travel request, it uses LangGraph orchestration to split one travel-planning task into multiple nodes: first parsing the request, then retrieving travel context and budget information, and finally summarizing daily attraction, food, and transportation suggestions. + +The tools in this sample simulate tool use in a real LangGraph project: + +- `search_travel_web`: simulates a tool that depends on external knowledge retrieval, and internally calls `veadk.tools.builtin_tools.web_search` +- `estimate_trip_budget`: simulates a local business calculation tool that evaluates the budget based on city, number of days, and total budget + +`agent.py` is an Agent built with LangGraph. It focuses on simulating a realistic scenario where users build an agent with LangGraph: + +- `StateGraph(TravelState)`: defines the shared state for travel planning and compiles it as `agent.py:agent` +- `parse_request` node: parses the city, number of days, budget, travelers, and preferences from the user's question +- `search_travel_context` node: calls `search_travel_web` and `estimate_trip_budget`, then writes external knowledge and local budget evaluation back into the graph state +- `build_final_answer` node: reads the state written by previous nodes and summarizes search context, budget evaluation, and itinerary planning +- `add_edge`: declares node execution order so the request flows through `START -> parse_request -> search_travel_context -> build_final_answer -> END` +- `InMemorySaver`: preserves session state under the same `thread_id`, simulating state continuity in a real LangGraph workflow + +When adapting the project to AgentKit Runtime, you do not need to rewrite the business logic in `agent.py`. `agentkit migrate` generates `agentkit_app.py` and `.agentkit/` configuration. The generated Runtime app calls the original `agent.py:agent` through `LangGraphAgentkitBridge(input_key="question")`. + +## Adapted Graph Orchestration Flow + +Before adaptation, users can call `agent.py:agent` directly. After adaptation, AgentKit Runtime calls the same entry point through the generated `agentkit_app.py`. Once execution enters `agent.py:agent`, the logic is still driven by LangGraph nodes and edges: + +```text +User question + | +AgentKit Runtime + | +agentkit_app.py + | +LangGraphAgentkitBridge(input_key="question") + | +agent.py:agent # compiled StateGraph + |-- parse_request + |-- search_travel_context + | |-- search_travel_web + | | `-- veadk.tools.builtin_tools.web_search + | `-- estimate_trip_budget + `-- build_final_answer +``` + +## Directory Layout + +```bash +langgraph/ +├── README.md +├── README_EN.md +├── agent.py # Native LangGraph graph, nodes, and tools +├── requirements.txt # Python dependencies +└── tests # Local behavior tests and migration-chain regression tests +``` + +## Local Run + +Install dependencies: + +```bash +pip install -r requirements.txt +``` + +Run the native Graph directly: + +```bash +python agent.py +``` + +Run tests: + +```bash +python -m unittest discover -s tests -v +``` + +The tests directly cover the real tool-call path of `search_travel_web`. + +## Search Configuration + +`search_travel_web` directly uses `veadk.tools.builtin_tools.web_search`. For local or cloud execution, follow the common setup used by other samples: authorize dependent services in the [AgentKit Console authorization page](https://console.volcengine.com/agentkit/region:agentkit+cn-beijing/auth?projectName=default), then configure Volcengine AK/SK: + +```bash +VOLCENGINE_ACCESS_KEY= +VOLCENGINE_SECRET_KEY= +``` + +If the environment has no search permission, the tool returns a search failure message. The Graph still returns a readable sample response. + +## Run Migration + +Run this command in the current directory: + +```bash +agentkit migrate . \ + --framework langgraph \ + --entry agent.py:agent \ + --name migration-langgraph-travel \ + --input-key question \ + --verify +``` + +Arguments: + +- `--framework langgraph`: migrate as a LangGraph compiled graph +- `--entry agent.py:agent`: specify the native Graph entry point +- `--input-key question`: write Runtime input into the `question` field +- `--verify`: run basic checks after generation + +Migration generates: + +```bash +langgraph/ +├── agentkit_app.py +├── .agentkit/ +│ ├── agentkit.yaml +│ ├── Dockerfile +│ └── migration-plan.json +└── requirements.txt +``` + +The migration command does not rewrite `agent.py`. The generated Runtime app calls the original `agent.py:agent` through `LangGraphAgentkitBridge(input_key="question")` and maps the AgentKit session to the LangGraph `thread_id`. + +## Deploy To AgentKit Runtime + +After reviewing `.agentkit/agentkit.yaml`, run: + +```bash +agentkit deploy +``` + +After deployment, the Runtime entry point is `agentkit_app.py`. The business logic is still handled by the LangGraph nodes, checkpointer, and original tools in `agent.py:agent`. + +## Example Prompt + +```text +I want to take my parents to Beijing for 3 days with a total budget of 3000 RMB. We like history and culture, hutongs, and old Beijing food. Please keep the itinerary relaxed and plan attractions, food, and transportation for each day. +``` diff --git a/python/03-integrations/migration/langgraph/agent.py b/python/03-integrations/migration/langgraph/agent.py new file mode 100644 index 00000000..5b880ace --- /dev/null +++ b/python/03-integrations/migration/langgraph/agent.py @@ -0,0 +1,265 @@ +from __future__ import annotations + +import re +from typing import Any, TypedDict + +from langchain_core.tools import tool +from langgraph.checkpoint.memory import InMemorySaver +from langgraph.graph import END, START, StateGraph +from veadk.tools.builtin_tools.web_search import web_search as builtin_web_search + + +class TravelState(TypedDict, total=False): + question: str + city: str + days: int + budget: int + interests: list[str] + travelers: str + search_query: str + search_context: str + budget_note: str + answer: str + request_count: int + + +def _format_web_search_results(results: Any) -> str: + if isinstance(results, str): + return results + if isinstance(results, list): + return "\n".join(str(result).strip() for result in results if str(result).strip()) + return str(results) + + +@tool +def search_travel_web(query: str) -> str: + """根据用户旅行需求进行联网搜索，返回可用于规划的摘要。""" + try: + result = _format_web_search_results(builtin_web_search(query)) + except Exception as exc: + return f"联网搜索失败：{exc}。搜索词：{query}" + return result or f"联网搜索没有返回可解析结果。搜索词：{query}" + + +@tool +def estimate_trip_budget(city: str, days: int, budget: int) -> str: + """估算国内城市旅行预算是否宽松。""" + daily = budget // max(days, 1) + if daily >= 1000: + level = "比较宽松" + elif daily >= 650: + level = "中等可控" + else: + level = "偏紧，需要压缩住宿和餐饮成本" + return f"{city}{days}天总预算{budget}元，人均每日约{daily}元，预算判断：{level}。" + + +def _parse_city(question: str, default: str = "北京") -> str: + direct_patterns = ( + r"(?:去|到)([\u4e00-\u9fff]{2,6})(?:玩|旅游|旅行)", + r"([\u4e00-\u9fff]{2,6})(?:玩|旅游|旅行)", + ) + for pattern in direct_patterns: + match = re.search(pattern, question) + if match: + return match.group(1) + + city_hints = ("北京", "上海", "杭州", "成都", "西安", "南京", "重庆", "广州", "深圳") + for city in city_hints: + if city in question: + return city + return default + + +def _parse_days(question: str, default: int = 3) -> int: + match = re.search(r"(\d+)\s*天", question) + return int(match.group(1)) if match else default + + +def _parse_budget(question: str, default: int = 3000) -> int: + match = re.search(r"(?:预算|总预算)?\s*(\d{3,5})\s*元", question) + return int(match.group(1)) if match else default + + +def _parse_travelers(question: str, default: str = "普通出行") -> str: + if "父母" in question or "长辈" in question: + return "带父母/长辈" + if "孩子" in question or "亲子" in question: + return "亲子" + if "同学" in question or "朋友" in question: + return "朋友同行" + if "一个人" in question or "独自" in question: + return "独自旅行" + return default + + +def _parse_interests(question: str) -> list[str]: + interests = [] + candidates = { + "历史文化": ("历史", "文化", "故宫", "博物馆", "遗迹"), + "胡同街区": ("胡同", "Citywalk", "街区"), + "亲子活动": ("孩子", "亲子", "博物馆"), + "城市景观": ("夜景", "城市", "轻轨", "外滩"), + "当地美食": ("美食", "火锅", "小吃", "老北京", "餐饮"), + "轻松慢游": ("轻松", "不想走太多路", "不太累", "休闲"), + } + for label, words in candidates.items(): + if any(word in question for word in words): + interests.append(label) + return interests or ["经典景点", "当地美食"] + + +def parse_request(state: TravelState) -> TravelState: + question = state.get("question", "") + city = _parse_city(question, state.get("city", "北京")) + days = _parse_days(question, int(state.get("days", 3))) + budget = _parse_budget(question, int(state.get("budget", 3000))) + travelers = _parse_travelers(question, state.get("travelers", "普通出行")) + return { + "question": question, + "city": city, + "days": days, + "budget": budget, + "travelers": travelers, + "interests": _parse_interests(question), + "request_count": int(state.get("request_count", 0)) + 1, + } + + +def _build_search_query(state: TravelState) -> str: + city = state.get("city", "北京") + days = int(state.get("days", 3)) + budget = int(state.get("budget", 3000)) + travelers = state.get("travelers", "普通出行") + interests = state.get("interests", ["经典景点", "当地美食"]) + parts = [ + city, + f"{days}天", + f"{budget}元", + travelers, + *interests, + "旅游", + "景点", + "美食", + "交通", + "预约", + "注意事项", + ] + return " ".join(part for part in parts if part and part != "普通出行") + + +def search_travel_context(state: TravelState) -> TravelState: + city = state.get("city", "北京") + days = int(state.get("days", 3)) + budget = int(state.get("budget", 3000)) + query = _build_search_query(state) + return { + "search_query": query, + "search_context": search_travel_web.invoke({"query": query}), + "budget_note": estimate_trip_budget.invoke( + {"city": city, "days": days, "budget": budget} + ), + } + + +def _unique(values: list[str]) -> list[str]: + seen = set() + result = [] + for value in values: + normalized = value.strip(" ，,；;。：:") + if not normalized or normalized in seen: + continue + seen.add(normalized) + result.append(normalized) + return result + + +def _extract_terms(context: str, suffixes: tuple[str, ...], fallback: list[str]) -> list[str]: + suffix_pattern = "|".join(re.escape(suffix) for suffix in suffixes) + terms = re.findall(rf"[\u4e00-\u9fffA-Za-z0-9]{{2,18}}(?:{suffix_pattern})", context) + return _unique(terms)[:5] or fallback + + +def _attractions_from_context(context: str) -> list[str]: + return _extract_terms( + context, + ("博物院", "博物馆", "公园", "胡同", "天坛", "景区", "街区", "场馆"), + ["联网搜索结果中的核心景点", "同一区域可串联景点"], + ) + + +def _foods_from_context(context: str) -> list[str]: + return _extract_terms( + context, + ("烤鸭", "炸酱面", "涮肉", "火锅", "小吃", "美食", "餐饮"), + ["当地代表性美食", "交通便利区域餐厅"], + ) + + +def _day_plan(day: int, attractions: list[str], foods: list[str], travelers: str) -> str: + morning = attractions[(day - 1) % len(attractions)] + afternoon = attractions[day % len(attractions)] + lunch = foods[(day - 1) % len(foods)] + dinner = foods[day % len(foods)] + pace_note = ( + "下午预留休息时间，减少连续步行。" + if "父母" in travelers or "长辈" in travelers + else "下午安排同一区域活动，避免来回折返。" + ) + return "\n".join( + [ + f"第{day}天：{morning} + {afternoon}", + f"- 上午：优先安排{morning}，出发前确认预约和开放时间。", + f"- 午餐：结合联网搜索结果尝试{lunch}，选择离上午景点较近的位置。", + f"- 下午：前往{afternoon}，{pace_note}", + f"- 晚餐：安排{dinner}，餐后就近返回住宿区域。", + ] + ) + + +def build_final_answer(state: TravelState) -> TravelState: + city = state.get("city", "北京") + days = int(state.get("days", 3)) + budget = int(state.get("budget", 3000)) + travelers = state.get("travelers", "普通出行") + search_context = state.get("search_context", "") + attractions = _attractions_from_context(search_context) + foods = _foods_from_context(search_context) + plans = "\n\n".join( + _day_plan(day, attractions, foods, travelers) for day in range(1, days + 1) + ) + lines = [ + f"{city}{days}天旅行规划（预算{budget}元，{travelers}，第{state.get('request_count', 1)}次规划）", + "", + f"需求偏好：{', '.join(state.get('interests', ['经典景点']))}。", + f"联网搜索：{search_context}", + f"预算建议：{state.get('budget_note', '')}", + "", + plans, + "", + "交通建议：优先选择地铁和短距离打车，连续景点尽量按同一区域串联。", + "说明：这是 LangGraph 迁移示例，旅行上下文来自搜索 tool；搜索能力由 veadk.tools.builtin_tools.web_search 提供。", + ] + return {"answer": "\n".join(lines)} + + +builder = StateGraph(TravelState) +builder.add_node("parse_request", parse_request) +builder.add_node("search_travel_context", search_travel_context) +builder.add_node("build_final_answer", build_final_answer) +builder.add_edge(START, "parse_request") +builder.add_edge("parse_request", "search_travel_context") +builder.add_edge("search_travel_context", "build_final_answer") +builder.add_edge("build_final_answer", END) + +agent = builder.compile(checkpointer=InMemorySaver()) + + +if __name__ == "__main__": + result = agent.invoke( + { + "question": "我想带父母去北京玩3天，总预算3000元，喜欢历史文化、胡同和老北京美食，行程轻松一点。请帮我规划每天的景点、美食和交通建议。" + }, + config={"configurable": {"thread_id": "local-demo"}}, + ) + print(result["answer"]) diff --git a/python/03-integrations/migration/langgraph/requirements.txt b/python/03-integrations/migration/langgraph/requirements.txt new file mode 100644 index 00000000..d140ebba --- /dev/null +++ b/python/03-integrations/migration/langgraph/requirements.txt @@ -0,0 +1,6 @@ +langchain-core +langgraph +veadk-python==0.5.37 +a2a-sdk>=0.3.7,<0.4 +agentkit-sdk-python>=0.7.12 +google-adk>=1.32