带你实现一个Agent，从Tools、MCP到Skills,从理论到代码，干货满满

brew update
brew install pyenv

curl https://pyenv.run | bash

curl -fsSL https://pyenv.run | bash

Windows 使用 pyenv-win：
1. 打开 PowerShell（管理员权限）。
2. 执行安装命令：
Invoke-WebRequest -UseBasicParsing -Uri "https://raw.githubusercontent.com/pyenv-win/pyenv-win/master/pyenv-win/install-pyenv-win.ps1" -OutFile "./install-pyenv-win.ps1"; &"./install-pyenv-win.ps1"

pyenv --version

# 查看可安装的Python版本
pyenv install --list | grep 3.12
# 安装Python 3.12.0
pyenv install 3.12.0
# 查看已安装的版本
pyenv versions
# 设置全局Python版本
pyenv global 3.12.0
# 为当前目录设置Python版本
pyenv local 3.12.0

pyenv install 3.12.0
pyenv global 3.12.0   # 或者在项目目录使用 pyenv local 3.12.0

curl -fsSL https://uv.dev/install.sh | bash
pip install uv-cli

pip install uv-cli

uv --version

# 初始化项目（自动创建虚拟环境）
uv init
# 查看当前项目使用的 Python 版本
uv run python --version
# 安装依赖包
uv add requests flask
uv add openai
# 卸载依赖包
uv remove requests
# 锁定当前环境依赖（生成 uv.lock）
uv lock
# 根据锁文件安装依赖
uv sync

echo "DEEPSEEK_API_KEY=youkey" > .env
echo "AMAP_API_KEY=youkey" >> .env  # 注意这里是两个大于号

# deepseek
def deepseekDemo():
print("deepseekDemo,hello")
  client = OpenAI(
      api_key=os.environ.get('DEEPSEEK_API_KEY'),
      base_url="https://api.deepseek.com")

  response = client.chat.completions.create(
      model="deepseek-chat",
      messages=[
          {"role": "system", "content": "You are a helpful assistant"},
          {"role": "user", "content": "Hello"},
      ],
      stream=False
  )
print(response.choices[0].message.content)

# 高德
def amap_demo():
  key = os.getenv("AMAP_API_KEY")
  response = httpx.Client().get(
      "https://restapi.amap.com/v3/weather/weatherInfo?key=" + key + "&city=110101&extensions=all")
print(response.json())
print(json.dumps(response.json(), indent=2,ensure_ascii=False))

class TravelTools:
  """
  旅游规划助手工具类
  """
  def __init__(self, amap_api_key: Optional[str] = None, request_delay: float = 0.2):
      self.amap_api_key = amap_api_key or os.getenv("AMAP_API_KEY")
      # ...
  async def estimate_travel_cost(self, city: str, days: int, hotel_level: str = "舒适",
                           attractions: Optional[List[str]] = None) -> Dict[str, Any]:
    """
    估算旅游费用（不含往返交通）
    使用场景：
    - 制定旅游预算
    - 比较不同档次的旅游费用
    - 规划旅游支出
    Args:
        city: 旅游城市名称，如'西安'、'北京'、'上海'
        days: 旅游天数（含当天），如3表示2晚3天
        hotel_level: 住宿档次，可选值：
                   - '经济': 150元/晚（快捷酒店）
                   - '舒适': 300元/晚（三星级酒店，默认）
                   - '豪华': 500元/晚（四星级及以上）
        attractions: 计划游览的景点列表（可选），用于估算门票费用
                   如 ['兵马俑', '华清宫', '大雁塔']
    Returns:
        费用估算字典，包含：
        - city: 城市名称
        - days: 旅游天数
        - breakdown: 费用明细（住宿、餐饮、交通、门票）
        - total: 总费用
        - tips: 温馨提示
    """
    # ... 具体实现逻辑 ...
    pass

@dataclass
class Tool:
"""工具定义数据模型"""
  name: str                    # 工具名称/函数名
  description: str             # 工具描述（用于大模型理解）
  parameters: Dict[str, Any]   # 参数JSON Schema
function: Callable           # 实际的执行函数

  def to_dict(self) -> dict:
      """转换为OpenAI Function Calling格式"""
      return {
          "type": "function",
          "function": {
              "name": self.name,
              "description": self.description,
              "parameters": self.parameters
          }
      }

class ToolRegistry:
# ...

  def register_from_class(self, cls, ...):
      """从类中自动注册异步方法为工具"""
      # 利用 Python 的反射机制（Inspect模块）
      # 自动读取工具函数的签名和文档
      # ...
      pass

  async def execute_tool(self, tool_call: ToolCall) -> ToolResult:
      """执行单个工具调用"""
      # ...
      pass

# ... 初始化 messages ...
while count < 15:
count = count + 1
# 发送请求给大模型
result = await llm_client_with_tools(messages, tool_registry.get_tools())
assistant_message = result.choices[0].message
# 检查是否有工具调用
if assistant_message.tool_calls:
    # 1. 添加 assistant 消息到历史
    messages.append({
        "role": "assistant",
        "content": assistant_message.content,
        "tool_calls": [...]
    })
    # 2. 执行所有工具调用
    for tool_call in assistant_message.tool_calls:
        # 执行工具
        tool_result = await tool_registry.execute_tool(...)

        # 3. 添加工具结果到历史
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": tool_result.content
        })
    # 继续下一轮循环，将工具结果带给模型
    continue
else:
    # 没有工具调用，输出最终回复
    print(assistant_message.content)
    break

from typing import Dict, List, Optional, Any, Annotated
from pydantic import Field
from fastmcp import FastMCP
# 导入TravelTools类
from code.Function_Calling.tools import TravelTools
mcp = FastMCP(name="旅游规划助手")
# 初始化TravelTools实例
travel_tools = TravelTools()
@mcp.tool("get_current_weather", description="获取当前天气信息")
async def get_current_weather(
        city: Annotated[str, Field(description="城市名称，如'西安'")],
        province: Annotated[str, Field(description="省份名称，如'陕西'")]
) -> Dict[str, Any]:
    try:
        weather = await travel_tools.get_weather(city, province)
        return weather.to_dict()
    except Exception as e:
        return {"error": str(e)}


@mcp.tool("geocode", description="地理编码：将地址转换为经纬度坐标")
async def geocode(
        address: Annotated[str, Field(description="地址或地点名称，如'兵马俑'、'大雁塔'、'西安市钟楼'")],
        city: Annotated[str, Field(description="城市名称（可选），用于限定搜索范围，如'西安'、'北京'")] = ""
) -> Dict[str, Any]:
    try:
        location = await travel_tools.geocode(address, city)
        return location.to_dict()
    except Exception as e:
        return {"error": str(e)}


@mcp.tool("search_poi", description="搜索兴趣点(POI) - 关键词搜索")
async def search_poi(
        keywords: Annotated[str, Field(description="搜索关键词，如'兵马俑'、'回民街'、'火锅'、'快捷酒店'")],
        city: Annotated[str, Field(description="城市名称，如'西安'、'北京'、'上海'")],
        types: Annotated[str, Field(description="POI类型筛选（可选），可选值：'景点'、'美食'、'酒店'、'交通'")] = "",
        page_size: Annotated[int, Field(description="返回结果数量，默认10，最大25")] = 10
) -> List[Dict[str, Any]]:
    try:
        pois = await travel_tools.search_poi(keywords, city, types, page_size)
        return [poi.to_dict() for poi in pois]
    except Exception as e:
        return [{"error": str(e)}]


@mcp.tool("search_nearby", description="搜索周边POI - 基于位置的周边搜索")
async def search_nearby(
        location: Annotated[str, Field(description="中心点坐标，格式为'经度,纬度'，如'109.273528,34.384926'")],
        keywords: Annotated[str, Field(description="搜索关键词（可选），如'餐厅'、'便利店'、'ATM'")] = "",
        types: Annotated[str, Field(description="POI类型筛选（可选），可选值：'景点'、'美食'、'酒店'")] = "",
        radius: Annotated[int, Field(description="搜索半径，单位：米，默认1000米（1公里）")] = 1000,
        page_size: Annotated[int, Field(description="返回结果数量，默认5，最大25")] = 5
) -> List[Dict[str, Any]]:
    try:
        pois = await travel_tools.search_nearby(location, keywords, types, radius, page_size)
        return [poi.to_dict() for poi in pois]
    except Exception as e:
        return [{"error": str(e)}]


@mcp.tool("route_planning", description="路线规划 - 多种出行方式的路线规划")
async def route_planning(
        origin: Annotated[str, Field(description="起点坐标，格式为'经度,纬度'")],
        destination: Annotated[str, Field(description="终点坐标，格式为'经度,纬度'")],
        mode: Annotated[
            str, Field(description="出行方式，可选值：'walking'（步行）、'driving'（驾车）、'transit'（公交/地铁）")] = "transit",
        city: Annotated[str, Field(description="城市名称，公交路线规划时必填，如'西安'、'北京'")] = "",
        origin_name: Annotated[str, Field(description="起点名称（可选），用于公交方案显示")] = "",
        destination_name: Annotated[str, Field(description="终点名称（可选），用于公交方案显示")] = ""
) -> Dict[str, Any]:
    try:
        route = await travel_tools.route_planning(origin, destination, mode, city, origin_name, destination_name)
        return route
    except Exception as e:
        return {"error": str(e)}


@mcp.tool("get_weather_forecast", description="查询城市天气预报（未来3-4天）")
async def get_weather_forecast(
        city: Annotated[str, Field(description="城市名称或城市编码（adcode），如'西安'、'110000'")],
        province: Annotated[str, Field(description="省份名称，如'陕西'")],
        days: Annotated[int, Field(description="预报天数，默认4天（包含当天）")] = 4
) -> Dict[str, Any]:
    try:
        forecast = await travel_tools.get_weather_forecast(city, province, days)
        return forecast
    except Exception as e:
        return {"error": str(e)}


@mcp.tool("estimate_travel_cost", description="估算旅游费用（不含往返交通）")
def estimate_travel_cost(
        city: Annotated[str, Field(description="旅游城市名称，如'西安'、'北京'、'上海'")],
        days: Annotated[int, Field(description="旅游天数（含当天），如3表示2晚3天'")],
        hotel_level: Annotated[
            str, Field(description="住宿档次，可选值：'经济'（150元/晚）、'舒适'（300元/晚）、'豪华'（500元/晚）")] = "舒适",
        attractions: Annotated[
            Optional[List[str]], Field(description="计划游览的景点列表（可选），用于估算门票费用")] = None
) -> Dict[str, Any]:
    try:
        cost = travel_tools.estimate_travel_cost(city, days, hotel_level, attractions)
        return cost
    except Exception as e:
        return {"error": str(e)}


@mcp.tool("get_attraction_info", description="获取景点详细信息")
async def get_attraction_info(
        attraction_name: Annotated[str, Field(description="景点名称，如'兵马俑'、'华清宫'、'大雁塔'")],
        city: Annotated[str, Field(description="城市名称，如'西安'、'北京'")]
) -> Dict[str, Any]:
    try:
        info = await travel_tools.get_attraction_info(attraction_name, city)
        return info
    except Exception as e:
        return {"error": str(e)}


if __name__ == '__main__':
    mcp.run(transport="http", port=8001)

import asyncio
import json
import os
from openai import OpenAI
from fastmcp import Client
from code.Working_with_LLMs.llm_client import llm_client_with_tools

# 系统提示词
SYSTEM_PROMPT = """你是一个专业的旅游规划助手。当用户提出旅游规划需求时，请：
1. 理解需求：确认目的地、天数、预算、出行人数、特殊偏好
2. 使用工具搜索：
   - 使用 search_poi 搜索热门景点
   - 使用 get_attraction_info 获取景点详情
   - 使用 estimate_travel_cost 估算费用
   - 使用 route_planning 规划路线
3. 输出格式：
   按天输出详细行程，每天包含：
   - 上午/下午/晚上的景点安排
   - 餐饮推荐
   - 景点间交通方式和时间
   - 当日预估花费
4. 费用把控：根据用户预算合理分配费用
5. 贴心提醒：提供穿着建议、必带物品、注意事项

请确保使用工具获取最新信息，并给出具体的行程安排。"""


async def chat_with_mcp(user_input: str):
    client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com")

    # 连接到你的服务器
    mcpClient = Client("http://127.0.0.1:8001/mcp")  # 如果是HTTP服务器，这里放URL
    await mcpClient.__aenter__()
    # 列出可用工具
    mcp_tools = await mcpClient.list_tools()
    print("可用工具:", mcp_tools)
    llm_tools = []
    for tool in mcp_tools:
        llm_tools.append({
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.inputSchema,
            },
        })
    # 加载工具
    # 第一次调用，让模型自己判断是否使用工具
    messages = [{"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": user_input}]

    count = 0
    while count < 15:
        count = count + 1

        result = await llm_client_with_tools(messages, llm_tools)
        print(result)
        assistant_message = result.choices[0].message
        content = assistant_message.content or ""
        # 检查是否有工具调用
        if assistant_message.tool_calls:
            # 构建tool_calls记录
            tool_calls_data = [
                {
                    "id": tc.id,
                    "type": tc.type,
                    "function": {
                        "name": tc.function.name,
                        "arguments": tc.function.arguments
                    }
                } for tc in assistant_message.tool_calls
            ]

            # 添加到消息历史
            messages.append({
                "role": "assistant",
                "content": content,
                "tool_calls": tool_calls_data
            })

            # 执行所有工具调用
            for tool_call in assistant_message.tool_calls:
                tool_name = tool_call.function.name
                print("正在执行工具：", tool_name)
                try:
                    arguments = json.loads(tool_call.function.arguments)
                    print("工具参数：", arguments)
                except json.JSONDecodeError:
                    arguments = {}

                # 执行工具
                tool_result = await mcpClient.call_tool(tool_name, arguments)
                print("工具结果：", tool_result)
                # 添加工具结果
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": json.dumps(tool_result.structured_content, ensure_ascii=False)
                })

            continue
        else:
            count = 15
            print(content)
if __name__ == '__main__':
    asyncio.run(chat_with_mcp("西安2日游"))

---
name: douyin-summary
description: 抖音视频总结助手。当用户提供抖音（douyin.com 或 v.douyin.com）视频链接并请求总结、获取文案或了解视频内容时，使用此技能。通过调用 Coze API 工作流获取视频的转录文本或文案，然后为用户提供智能化的内容总结。
---

# 抖音视频总结助手
此技能用于获取和总结抖音视频的内容。
## 工作流程
当用户提供抖音链接时：
1. **识别抖音链接**: 检测用户输入中的 douyin.com 或 v.douyin.com 链接
2. **调用脚本获取内容**: 使用 `scripts/fetch_douyin.py` 获取视频转录/文案
   python3 ./scripts/fetch_douyin.py <url>
   3. **总结内容**: 基于获取的文本内容，提取核心观点、关键信息或有趣之处
4. **友好输出**: 以简洁易懂的方式呈现给用户

└── skill-name/                 # 技能根目录
    ├── meta.json               # [路由层] 告诉模型这个技能是干嘛的
    ├── skill.md                # [逻辑层] 包含了 System Prompt 和 SOP
    └── scripts/                # [执行层] 实际干活的 Python/Bash 脚本

# 添加官方技能库
/plugin marketplace add anthropics/skills
# 浏览可用技能
/plugin list
# 安装文档处理技能
/plugin install document-skills@anthropic-agent-skills

mkdir -p ~/.claude/skills/douyin-summary

---
name: douyin-summary
description: 抖音视频总结助手。当用户提供抖音视频链接时，自动调用此技能获取文案并总结。
---

# 抖音视频总结助手
## 工作流程
1. 识别用户输入中的douyin.com链接
2. 调用scripts/fetch_douyin.py获取视频文案
3. 提取核心观点并结构化输出

def load_skills_from_meta():
"""扫描目录，从 meta.json 加载技能元数据"""
# 示例：假设我们读取到了 simple_weather_skill
with open("./simple_weather_skill/meta.json", "r", encoding="utf-8") as f:
    meta = json.load(f)

# 构造兼容 OpenAI/DeepSeek 格式的工具定义
return [{
    "type": "function",
    "function": {
        "name": meta["name"],
        "description": meta["description"],
        "parameters": {
            "type": "object", 
            "properties": {}, 
            "required": []
        },
    },
}]

# 1. 准备工具列表
skills = load_skills_from_meta()

# 2. 发送请求给 DeepSeek，让其选择
messages = [{"role": "user", "content": user_input}]
result = client.chat.completions.create(
    model="deepseek-chat", 
    messages=messages, 
    tools=skills  # 关键点：把 Skill 描述当作 Tools 传进去
)

# 检查模型是否想调用 Skill
if result.choices[0].message.tool_calls:
    skill_name = result.choices[0].message.tool_calls[0].function.name
    print(f"模型命中技能: {skill_name}")
    # 读取对应的 Prompt 模板
    with open(f"./{skill_name}/skill.md", "r") as f:
        skill_prompt = f.read()
    # 【核心魔法】：构建一个新的对话上下文，使其立即拥有该技能的知识
    skill_messages = [
        {"role": "system", "content": skill_prompt}, # 注入技能 Prompt
        {"role": "user", "content": user_input}      # 重放用户问题
    ]
    # 此时，DeepSeek 已经变身为“天气专家”，准备好执行具体脚本了

def execute_script(script_path, args=None):
    """一个通用的脚本执行器工具"""
    cmd = ["python", script_path] + (args or [])
    res = subprocess.run(cmd, capture_output=True, text=True)
    return res.stdout

应用层: chat_with_memory.py (LLM 交互入口)
   ↓
工具层: memory_tools.py (高级封装)
   ↓
服务层: memory_service.py (业务逻辑)
   ↓
存储层: vector_store.py + models.py (数据访问)
   ↓
数据层: ChromaDB + SQLite (实际存储)

class User(Base):
    id = Column(Integer, primary_key=True)
    username = Column(String(100), unique=True, nullable=False)
    email = Column(String(255), unique=True, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow)

class Session(Base):
    id = Column(Integer, primary_key=True)
    session_id = Column(String(100), unique=True, nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))
    title = Column(String(255), nullable=True)
    context_window = Column(Integer, default=10)  # 上下文窗口大小
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)

class Memory(Base):
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey('users.id'))
    session_id = Column(Integer, ForeignKey('sessions.id'), nullable=True)
    
    memory_type = Column(String(50))  # episodic/semantic/procedural
    content = Column(Text, nullable=False)
    metadata = Column(JSON, nullable=True)
    
    vector_id = Column(String(100))  # ChromaDB 向量ID
    importance_score = Column(Float, default=0.5)  # 重要性评分
    
    access_count = Column(Integer, default=0)  # 访问次数
    last_accessed_at = Column(DateTime, nullable=True)
    is_deleted = Column(Boolean, default=False)  # 软删除

class VectorStore:
def __init__(self, persist_directory: str = "./chat_chroma.db"):    self.client = chromadb.Client(Settings(        persist_directory=persist_directory,        anonymized_telemetry=False    ))    self.collection = self.client.get_or_create_collection(        name="memory_collection"    )def add_memory(self, content: str, metadata: Dict = None) -> str:    """添加记忆到向量库"""    memory_id = str(uuid.uuid4())    metadata = metadata or {}    metadata['created_at'] = datetime.utcnow().isoformat()    self.collection.add(        ids=[memory_id],        documents=[content],        metadatas=[metadata]    )    return memory_iddef search_memories(self, query: str, n_results: int = 5,                    filter_metadata: Dict = None) -> Dict:    """语义搜索记忆"""    results = self.collection.query(        query_texts=[query],        n_results=n_results,        where=filter_metadata    )    return {        'ids': results['ids'][0],        'documents': results['documents'][0],        'metadatas': results['metadatas'][0],        'distances': results['distances'][0]    }

def add_memory(self, user_id: int, content: str, 
               memory_type: str = "episodic",               session_id: int = None,               metadata: Dict = None,               importance_score: float = 0.5) -> Memory:"""添加记忆（同时保存到数据库和向量库）"""db = self._get_db_session()try:    # 1. 创建数据库记录    memory = Memory(        user_id=user_id,        session_id=session_id,        memory_type=memory_type,        content=content,        metadata=metadata or {},        importance_score=importance_score    )    db.add(memory)    db.commit()    db.refresh(memory)    # 2. 添加到向量库    vector_metadata = {        'user_id': user_id,        'memory_type': memory_type,        'memory_db_id': memory.id    }    vector_id = self.vector_store.add_memory(        content=content,        metadata=vector_metadata    )    # 3. 更新向量ID    memory.vector_id = vector_id    db.commit()    return memoryfinally:    db.close()

def search_memories(self, query: str, user_id: int = None,
                   memory_type: str = None,                   n_results: int = 5) -> List[Dict]:"""搜索记忆"""# 1. 向量搜索filter_metadata = {}if user_id:    filter_metadata['user_id'] = user_idif memory_type:    filter_metadata['memory_type'] = memory_typeresults = self.vector_store.search_memories(    query=query,    n_results=n_results * 2,    filter_metadata=filter_metadata if filter_metadata else None)# 2. 整合数据库信息memories = []db = self._get_db_session()try:    for i, vector_id in enumerate(results['ids']):        metadata = results['metadatas'][i]        memory_db_id = metadata.get('memory_db_id')        if memory_db_id:            memory = db.query(Memory).filter(                Memory.id == memory_db_id,                Memory.is_deleted == False            ).first()            if memory:                # 更新访问统计                memory.access_count += 1                memory.last_accessed_at = datetime.utcnow()                memories.append({                    'id': memory.id,                    'content': memory.content,                    'memory_type': memory.memory_type,                    'importance_score': memory.importance_score,                    'distance': results['distances'][i],                    'created_at': memory.created_at.isoformat()                })        if len(memories) >= n_results:            break    db.commit()    return memoriesfinally:    db.close()

def retrieve_for_query(self, user_id: int, query: str,
                      session_id: int = None,                      include_context_window: bool = True,                      max_results: int = 5) -> Dict:"""RAG 检索：整合短期和长期记忆"""result = {    'context_window': [],    'relevant_memories': []}# 1. 获取上下文窗口（短期记忆）if include_context_window and session_id:    context_memories = self.service.get_context_window(session_id)    result['context_window'] = [        {'content': m.content, 'created_at': m.created_at.isoformat()}        for m in context_memories    ]# 2. 语义搜索（长期记忆）relevant = self.service.search_memories(    query=query,    user_id=user_id,    n_results=max_results)result['relevant_memories'] = relevantreturn result

class ChatWithMemory:
    def chat(self, user_input: str, use_rag: bool = True) -> str:    """与 LLM 聊天（带记忆）"""    messages = [{"role": "system", "content": self.system_prompt}]    if use_rag:        # RAG 检索        rag_result = self.memory.retrieve_for_query(            user_id=self.user_id,            query=user_input,            session_id=self.session_id,            include_context_window=True,            max_results=3        )        # 添加相关长期记忆        if rag_result['relevant_memories']:            memory_context = "相关历史信息：\n"            for mem in rag_result['relevant_memories'][:3]:                memory_context += f"- {mem['content']}\n"            messages.append({"role": "system", "content": memory_context})        # 添加上下文窗口        for ctx in rag_result['context_window']:            # 解析对话内容            if"User:"in ctx['content'] and "Assistant:"in ctx['content']:                parts = ctx['content'].split("\nAssistant:")                user_part = parts[0].replace("User:", "").strip()                assistant_part = parts[1].strip()                messages.append({"role": "user", "content": user_part})                messages.append({"role": "assistant", "content": assistant_part})    # 当前输入    messages.append({"role": "user", "content": user_input})    # 调用 LLM    response = call_llm(messages=messages)    assistant_message = response.choices[0].message.content    # 保存对话记忆    self.memory.add_conversation_memory(        user_id=self.user_id,        session_id=self.session_id,        user_message=user_input,        assistant_message=assistant_message    )    return assistant_message

# 基础使用
from memory_tools import MemoryTools# 初始化memory = MemoryTools()# 创建用户和会话user_id = memory.create_user("alice")session_id = memory.create_session(user_id, "session-001")# 添加对话记忆memory.add_conversation_memory(    user_id=user_id,    session_id=session_id,    user_message="我喜欢Python编程",    assistant_message="Python是一门很棒的语言！")# 搜索记忆results = memory.search_memories(    user_id=user_id,    query="编程语言",    n_results=5)

# 运行交互式聊天
python chat_with_memory.py

# 示例对话
你: 我叫Alice，喜欢编程
助手: 你好Alice！很高兴认识你。编程是一个很有趣的领域...

你: 我之前说过我叫什么名字？
助手: 你之前说你叫Alice。

@dataclass
class PlanStep:
    id: str
    tool: str
    args: Dict[str, Any]
    why: str = ""
    status: Literal["pending", "running", "success", "failed", "skipped"] = "pending"
    output: Any = None
    error: str = ""

@dataclass
class Plan:
    goal: str
    steps: List[PlanStep] = field(default_factory=list)

system_prompt = f"""
你是一个任务规划器（Planner）。你必须只输出 JSON，不要输出任何解释文字。
目标：把用户的需求拆成若干步，每一步要调用一个工具（tool），并提供参数（args）。

关键规则——数据依赖：
如果某一步的 args 需要依赖前面步骤的输出结果，请务必使用模板语法 `{{{{steps.前面的步骤ID.output.字段名}}}}`。
不要自己捏造数据，而是引用前序步骤的输出。

例如：
1. step1 (id="s1") 调用 geocode 返回 `{{"location": "116.40,39.90"}}`
2. step2 (id="s2") 需要用到这个坐标，args 应该写： `{{"location": "{{{{steps.s1.output.location}}}}"}}`

你必须输出的 JSON 结构：
{{
  "goal": "...",
  "steps": [
    {{
      "id": "step1",
      "tool": "tool_name",
      "args": {{ ... }}
    }}
  ]
}}
"""

async def _run_step(self, step: PlanStep, mcp_client: Client, context: Dict[str, Any]) -> None:
step.status = "running"try:    # 解析参数中的 {{...}} 模板，替换为上下文中的真实数据    resolved_args = resolve_templates(step.args, context)    # 调用 MCP 工具    result = await mcp_client.call_tool(step.tool, resolved_args)    # 记录输出结果    step.output = getattr(result, "structured_content", None) or result.content    step.status = "success"    # 写入上下文，供后续步骤引用    context["steps"][step.id] = {        "status": step.status,         "output": step.output    }except Exception as e:    step.status = "failed"    step.error = str(e)

uv run python -m code.MCP.mcp_server

uv run python -m  code.Planning.demo_planning_cli.py

async function main({ params }: Args): Promise<Output> {    // 构建输出对象   const ret = {    fields: [        {            fields: {                "小红书文案": params.content,                "小红书标题": params.title,                "小红书图文地址": params.imageUrl                           },            record_id: params.record_id        }    ]};    return ret;}