From 5df764a0e2eb04426b842cb9b3fe06ea9c26aa2b Mon Sep 17 00:00:00 2001 From: "wuchengan.2003" Date: Tue, 28 Jul 2026 15:53:31 +0800 Subject: [PATCH] feat: add strands migration sample --- .../migration/strands/README.md | 133 +++++++++ .../migration/strands/README_EN.md | 134 +++++++++ .../migration/strands/agent.py | 280 ++++++++++++++++++ .../migration/strands/requirements.txt | 4 + 4 files changed, 551 insertions(+) create mode 100644 python/03-integrations/migration/strands/README.md create mode 100644 python/03-integrations/migration/strands/README_EN.md create mode 100644 python/03-integrations/migration/strands/agent.py create mode 100644 python/03-integrations/migration/strands/requirements.txt diff --git a/python/03-integrations/migration/strands/README.md b/python/03-integrations/migration/strands/README.md new file mode 100644 index 00000000..5ee00c7d --- /dev/null +++ b/python/03-integrations/migration/strands/README.md @@ -0,0 +1,133 @@ +# Strands 项目适配 AgentKit Runtime 示例 + +本示例将演示如何将 Strands 项目适配到 AgentKit Runtime 上。 + +示例项目模拟一个用户已有的 Strands 旅行规划项目。该项目的业务入口是 `agent.py:build_agent`，它创建并返回一个 Strands `Agent`。Agent 接收用户的旅行问题后，会通过 Strands 的 Agent + tools 运行方式，把旅行规划能力组织成可注册、可调试、可迁移的工具调用链路。 + +示例中的工具用于模拟真实 Strands 项目中的 tool use： + +- `search_travel_web`：模拟依赖外部知识检索的工具，内部调用 `veadk.tools.builtin_tools.web_search` +- `estimate_trip_budget`：模拟本地业务计算工具，根据城市、天数和预算生成预算判断 + +`agent.py` 是一个基于 Strands 构建的 Agent，重点展示原生 Strands 项目常见的 Agent 工厂和工具注册方式： + +- `build_agent()`：创建原生 Strands `Agent`，并作为 `agent.py:build_agent` 暴露给迁移命令 +- `TRAVEL_TOOLS`：集中注册 `search_travel_web` 和 `estimate_trip_budget` +- `@tool`：把普通 Python 函数声明为 Strands tools，让 Agent 可以按工具调用方式使用它们 +- `search_travel_web`：在工具内部调用 `veadk.tools.builtin_tools.web_search`，模拟真实项目中依赖外部知识检索的能力 +- `estimate_trip_budget`：保留本地业务计算逻辑，模拟真实项目中的内部工具 +- `LocalTravelModel`：让样例在本地调试时可以返回稳定的可读结果，同时保留 Strands `Agent` 的运行入口和工具配置 + +适配到 AgentKit Runtime 时，不需要改写 `agent.py` 的业务逻辑。`agentkit migrate` 会生成 `agentkit_app.py` 和 `.agentkit/` 配置；生成的 Runtime 应用通过 `StrandsAgentkitBridge(agent_factory=True)` 调用原始 `agent.py:build_agent`。 + +## 适配后的 Agent 调用链路 + +适配前，用户可以直接调用 `agent.py:build_agent` 创建 Strands Agent。适配后，AgentKit Runtime 会通过生成的 `agentkit_app.py` 调用同一个入口；进入 `agent.py:build_agent` 后，业务逻辑仍然由 Strands `Agent` 和已注册的 tools 执行： + +```text +用户问题 + ↓ +AgentKit Runtime + ↓ +agentkit_app.py + ↓ +StrandsAgentkitBridge(agent_factory=True) + ↓ +agent.py:build_agent + ↓ +Strands Agent + ├── search_travel_web + │ └── veadk.tools.builtin_tools.web_search + └── estimate_trip_budget +``` + +## 目录结构 + +```bash +strands/ +├── README.md +├── agent.py # 原生 Strands Agent、工厂入口和 tools +├── requirements.txt # Python 依赖 +└── tests # 本地行为测试和迁移链路回归测试 +``` + +## 本地运行 + +安装依赖： + +```bash +pip install -r requirements.txt +``` + +直接运行原生 Agent： + +```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= +``` + +如果环境没有搜索权限，工具会返回搜索失败说明，Agent 仍会按示例逻辑生成可读结果。 + +## 执行迁移 + +在当前目录执行： + +```bash +agentkit migrate . \ + --framework strands \ + --entry agent.py:build_agent \ + --name migration-strands-travel \ + --verify +``` + +参数含义： + +- `--framework strands`：按 Strands Agent 方式迁移 +- `--entry agent.py:build_agent`：指定原生 Strands Agent 工厂入口 +- `--verify`：生成后执行基础校验 + +迁移会生成： + +```bash +strands/ +├── agentkit_app.py +├── .agentkit/ +│ ├── agentkit.yaml +│ ├── Dockerfile +│ └── migration-plan.json +└── requirements.txt +``` + +迁移命令不会改写 `agent.py`。生成的 Runtime 应用会通过 `StrandsAgentkitBridge(agent_factory=True)` 调用原始 `agent.py:build_agent`。 + +## 部署到 AgentKit Runtime + +确认 `.agentkit/agentkit.yaml` 后执行： + +```bash +agentkit deploy +``` + +部署后，Runtime 入口是 `agentkit_app.py`，业务逻辑仍由 `agent.py:build_agent` 创建的 Strands Agent 和原有 tools 执行。 + +## 示例问题 + +```text +我想带父母去北京玩3天，总预算3000元，喜欢历史文化、胡同和老北京美食，行程轻松一点。请帮我规划每天的景点、美食和交通建议。 +``` diff --git a/python/03-integrations/migration/strands/README_EN.md b/python/03-integrations/migration/strands/README_EN.md new file mode 100644 index 00000000..9595bfcd --- /dev/null +++ b/python/03-integrations/migration/strands/README_EN.md @@ -0,0 +1,134 @@ +# Strands Project Adaptation to AgentKit Runtime Sample + +This sample shows how to adapt a Strands project to AgentKit Runtime. + +The sample project simulates an existing Strands travel-planning project. Its business entry point is `agent.py:build_agent`, which creates and returns a Strands `Agent`. After receiving a user's travel request, the Agent uses the Strands Agent + tools execution model to organize travel-planning capabilities into a tool-call chain that can be registered, debugged, and migrated. + +The tools in this sample simulate tool use in a real Strands 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 Strands. It focuses on common native Strands project patterns for Agent factories and tool registration: + +- `build_agent()`: creates a native Strands `Agent` and exposes it as `agent.py:build_agent` for the migration command +- `TRAVEL_TOOLS`: centrally registers `search_travel_web` and `estimate_trip_budget` +- `@tool`: declares regular Python functions as Strands tools so the Agent can use them through tool calls +- `search_travel_web`: calls `veadk.tools.builtin_tools.web_search` inside the tool, simulating a real project's dependency on external knowledge retrieval +- `estimate_trip_budget`: preserves local business calculation logic, simulating an internal tool in a real project +- `LocalTravelModel`: lets the sample return stable, readable results during local debugging while preserving the Strands `Agent` execution entry point and tool configuration + +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:build_agent` through `StrandsAgentkitBridge(agent_factory=True)`. + +## Adapted Agent Call Flow + +Before adaptation, users can directly call `agent.py:build_agent` to create the Strands Agent. After adaptation, AgentKit Runtime calls the same entry point through the generated `agentkit_app.py`. Once execution enters `agent.py:build_agent`, the business logic is still handled by the Strands `Agent` and the registered tools: + +```text +User question + | +AgentKit Runtime + | +agentkit_app.py + | +StrandsAgentkitBridge(agent_factory=True) + | +agent.py:build_agent + | +Strands Agent + |-- search_travel_web + | `-- veadk.tools.builtin_tools.web_search + `-- estimate_trip_budget +``` + +## Directory Layout + +```bash +strands/ +├── README.md +├── README_EN.md +├── agent.py # Native Strands Agent, factory entry point, 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 Agent 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 Agent still returns a readable sample response. + +## Run Migration + +Run this command in the current directory: + +```bash +agentkit migrate . \ + --framework strands \ + --entry agent.py:build_agent \ + --name migration-strands-travel \ + --verify +``` + +Arguments: + +- `--framework strands`: migrate as a Strands Agent +- `--entry agent.py:build_agent`: specify the native Strands Agent factory entry point +- `--verify`: run basic checks after generation + +Migration generates: + +```bash +strands/ +├── 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:build_agent` through `StrandsAgentkitBridge(agent_factory=True)`. + +## 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 Strands Agent created by `agent.py:build_agent` and the original tools. + +## 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/strands/agent.py b/python/03-integrations/migration/strands/agent.py new file mode 100644 index 00000000..51541f78 --- /dev/null +++ b/python/03-integrations/migration/strands/agent.py @@ -0,0 +1,280 @@ +from __future__ import annotations + +import re +from typing import Any + +from strands import Agent, tool +from strands.models import Model +from veadk.tools.builtin_tools.web_search import web_search as builtin_web_search + + +SYSTEM_PROMPT = ( + "你是北京及中国本地旅行规划助手，需要结合联网搜索、预算判断和用户偏好，" + "给出可执行的每日景点、美食和交通建议。" +) + + +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}。" + + +TRAVEL_TOOLS = [search_travel_web, estimate_trip_budget] + + +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 _build_search_query( + city: str, + days: int, + budget: int, + travelers: str, + interests: list[str], +) -> str: + parts = [ + city, + f"{days}天", + f"{budget}元", + travelers, + *interests, + "旅游", + "景点", + "美食", + "交通", + "预约", + "注意事项", + ] + return " ".join(part for part in parts if part and part != "普通出行") + + +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_itinerary(question: str) -> str: + city = _parse_city(question) + days = _parse_days(question) + budget = _parse_budget(question) + travelers = _parse_travelers(question) + interests = _parse_interests(question) + search_query = _build_search_query(city, days, budget, travelers, interests) + search_context = search_travel_web(query=search_query) + budget_result = estimate_trip_budget(city=city, days=days, budget=budget) + 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) + ) + return ( + f"{city}{days}天旅行规划（预算{budget}元，{travelers}）\n\n" + f"需求摘要：偏好{', '.join(interests)}。\n" + f"联网搜索：{search_context}\n" + f"预算建议：{budget_result}\n\n" + f"{plans}\n\n" + "交通建议：优先选择地铁和短距离打车，连续景点尽量按同一区域串联。\n" + "说明：这是 Strands 迁移示例，旅行上下文来自 Strands tool；搜索能力由 veadk.tools.builtin_tools.web_search 提供。" + ) + + +class LocalTravelModel(Model): + """用于本地调试和样例测试的 Strands Model。""" + + def update_config(self, **model_config: Any) -> None: + self.model_config = model_config + + def get_config(self) -> dict[str, Any]: + return getattr(self, "model_config", {}) + + async def structured_output( + self, + output_model, + prompt, + system_prompt=None, + **kwargs: Any, + ): + del prompt, system_prompt, kwargs + yield {"output": output_model()} + + async def stream( + self, + messages, + tool_specs=None, + system_prompt=None, + **kwargs: Any, + ): + del tool_specs, system_prompt, kwargs + user_text = messages[-1]["content"][0]["text"] + if "旅游" in user_text or "旅行" in user_text or "玩" in user_text: + text = build_itinerary(user_text) + else: + text = f"Strands 北京旅游规划助手：我可以根据预算、天数和偏好规划北京旅游行程。收到：{user_text}" + + yield {"messageStart": {"role": "assistant"}} + yield {"contentBlockStart": {"start": {}}} + yield {"contentBlockDelta": {"delta": {"text": text}}} + yield {"contentBlockStop": {}} + yield {"messageStop": {"stopReason": "end_turn"}} + yield { + "metadata": { + "usage": { + "inputTokens": 1, + "outputTokens": max(1, len(text) // 4), + "totalTokens": max(2, len(text) // 4 + 1), + }, + "metrics": {"latencyMs": 0}, + } + } + + +def build_agent() -> Agent: + """创建可被 agentkit migrate 识别的 Strands Agent。""" + return Agent( + name="strands_travel_planner", + model=LocalTravelModel(), + tools=TRAVEL_TOOLS, + system_prompt=SYSTEM_PROMPT, + callback_handler=None, + ) + + +agent = build_agent() + + +def invoke_agent(prompt: str) -> str: + """使用 Strands Agent 调试入口，并返回可读文本。""" + return str(agent(prompt)) + + +if __name__ == "__main__": + demo = "我想带父母去北京玩3天，总预算3000元，喜欢历史文化和轻松一点的行程。请帮我规划每天的景点、美食和交通建议。" + print(invoke_agent(demo)) diff --git a/python/03-integrations/migration/strands/requirements.txt b/python/03-integrations/migration/strands/requirements.txt new file mode 100644 index 00000000..7a188a23 --- /dev/null +++ b/python/03-integrations/migration/strands/requirements.txt @@ -0,0 +1,4 @@ +strands-agents +a2a-sdk>=0.3.7,<0.4 +agentkit-sdk-python>=0.7.12 +google-adk>=1.32