分享我的Colab文件：
https://colab.research.google.com/drive/1XIi1E_L0dod2CoJIz9QVEYCjQKTKAuhT?usp=sharing
把它另存到你的本地Colab笔记本（详细步骤见2025-08-11文章）
点击左侧上传你的数据集文件.json格式。
一定要修改第二个单元格代码filepath = ” “（你上传的数据集的名称）
免费 T4 申请教程：https://zhuanlan.zhihu.com/p/642542618
点击Run all一键运行所有代码格
点击Running on public URL的链接: https://b1b7e1e391675e8e31.gradio.live
GitHub提供文档有详细介绍：https://github.com/hiyouga/LLaMA-Factory/blob/main/README_zh.md
提供的视频非常详细，一看就会。
分享我的Colab文件：
https://colab.research.google.com/drive/11jEpOECqYKEWh_75Bc3hSw2CmWqE4oF3?usp=sharing
点开链接会进入Colab界面，在这所做的修改不会保存，需要将它复制保存到Drive中。
上传文件
在这个单元格data_files=填入你的数据集名称；如果是别的数据集格式需要修改代码
修改微调细节：
最后一个单元格是测试微调效果，在原项目上是没有的，如果不需要可以直接删除：
如果使用T4 GPU会有概率超出可使用显存，导致失败；推荐选择L4 GPU性价比最高；A100 GPU速度更快，收费更高，具体看你需求。
点击Restart session and run all即可一键运行：
import requests
import time
import os
class PDFProcessor:
def __init__(self, token):
self.token = token
self.headers = {
'Authorization': f'Bearer {token}'
}
def upload_file(self, file_path):
"""上传本地文件到临时存储"""
print(f"📤 正在上传文件: {os.path.basename(file_path)}")
# 使用 tmpfiles.org
try:
with open(file_path, 'rb') as f:
response = requests.post(
'https://tmpfiles.org/api/v1/upload',
files={'file': f}
)
if response.status_code == 200:
result = response.json()
# 获取直接下载链接
url = result['data']['url']
direct_url = url.replace('tmpfiles.org/', 'tmpfiles.org/dl/')
print(f"✅ 上传成功: {direct_url}")
return direct_url
except Exception as e:
print(f"❌ 上传失败: {e}")
return None
def process_pdf(self, file_path):
"""处理本地PDF文件"""
# 1. 上传文件
pdf_url = self.upload_file(file_path)
if not pdf_url:
return None
# 2. 创建MinerU任务
print("📄 创建解析任务...")
task_url = 'https://mineru.net/api/v4/extract/task'
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {self.token}'
}
data = {
'url': pdf_url,
'is_ocr': True,
'enable_formula': True,
'enable_table': True,
'language': 'auto' # 中文文档
}
response = requests.post(task_url, headers=headers, json=data)
result = response.json()
if result['code'] != 0:
print(f"❌ 创建任务失败: {result['msg']}")
return None
task_id = result['data']['task_id']
print(f"✅ 任务ID: {task_id}")
# 3. 等待处理完成
print("⏳ 等待处理...")
while True:
time.sleep(5)
status_url = f'https://mineru.net/api/v4/extract/task/{task_id}'
status_response = requests.get(status_url, headers=headers)
status_data = status_response.json()
state = status_data['data']['state']
if state == 'done':
zip_url = status_data['data']['full_zip_url']
print(f"✅ 处理完成！")
print(f"📦 下载地址: {zip_url}")
# 下载结果
self.download_result(zip_url, task_id)
return status_data
elif state == 'failed':
print(f"❌ 处理失败: {status_data['data']['err_msg']}")
return None
elif state == 'running':
progress = status_data['data'].get('extract_progress', {})
extracted = progress.get('extracted_pages', 0)
total = progress.get('total_pages', 0)
print(f"⏳ 正在处理... {extracted}/{total} 页")
else:
print(f"📊 状态: {state}")
def download_result(self, zip_url, task_id):
"""下载结果文件"""
save_path = f"mineru_result_{task_id}.zip"
try:
response = requests.get(zip_url, stream=True)
with open(save_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
print(f"✅ 结果已保存到: {save_path}")
except Exception as e:
print(f"❌ 下载失败: {e}")
# 使用您的Token和文件
TOKEN = "eyJ0eXBlIjoiSldUIiwiYWxnIjoiSFM1MTIifQ.eyJqdGkiOiIyOTkwMzQ1NiIsInJvbCI6IlJPTEVfUkVHSVNURVIiLCJpc3MiOiJPcGVuWExhYiIsImlhdCI6MTc1MTM3NDYyOCwiY2xpZW50SWQiOiJsa3pkeDU3bnZ5MjJqa3BxOXgydyIsInBob25lIjoiMTg5NTIxNTUyNTAiLCJvcGVuSWQiOm51bGwsInV1aWQiOiI1YTA0YmUxMC1kMTJkLTQ3NzktYjYyYi1mM2U4NTRmZWI0YTQiLCJlbWFpbCI6IiIsImV4cCI6MTc1MjU4NDIyOH0.kKzBBW2Jp2sVh3HXRVRlz-Df8WMHcDB7PM1pZbrmn3_QEt39bw3OrNAf8OkrmgY9Kign1fpTxPRzxopOenqO7Q"
# 处理您的PDF文件
processor = PDFProcessor(TOKEN)
local_pdf = "/home/wang/1_time/00bjfei/宁德时代宁德时代新能源科技股份有限公司2024年半年度报告174页(1).pdf"
# 检查文件是否存在
if os.path.exists(local_pdf):
processor.process_pdf(local_pdf)
else:
print(f"❌ 文件不存在: {local_pdf}")
微信：adoresever
我的开源项目：
https://github.com/adoresever/DataGraphX_Learn
https://github.com/adoresever/Pretuning
代码
import os
import httpx
from fastapi import FastAPI, HTTPException, Query
from fastapi_mcp import FastApiMCP
from pydantic import BaseModel, Field
from dotenv import load_dotenv
import logging
import json
# --- 配置与设置 ---
load_dotenv()
API_KEY = os.getenv("TAVILY_API_KEY")
if not API_KEY:
raise ValueError("在环境变量或 .env 文件中未找到 TAVILY_API_KEY")
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
TAVILY_API_URL = "https://api.tavily.com/search"
# --- Pydantic 模型 (目前保持不变) ---
class NewsResponse(BaseModel):
"""描述 Tavily 搜索工具返回的新闻摘要。"""
query: str = Field(..., description="用于搜索新闻的原始查询。")
news_summary: str = Field(..., description="由 Tavily AI 基于查询找到的近期新闻的更详细摘要，合并了多个来源。")
# --- FastAPI 应用 ---
app = FastAPI(
title="通过 MCP 使用 Tavily 搜索详细新闻", # 更新标题
description="使用 Tavily AI 搜索查找并合并近期新闻细节，基于查询，并通过 MCP 暴露。",
version="1.0.0",
)
# --- 新闻获取逻辑 (修改以获取更多细节) ---
async def fetch_detailed_news_from_tavily(news_query: str) -> str:
"""使用 Tavily AI 搜索新闻并合并结果以获取更多细节。"""
payload = json.dumps({
"api_key": API_KEY,
"query": news_query,
"search_depth": "advanced", # 使用高级搜索以获取可能更多的细节
"include_answer": False, # 如果你只想依赖结果内容，设为 False
"max_results": 7 # 增加考虑的结果数量
# "include_raw_content": True # 可选：可能获取更多原始内容，但需要仔细解析
})
headers = {'Content-Type': 'application/json'}
async with httpx.AsyncClient(timeout=30.0) as client: # 为高级搜索增加超时时间
response_data = None
try:
logger.info(f"正在向 Tavily AI 请求关于查询 '{news_query}' 的详细新闻摘要...")
response = await client.post(TAVILY_API_URL, headers=headers, content=payload)
response.raise_for_status()
response_data = response.json()
logger.debug(f"Tavily 对于查询 '{news_query}' 的原始响应: {response_data}")
# --- 重点在于合并结果 ---
combined_summary = ""
if response_data.get("results"):
summaries = []
for i, res in enumerate(response_data["results"]):
title = res.get('title', f"来源 {i+1}")
content = res.get('content', '').strip() # 获取内容并去除首尾空格
# url = res.get('url', '#') # 如果需要，可以包含 URL
if content: # 仅当有内容时才包含
#清晰地格式化每个结果
summaries.append(f"--- 结果 {i+1}: {title} ---\n{content}")
if summaries:
# 使用两个换行符连接所有摘要
combined_summary = "\n\n".join(summaries)
# 如果需要，可以限制总长度
# max_len = 2000 # 示例长度限制
# if len(combined_summary) > max_len:
# combined_summary = combined_summary[:max_len] + "..."
logger.info(f"已合并来自 Tavily 对于查询 '{news_query}' 的 {len(summaries)} 条结果。")
return combined_summary
else:
logger.warning(f"Tavily 对于查询 '{news_query}' 的结果没有可用的内容进行合并。")
# 如果根本没有找到可用的结果
logger.warning(f"Tavily 对于查询 '{news_query}' 没有返回可用的结果。")
raise HTTPException(status_code=404, detail=f"无法使用 Tavily AI 找到与查询 '{news_query}' 匹配的近期新闻细节。")
# --- 异常处理 (与天气示例大部分相同) ---
except httpx.HTTPStatusError as exc:
error_detail = "与 Tavily AI 服务通信时出错。"
try: error_body = exc.response.json(); error_detail = error_body.get("error", error_detail)
except ValueError: error_detail = exc.response.text or error_detail
logger.error(f"从 Tavily 获取查询 '{news_query}' 的新闻时发生 HTTP 错误: {exc.response.status_code} - 详情: {error_detail}")
if exc.response.status_code in [401, 403]: raise HTTPException(status_code=exc.response.status_code, detail=f"Tavily API 密钥错误: {error_detail}") from exc
raise HTTPException(status_code=exc.response.status_code, detail=error_detail) from exc
except httpx.RequestError as exc:
logger.error(f"从 Tavily 获取查询 '{news_query}' 的新闻时发生网络错误: {exc}")
raise HTTPException(status_code=503, detail="无法连接到 Tavily AI 服务。") from exc
except ValueError as exc:
logger.error(f"解码来自 Tavily 对查询 '{news_query}' 的 JSON 响应时出错: {exc}")
raise HTTPException(status_code=500, detail="从 Tavily AI 服务收到无效的数据格式。") from exc
except Exception as exc:
logger.exception(f"获取查询 '{news_query}' 的 Tavily 新闻时发生意外错误: {exc}")
raise HTTPException(status_code=500, detail="获取新闻摘要时发生意外的内部错误。") from exc
# --- FastAPI 端点 (调用新的获取函数) ---
@app.get(
"/news/search",
response_model=NewsResponse,
operation_id="search_detailed_recent_news_tavily", # 更新 operation_id
summary="搜索详细近期新闻 (通过 Tavily AI)", # 更新摘要
tags=["News Tools", "Tavily AI"]
)
async def search_news(
q: str = Query(..., min_length=3, description="用于搜索近期新闻的主题或查询。")
):
"""
使用 Tavily AI 的高级搜索查找并合并来自近期新闻的细节，基于用户查询。
返回一个合并了多个来源的更长的文本摘要。
"""
logger.info(f"收到关于查询的详细新闻搜索请求: {q}")
# 调用修改后的获取函数
news_summary_text = await fetch_detailed_news_from_tavily(q)
response_data = NewsResponse(
query=q,
news_summary=news_summary_text
)
logger.info(f"已通过 Tavily 成功检索到查询 '{q}' 的详细新闻摘要。")
return response_data
# --- FastAPI-MCP 集成 ---
logger.info("正在初始化 FastAPI-MCP...")
mcp = FastApiMCP(app)
mcp.mount()
logger.info("FastAPI-MCP 服务器已挂载到 /mcp")
# --- 运行服务器 ---
if __name__ == "__main__":
import uvicorn
logger.info("正在启动用于 Tavily 详细新闻的 Uvicorn 服务器...")
uvicorn.run(
"news_mcp:app", # 确保文件名匹配
host="127.0.0.1",
port=8000, # 如果天气服务也在运行，可以更改端口，例如 8001
reload=True
)测试的环境：Goodle AI Studio
测试的模型：选择Gemini 2.5 Pro Experimental 03-25
测试形式：通过向模型发送指令（prompt）
将此模型输出的三个文件放入Vs Code，注意三个文件要在同一个文件下。
完成后我们打开HTML文件可以看到
操作结果：操作通过。用户可以自由添加支出收入金额，添加相应描述，同时系统会统计总支出，总收入和余额。
康威生命游戏是什么：
操作过程：向模型发出指令，但这次我们选择在python脚本中运行。
操作结果：测试通过，生成康威生命游戏
操作过程：向模型发出指令，并将SVG 代码输入在线工具中来查看。
操作结果：测试通过，利用svg语法生成蝴蝶。
测试结果：通过。
测试结果：通过。
模型进行了这几个步骤：分析初始阶段 (8:00 AM – 9:00 AM)——分析第二阶段 (9:00 AM – 10:00 AM)——分析第三阶段 (10:00 AM 之后)——计算相遇时间——计算相遇地点 (距离A市)
测试结果：通过
测试结果：通过
测试结果：
操作过程：使用OpenRouter同时选择这两个模型并提问测试八的问题
操作结果：内容一致，Gemini2.5速度更快。
Archon是一个创新的AI元代理系统，其核心理念是通过抓取各种框架的文档，来构建能够使用这些框架的AI代理。作为世界上第一个”Agenteer”（代理工程师），它能够自主构建、改进和优化其他AI代理。
Archon的核心是通过抓取各种框架的文档，用来构建能够使用这些框架的AI代理。
Supabase是一个开源的后端即服务(BaaS)平台，旨在简化应用程序开发过程。在Archon中，它主要用于存储和检索向量化的框架文档，为代理提供知识基础。
git clone https://github.com/coleam00/archon.git
cd archon
python run_docker.py部署完成后访问 http://localhost:8501
git clone https://github.com/coleam00/archon.git
cd archon
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
pip install -r requirements.txt
streamlit run streamlit_ui.pyAI IDE是集成了人工智能功能的开发环境，比如Windsurf、Cursor和Cline，可以提供代码补全、生成代码建议等智能功能。
完成安装后，需要在Streamlit UI配置以下环境：
MCP是”Model Context Protocol”(模型上下文协议)，这是一种允许AI IDE与外部AI服务通信的标准化协议。通过MCP配置，可以：
完成所有配置后，可以：
如果配置了MCP，还可以直接在AI IDE中使用Archon的功能，让Archon为你生成代理代码，并将其集成到你的项目中。
Archon通过创新的文档抓取和向量化方法，展示了元学习在AI系统构建中的潜力。它不仅是一个实用的开发工具，也是一个展示智能代理系统演化的教育框架。随着其不断迭代完善，Archon有望成为AI代理开发生态系统的重要组成部分。
微信：adoresever
qa的差异部分
# 修正 pyproject.toml 文件（如果需要） # 删除 [project] 部分，确保正确的配置
[tool.poetry]
name = "my_agent"
version = "0.1.0"
description = "Starter"
authors = ["Ariel Weinberger <weinberger.ariel@gmail.com>"]
license = "MIT"
packages = [
{ include = "my_agent" }
]
[tool.poetry.dependencies]
python = "^3.12"
langchain-openai = "^0.2.1"
langchain-anthropic = "^0.2.1"
langchain = "^0.3.1"
openai = "^1.51.0"
langchain-community = "^0.3.1"
copilotkit = "0.1.33"
uvicorn = "^0.31.0"
python-dotenv = "^1.0.1"
langchain-core = "^0.3.25"
langgraph-cli = {extras = ["inmem"], version = "^0.1.64"}
[tool.poetry.scripts]
demo = "my_agent.demo:main"
[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"然后运行：
poetry lock
poetry install
其他一样！
coagents-research-canvas安装过程
源码克隆
git clone https://github.com/CopilotKit/CopilotKit.git
cd 该目录
虚拟环境
conda create -n coagents-re python=3.12
conda activate coagents-re
cd agent
poetry install
.env添加OPENAI_API_KEY= TAVILY_API_KEY=
api获取：https://tavily.com/ https://platform.openai.com/api-keys
poetry run demo运行
cd ui
pnpm add @copilotkit/react-ui@1.4.8-coagents-v0-3.1 @copilotkit/react-core@1.4.8-coagents-v0-3.1
pnpm i
.env
pnpm run dev运行
若端口报错视频里有解决方法！
About me:
