ThinkAi - 企业级 AI 大模型集成框架

框架概述

ThinkAi 是一个基于 Python 异步的企业级 AI 大模型集成框架。核心特性：轻量零依赖（不依赖任何 Web 框架，完全脱离 FastAPI）、多模型支持、统一接口、同步/异步双模式。

🏆 核心优势

一次配置，多模型自由切换。支持 12 种主流大模型，包括本地 Ollama 和云端 OpenAI、通义千问、DeepSeek、Claude、Gemini 等。

安装

      # 基础安装（轻量核心，无任何 Web 框架依赖）
pip install thinkai-framework

# Web 开发支持
pip install thinkai-framework[web]

# 安装特定 Provider
pip install thinkai-framework[ollama]
pip install thinkai-framework[openai]
pip install thinkai-framework[qwen]

# 安装全部依赖
pip install thinkai-framework[all]
    

快速开始

1. 最简使用（3行代码）

      from thinkai import ThinkAI

ai = ThinkAI(provider="ollama", model="llama3")
response = await ai.chat("你好")
print(response.content)
    

2. 同步模式（无需 async/await）

      from thinkai import SyncThinkAI

ai = SyncThinkAI(provider="ollama", model="llama3")
response = ai.chat("你好")
print(response.content)
    

3. CLI 命令行

      thinkai version
thinkai chat "你好" --provider ollama
thinkai providers
    

4. FastAPI 集成

      from fastapi import FastAPI
from thinkai import ThinkAI

app = FastAPI()
ai = ThinkAI(provider="ollama", model="llama3")

@app.post("/chat")
async def chat(message: str):
    response = await ai.chat(message)
    return {"content": response.content}
    

5. 多模型配置与切换

      ai = ThinkAI(provider="ollama", model="llama3")

# 注册多个模型
ai.register_model("qwen", provider="qwen", model="qwen-turbo")
ai.register_model("deepseek", provider="deepseek", model="deepseek-chat")
ai.register_model("gpt4", provider="openai", model="gpt-4")

# 自由切换
response1 = await ai.chat("你好", model="llama3")
response2 = await ai.chat("你好", model="qwen")
    

6. 流式响应

      async for chunk in ai.chat_stream("讲一个故事"):
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)
    

支持的 AI 模型

Provider	默认模型	API 地址	类型
Ollama	llama3	http://localhost:11434	本地免费
OpenAI	gpt-3.5-turbo	api.openai.com	云端
DeepSeek	deepseek-chat	api.deepseek.com	云端
通义千问	qwen-turbo	dashscope.aliyuncs.com	云端
智谱 GLM	glm-4	open.bigmodel.cn	云端
百度文心	ernie-bot	qianfan.baidubce.com	云端
腾讯混元	hunyuan	api.hunyuan.cloud.tencent.com	云端
豆包 ARK	doubao-pro	ark.cn-beijing.volces.com	云端
Kimi	moonshot-v1-8k	api.moonshot.cn	云端
MiniMax	abab6-chat	api.minimax.chat	云端
Claude	claude-3-sonnet	api.anthropic.com	云端
Gemini	gemini-pro	generativelanguage.googleapis.com	云端

RAG 检索增强生成

      from thinkai import ThinkAI
from thinkai.rag import RAGPipeline

ai = ThinkAI(provider="ollama")

# 创建 RAG Pipeline（自动使用 Ollama Embedding）
rag = RAGPipeline(
    documents=["./docs", "./knowledge"],
    ai_client=ai,
)

# 索引文档
await rag.index()

# 查询
answer = await rag.query("ThinkAi 框架支持哪些 AI 模型?")
print(answer)
    

Agent 系统

Function Calling Agent

      from thinkai import ThinkAI
from thinkai.agent import FunctionCallingAgent
from thinkai.skill import MathSkill, CodeSkill

ai = ThinkAI(provider="openai", api_key="your-key")

agent = FunctionCallingAgent(
    name="math-coder",
    tools=MathSkill().get_tools() + CodeSkill().get_tools(),
    ai_client=ai,
    verbose=True,
)

result = await agent.run("计算123*456, 然后用Python写一个斐波那契数列函数")
    

多 Agent 协作

      from thinkai.agent.orchestrator import MultiAgentOrchestrator, AgentRole

orchestrator = MultiAgentOrchestrator(ai_client=ai)

# 添加 Agent
orchestrator.add_agent(researcher_agent, role=AgentRole.SPECIALIST)
orchestrator.add_agent(analyzer_agent, role=AgentRole.WORKER)

# 顺序执行
result = await orchestrator.run(
    task=["搜索AI最新进展", "分析搜索结果"],
    mode="sequential"
)

# 并行执行
result = await orchestrator.run(
    task=["搜索AI", "搜索量子计算", "搜索区块链"],
    mode="parallel"
)
    

记忆系统

      from thinkai.memory import MemoryManager, FileMemoryStore

memory = MemoryManager(
    store=FileMemoryStore(base_path="./my_memory"),
    max_memories=100,
    enable_forgetting=True,     # 自动遗忘
)

await memory.remember("用户喜欢Python编程", memory_type="preference", importance=0.8)
await memory.remember("项目使用FastAPI框架", memory_type="fact", importance=0.9)

results = await memory.recall("编程语言")
context = await memory.build_context_prompt("Python开发", max_tokens=500)
    

项目结构

      thinkai/
├── core/            # 核心模块（客户端、配置、数据模型）
├── providers/       # Provider 实现（12 种模型）
├── session/         # 会话管理
├── prompt/          # Prompt 模板（16 种）
├── middleware/      # 中间件（日志、重试、限流）
├── rag/             # RAG 模块
├── agent/           # Agent 模块（ReAct、Function Calling、多Agent协作）
├── skill/           # Skill 系统（9 大内置 Skill）
├── memory/          # 记忆系统
├── mcp/             # MCP 协议支持
├── cache/           # 缓存层
├── structured/      # 结构化输出
├── tracing/         # 链路追踪
├── plugin/          # 插件系统
├── sync.py          # 同步 API
├── cli.py           # CLI 工具
└── tests/           # 223 个测试用例
    

配置方式

方式一：代码配置

      ai = ThinkAI(
    provider="ollama",
    model="llama3",
    temperature=0.7,
    max_tokens=2048,
    timeout=60,
)
    

方式二：环境变量

      export THINKAI_DEFAULT_PROVIDER=ollama
export THINKAI_DEFAULT_MODEL=llama3
export OPENAI_API_KEY=your_key_here
    

方式三：YAML 配置文件

      default_provider: "ollama"
default_model: "llama3"
providers:
  openai:
    api_key: "${OPENAI_API_KEY}"
    api_base: "https://api.openai.com/v1"
    

高级功能

结构化输出（Pydantic 模型约束）

      from pydantic import BaseModel, Field

class PersonInfo(BaseModel):
    name: str = Field(description="人名")
    age: int = Field(description="年龄")
    occupation: str = Field(description="职业")

person = await ai.structured.extract(
    "张三今年28岁, 是一名软件工程师",
    PersonInfo,
)
print(person.name)   # 张三
    

@tool 装饰器

      from thinkai.agent.tool import tool

@tool
def search_web(query: str, num_results: int = 5) -> str:
    """Search the web for information."""
    return f"Results for: {query}"

agent = FunctionCallingAgent(tools=[search_web, calculate], ai_client=ai)
    

速率限制

      from thinkai.middleware.rate_limit import RateLimitMiddleware

ai.add_middleware(RateLimitMiddleware(
    requests_per_minute=60,
    burst=10,
    max_concurrent=10,
))
    

MCP（Model Context Protocol）

      from thinkai.mcp import MCPAdapter, MCPRegistry

adapter = MCPAdapter()
fs_config = MCPRegistry.filesystem("/workspace")
adapter.add_server(**fs_config)
await adapter.connect_all()

agent = adapter.create_agent(ai_client=ai)
result = await agent.run("读取 /workspace/readme.md 文件内容")
    

安全特性

代码执行沙箱

AST 验证 + 受限内置函数，禁止 import、文件访问、系统调用

数学表达式安全

AST 解析，只允许数学运算符，杜绝 eval() 注入

文件路径限制

默认只允许访问当前目录，可配置 allowed_dirs

环境变量保护

只允许读取指定前缀的环境变量

仓库地址

平台	地址
Gitee	https://gitee.com/hongxinge/think-ai
GitHub	https://github.com/hongxinge/ThinkAi
PyPI	https://pypi.org/project/thinkai-framework/