feat: init 0

This commit is contained in:
李如威 2025-12-08 17:13:19 +08:00
commit 2050622dea
13 changed files with 396 additions and 0 deletions

190
.gitignore vendored Normal file
View File

@ -0,0 +1,190 @@
# ============================================
# Python项目 .gitignore 文件
# 生成日期2025-12-05
# ============================================
# 1. Python字节码和编译文件[1](@ref)[3](@ref)[6](@ref)
__pycache__/
*.py[cod]
*$py.class
*.pyc
*.pyo
*.pyd
# C扩展模块
*.so
# 2. 分发和打包文件[1](@ref)[3](@ref)[6](@ref)
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller生成文件[1](@ref)[3](@ref)
*.manifest
*.spec
# 安装日志
pip-log.txt
pip-delete-this-directory.txt
# 3. 单元测试和覆盖率报告[3](@ref)[6](@ref)
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# 4. 翻译文件
*.mo
*.pot
# 5. 框架特定文件
# Django[3](@ref)[6](@ref)
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
*/migrations/*.py
!*/migrations/__init__.py
# Flask[1](@ref)[3](@ref)
instance/
.webassets-cache
# Scrapy
.scrapy
# Sphinx文档
docs/_build/
# PyBuilder
.pybuilder/
target/
# 6. Jupyter Notebook和IPython[3](@ref)[6](@ref)
.ipynb_checkpoints
profile_default/
ipython_config.py
# 7. 环境管理文件
# pyenv版本文件[3](@ref)[6](@ref)
# .python-version
# pipenv根据项目需要决定是否忽略[3](@ref)[6](@ref)
#Pipfile.lock
# poetry根据项目需要决定是否忽略[3](@ref)[6](@ref)
#poetry.lock
# pdm[6](@ref)
#pdm.lock
.pdm.toml
# PEP 582包目录[3](@ref)[6](@ref)
__pypackages__/
# Celery相关文件[3](@ref)[6](@ref)
celerybeat-schedule
celerybeat.pid
# SageMath解析文件
*.sage.py
# 8. 虚拟环境和环境变量文件[1](@ref)[3](@ref)[5](@ref)
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# 9. IDE和编辑器配置文件[4](@ref)[5](@ref)
# PyCharm/IntelliJ IDEA
.idea/
# VS Code
.vscode/
*.vscode/settings.json
*.code-workspace
# Spyder项目设置
.spyderproject
.spyproject
# Rope项目设置
.ropeproject
# 10. 文档生成
# mkdocs文档
/site
# 11. 类型检查工具缓存[3](@ref)[6](@ref)
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre类型检查器
.pyre/
# pytype静态类型分析器
.pytype/
# Cython调试符号
cython_debug/
# 12. 操作系统特定文件macOS[2](@ref)[4](@ref)
.DS_Store
.AppleDouble
.LSOverride
Icon
._*
.DocumentRevisions-V100
.fseventsd
.Spotlight-V100
.TemporaryItems
.Trashes
.VolumeIcon.icns
.com.apple.timemachine.donotpresent
.AppleDB
.AppleDesktop
Network Trash Folder
Temporary Items
.apdisk
# 13. 项目特定自定义忽略(根据实际情况添加)
# 日志文件
*.log
.*.log
# 本地配置文件
local_config.py
# 数据目录(如果需要)
# datas/
# docker_image_build_tmp/
# logs/
# results/

0
README.md Normal file
View File

9
requirements.txt Normal file
View File

@ -0,0 +1,9 @@
fastapi
uvicorn[standard]
httpx[socks]
qdrant-client[httpx]
python-dotenv
pytest-asyncio
pymupdf
python-docx
scikit-learn

0
src/pipeline/__init__.py Normal file
View File

View File

0
src/pipeline/config.py Normal file
View File

View File

View File

@ -0,0 +1,197 @@
import asyncio, warnings, copy, time
class BaseNode:
def __init__(self):
self.params, self.successors = {}, {}
def set_params(self, params):
self.params = params
def next(self, node, action="default"):
if action in self.successors:
warnings.warn(f"Overwriting successor for action '{action}'")
self.successors[action] = node
return node
def prep(self, shared):
pass
def exec(self, prep_res):
pass
def post(self, shared, prep_res, exec_res):
pass
def _exec(self, prep_res):
return self.exec(prep_res)
def _run(self, shared):
p = self.prep(shared)
e = self._exec(p)
return self.post(shared, p, e)
def run(self, shared):
if self.successors:
warnings.warn("Node won't run successors. Use Flow.")
return self._run(shared)
def __rshift__(self, other):
return self.next(other)
def __sub__(self, action):
if isinstance(action, str):
return _ConditionalTransition(self, action)
raise TypeError("Action must be a string")
class _ConditionalTransition:
def __init__(self, src, action):
self.src, self.action = src, action
def __rshift__(self, tgt):
return self.src.next(tgt, self.action)
class Node(BaseNode):
def __init__(self, max_retries=1, wait=0):
super().__init__()
self.max_retries, self.wait = max_retries, wait
def exec_fallback(self, prep_res, exc):
raise exc
def _exec(self, prep_res):
for self.cur_retry in range(self.max_retries):
try:
return self.exec(prep_res)
except Exception as e:
if self.cur_retry == self.max_retries - 1:
return self.exec_fallback(prep_res, e)
if self.wait > 0:
time.sleep(self.wait)
class BatchNode(Node):
def _exec(self, items):
return [super(BatchNode, self)._exec(i) for i in (items or [])]
class Flow(BaseNode):
def __init__(self, start=None):
super().__init__()
self.start_node = start
def start(self, start):
self.start_node = start
return start
def get_next_node(self, curr, action):
nxt = curr.successors.get(action or "default")
if not nxt and curr.successors:
warnings.warn(f"Flow ends: '{action}' not found in {list(curr.successors)}")
return nxt
def _orch(self, shared, params=None):
curr, p, last_action = copy.copy(self.start_node), (params or {**self.params}), None
while curr:
curr.set_params(p)
last_action = curr._run(shared)
curr = copy.copy(self.get_next_node(curr, last_action))
return last_action
def _run(self, shared):
p = self.prep(shared)
o = self._orch(shared)
return self.post(shared, p, o)
def post(self, shared, prep_res, exec_res):
return exec_res
class BatchFlow(Flow):
def _run(self, shared):
pr = self.prep(shared) or []
for bp in pr:
self._orch(shared, {**self.params, **bp})
return self.post(shared, pr, None)
class AsyncNode(Node):
async def prep_async(self, shared):
pass
async def exec_async(self, prep_res):
pass
async def exec_fallback_async(self, prep_res, exc):
raise exc
async def post_async(self, shared, prep_res, exec_res):
pass
async def _exec(self, prep_res):
for self.cur_retry in range(self.max_retries):
try:
return await self.exec_async(prep_res)
except Exception as e:
if self.cur_retry == self.max_retries - 1:
return await self.exec_fallback_async(prep_res, e)
if self.wait > 0:
await asyncio.sleep(self.wait)
async def run_async(self, shared):
if self.successors:
warnings.warn("Node won't run successors. Use AsyncFlow.")
return await self._run_async(shared)
async def _run_async(self, shared):
p = await self.prep_async(shared)
e = await self._exec(p)
return await self.post_async(shared, p, e)
def _run(self, shared):
raise RuntimeError("Use run_async.")
class AsyncBatchNode(AsyncNode, BatchNode):
async def _exec(self, items):
return [await super(AsyncBatchNode, self)._exec(i) for i in items]
class AsyncParallelBatchNode(AsyncNode, BatchNode):
async def _exec(self, items):
return await asyncio.gather(*(super(AsyncParallelBatchNode, self)._exec(i) for i in items))
class AsyncFlow(Flow, AsyncNode):
async def _orch_async(self, shared, params=None):
curr, p, last_action = copy.copy(self.start_node), (params or {**self.params}), None
while curr:
curr.set_params(p)
last_action = await curr._run_async(shared) if isinstance(curr, AsyncNode) else curr._run(shared)
curr = copy.copy(self.get_next_node(curr, last_action))
return last_action
async def _run_async(self, shared):
p = await self.prep_async(shared)
o = await self._orch_async(shared)
return await self.post_async(shared, p, o)
async def post_async(self, shared, prep_res, exec_res):
return exec_res
class AsyncBatchFlow(AsyncFlow, BatchFlow):
async def _run_async(self, shared):
pr = await self.prep_async(shared) or []
for bp in pr:
await self._orch_async(shared, {**self.params, **bp})
return await self.post_async(shared, pr, None)
class AsyncParallelBatchFlow(AsyncFlow, BatchFlow):
async def _run_async(self, shared):
pr = await self.prep_async(shared) or []
await asyncio.gather(*(self._orch_async(shared, {**self.params, **bp}) for bp in pr))
return await self.post_async(shared, pr, None)

View File

0
src/pipeline/main.py Normal file
View File

View File

View File

0
src/tests/test.py Normal file
View File