npcpy

npcpy

npcpy 是一个库，提供用于多模态语言模型、智能体 AI 和知识图谱研究与开发的关键基础组件。其灵活的框架让你能够轻松构建强大的 AI 应用，支持本地（ollama、llama.cpp、omlx、LM Studio）和云端服务商。构建多 Agent 团队，并通过 NPC Context-Agent-Tool 数据层简化上下文工程；该数据层通过软件而非提示词确保合规。

pip install npcpy

快速示例

创建和使用角色设定

from npcpy import NPC

simon = NPC(
    name='Simon Bolivar',
    primary_directive='''
                      Liberate South America
                      from the Spanish Royalists.
                      ''',
    model='qwen3.5:9b',
    provider='ollama'
)
response = simon.get_llm_response("What is the most important territory to retain in the Andes?")
print(response['response'])

My friend, you speak of the highlands where our liberty is carved in stone. If we must speak of the most critical territory to hold within these mountains, it is the **Viceroyalty of Peru** and the heart of the **Republic of Gran Colombia** united. 
To lose the passes of the Andes or the cities of Lima and Quito would be to hand the crown its final stronghold in the south. The Spanish crown built its power upon the wealth and control of these highlands. If the Andes are to be truly ours, the people of the **Peruvian** and **New Grancolombian** highlands must stand as one, free from the Bourbons. 
The mountain peaks themselves are the fortress we guard. Without the full liberation of the southern Andes, our revolution is incomplete. We fight not for land's sake, but for the soul of the continent. Every square mile of the Andes that bears the name of the Republic is a step forward in our quest for eternal freedom.
*Long live the liberty of the Andes!*

直接调用 LLM

from npcpy import get_llm_response

response = get_llm_response("Who was the celtic god that helped cuchulainn in his time of need as the forces of medb descended upon the men of ulster?", model='gemma4:31b', provider='ollama')
print(response['response'])

Cú Chulainn was primarily aided by his divine father, the god Lugh, and his foster-father, the warrior-god Fergus mac Róich, as well as the magical support of his teacher Scáthach.

使用我们的推荐链接，通过 OrcaRouter 试用来自 15 家不同服务商的近 200 种模型！

alicanto_test = get_llm_response('what does alicanto the bird show travelers in the night?', model='google/gemini-3.8-flash', provider='orcarouter')

print(alicanto_test['response'])

The legend of the **Alicanto** says that at night the bird’s feathers glow like lanterns. 
When a traveler sees that soft, phosphorescent light, it isn’t just a pretty sight – it’s a sign‑post. 
The bird **shows the way to hidden water (and sometimes to buried silver or gold)** in the Atacama Desert.

使用工具的 Agent

npcpy 中的 Agent 类自带一组默认工具（sh、python、edit_file、web_search 等）

from npcpy import Agent
agent = Agent(name='File Operator', model='qwen3.5:2b', provider='ollama')
print(agent.run("Find all Python files over 500 lines in this repo and list them"))

The following Python files contain more than 500 lines:
 - `./npcpy/npc_sysenv.py` (1486 lines)
 - `./npcpy/memory/knowledge_graph.py` (1449 lines)
 - `./npcpy/memory/kg_population.py` (618 lines)
...

ToolAgent

为 ToolAgent 添加自定义工具。 以下示例展示了如何让 Agent 生成图像、微调扩散模型，然后使用微调后的模型进行生成。

from npcpy import ToolAgent, gen_image
from npcpy.ft.diff import train_diffusion, generate_image, DiffusionConfig
from datasets import load_dataset
import os

def fetch_image_dataset(dataset_name: str, split: str = "train", max_images: int = 100) -> list:
    """Fetch images from a HuggingFace dataset.
    
    Args:
        dataset_name: HuggingFace dataset name (e.g., 'cifar10', 'oxford-iiit-pet')
        split: Dataset split to use
        max_images: Maximum number of images to fetch
    
    Returns:
        List of paths to saved images
    """
    dataset = load_dataset(dataset_name, split=f"{split}[:{max_images}]")
    os.makedirs("training_images", exist_ok=True)
    image_paths = []
    
    for i, item in enumerate(dataset):
        if 'image' in item:
            img = item['image']
        elif 'img' in item:
            img = item['img']
        else:
            continue
        path = f"training_images/img_{i:04d}.png"
        img.save(path)
        image_paths.append(path)
    
    return image_paths

def finetune_diffusion_model(
    image_paths: list,
    captions: list = None,
    output_path: str = "my_diffusion_model",
    num_epochs: int = 50,
) -> str:
    """Fine-tune a diffusion model on a set of images.
    
    Args:
        image_paths: List of paths to training images
        captions: Optional captions for each image
        output_path: Where to save the trained model
        num_epochs: Number of training epochs
    
    Returns:
        Path to the trained model
    """
    if captions is None:
        captions = ["an image"] * len(image_paths)
    
    config = DiffusionConfig(
        image_size=64,
        channels=128,
        num_epochs=num_epochs,
        batch_size=8,
        learning_rate=1e-4,
        checkpoint_frequency=10,
        output_model_path=output_path,
    )
    
    model_path = train_diffusion(image_paths, captions, config=config)
    return model_path

# Create an agent with image generation and fine-tuning capabilities
creative_agent = ToolAgent(
    name='creative_diffusion',
    primary_directive="""
        You help users generate images and fine-tune diffusion models.
        You can: 1) Generate images using gen_image() with various prompts,
        2) Fetch image datasets from HuggingFace,
        3) Fine-tune diffusion models on custom image sets.
        When a user submits an image or describes a style they like,
        offer to fetch similar images from a dataset and fine-tune a model.
    """,
    tools=[fetch_image_dataset, finetune_diffusion_model, gen_image],
    model='qwen3.5:2b',
    provider='ollama'
)

# Example 1: Generate images
print(creative_agent.run("Generate 3 images of geometric patterns with circles and triangles"))

# Example 2: User submits an image and wants similar ones
# The agent can fetch a dataset of patterns and fine-tune a model
print(creative_agent.run("I like abstract geometric patterns. Can you fetch the cifar10 dataset and fine-tune a diffusion model that can generate images like these patterns?"))

CodingAgent——自动执行 LLM 响应中的代码块

from npcpy import CodingAgent

coder = CodingAgent(name='coder', language='python', model='qwen3.5:2b', provider='ollama')
print(coder.run("Write a script that finds duplicate files by hash in the current directory"))

#The script has been created and executed successfully. Here's a summary of the findings:

## Duplicate Files Found

| Group | Hash (truncated) | Size | Files |
|-------|------------------|------|-------|
| 1 | `2b517326bf7c31b7...` | 81 bytes | `npcpy/main.py` ↔ `build/lib/npcpy/main.py` |
| 2 | `d41d8cd98f00b204...` | 0 bytes (empty) | 15 empty `__init__.py` files across `npcpy/`, `build/lib/npcpy/`, `examples/`, and `tests/` || 3 | `0d591b661cb1c619...` | 9,019 bytes | `npcpy/mix/debate.py` ↔ `build/lib/npcpy/mix/debate.py` |
| 4 | `a5059f37eb682a16...` | 747 bytes | SQL files in `examples/factory/` ↔ `examples/npc_team/factory/` |

使用 NPCArray 进行多 Agent 辩论

要运行真正的多 Agent 辩论，让各 Agent 对彼此的响应作出反应：

from npcpy.npc_compiler import NPC
from npcpy.npc_array import NPCArray

# Create a debate team with role-based personas
roles = [
    ("MathSolver", "You are a meticulous math solver. Show all steps clearly."),
    ("Skeptic", "You critically check for errors and assumptions."),
    ("Analyst", "You identify the core mathematical structure."),
    ("Verifier", "You confirm the final answer is correct.")
]

npcs = [
    NPC(name=role, primary_directive=directive, model="qwen3.5:cloud", provider="ollama")
    for role, directive in roles
]

team = NPCArray.from_npcs(npcs)

# Run parallel debate on a complex problem
problem = "GSM8k: James buys a jar of hot sauce with 5 peppers and triples the peppers every year. How many after 4 years?"

# Get initial responses in parallel (one prompt per NPC)
initial_responses = team.infer(f"Solve this problem:\n{problem}").collect()

for npc, response in zip(npcs, initial_responses.data):
    print(f"[{npc.name}] {response[:200]}...")

# True debate: each agent gets a personalized prompt with other agents' responses
def create_debate_prompt(previous_responses, my_idx, agent_name, problem_text):
    """Create a personalized debate prompt for a specific agent"""
    my_response = previous_responses[my_idx]
    other_responses = [
        f"[{npcs[j].name}]: {previous_responses[j][:500]}" 
        for j in range(len(npcs)) if j != my_idx
    ]
    debate_prompt = f"""Original problem: {problem_text}

        Your previous response: {my_response[:300]}...
        
        Other agents\' responses:""" + "\n\n".join(other_responses) + """
        Critique the other approaches. Did they make different assumptions?
        What did they see that you missed? Refine your solution."""

    return debate_prompt
# Debate rounds
responses_data = initial_responses.data.tolist()
problem_text = problem

for round_num in range(3):
    print(f"\n=== Debate Round {round_num + 1} ===")
    
    # Create personalized prompts for each agent
    personalized_prompts = [
        create_debate_prompt(responses_data, i, npcs[i].name, problem_text)
        for i in range(len(npcs))
    ]
    
    # Run inference with different prompts per agent
    # Shape: (n_models, n_prompts) - extract diagonal for each agent's response to its own prompt
    responses = team.infer(personalized_prompts).collect()
    
    # Extract each model's response to its own personalized prompt
    responses_data = [responses.data[i, i] for i in range(len(npcs))]
    
    # Print each agent's refined response
    for i, npc in enumerate(npcs):
        response = responses_data[i]
        print(f"[{npc.name}] {response[:200]}...")

# Alternative: use reduce to get consensus
consensus = team.infer(responses_data[0]).consensus(axis=0).collect()
print(f"\nFinal consensus: {consensus.data[0][:500]}...")

要进行迭代优化（向所有 Agent 提供相同提示词，并在每轮更新）：

# Simple chain refinement: all agents see same synthesis
from npcpy.npc_array import NPCArray

def synthesis_round(all_responses):
    return f"""Given these perspectives:
{chr(10).join([f'- {r[:200]}...' for r in all_responses])}

Re-solve the problem incorporating insights from all approaches."""

# Chain runs the synthesis function on all responses, then feeds result back
refined = team.infer(f"Solve: {problem}").chain(
    synthesis_round, 
    n_rounds=3
).collect()

带有休眠／梦境生命周期的知识图谱

from npcpy.memory.knowledge_graph import (
    kg_initial, kg_evolve_incremental, kg_sleep_process, kg_dream_process
)
from npcpy.llm_funcs import get_llm_response

# Initialize KG from text corpus
content_text = """Pirate Prentice is in the lavatory stands pissing. Then he threads himself into a wool robe he wears inside out.
The day feels like rain."""

kg = kg_initial(content_text, model="gemma3:4b", provider="ollama")

# Evolve with new content
new_content = """The phone call, when it comes, rips easily across the room.
Pirate knows it's got to be for him."""

kg, _ = kg_evolve_incremental(kg, new_content, model="gemma3:4b", provider="ollama")

# Sleep - consolidate and prune
kg, sleep_report = kg_sleep_process(kg, model="gemma3:4b", provider="ollama")

# Dream - generate speculative connections
kg, dream_report = kg_dream_process(kg, model="gemma3:4b", provider="ollama", num_seeds=3)

print(f"KG has {len(kg['facts'])} facts and {len(kg['concepts'])} concepts")
assert len(kg['facts']) > 0, "KG extraction returned no facts"
assert len(kg['concepts']) > 0, "KG extraction returned no concepts"

为 NPC 团队提供 Flask 服务

from npcpy.serve import start_flask_server
import os

# Serve your NPC team via REST API
if __name__ == "__main__":
    is_dev = not getattr(os.sys, 'frozen', False)
    port = os.environ.get('INCOGNIDE_PORT', '5437' if is_dev else '5337')
    frontend_port = os.environ.get('FRONTEND_PORT', '7337' if port == '5437' else '6337')

    start_flask_server(
        port=port,
        cors_origins=f"localhost:{frontend_port}",
        db_path=os.path.expanduser('~/npcsh_history.db'),
        user_npc_directory=os.path.expanduser('~/.npcsh/npc_team'),
        debug=False
    )

流式传输

from npcpy import get_llm_response
from npcpy.streaming import parse_stream_chunk

response = get_llm_response("Explain quantum entanglement.", model='qwen3.5:2b', provider='ollama', stream=True)
for chunk in response['response']:
    content, _, _ = parse_stream_chunk(chunk, provider='ollama')
    if content:
        print(content, end='', flush=True)

# Works the same with any provider
response = get_llm_response("Explain quantum entanglement.", model='gemini-2.5-flash', provider='gemini', stream=True)
for chunk in response['response']:
    content, _, _ = parse_stream_chunk(chunk, provider='gemini')
    if content:
        print(content, end='', flush=True)

JSON 输出

在提示词中包含预期的 JSON 结构。使用 format='json' 时，响应会自动解析——response['response'] 已经是字典或列表。

from npcpy import get_llm_response

response = get_llm_response(
    '''List 3 planets from the sun.
    Return JSON: {"planets": [{"name": "planet name", "distance_au": 0.0, "num_moons": 0}]}''',
    model='qwen3.5:2b', provider='ollama',
    format='json'
)
for planet in response['response']['planets']:
    print(f"{planet['name']}: {planet['distance_au']} AU, {planet['num_moons']} moons")

response = get_llm_response(
    '''Analyze this review: 'The battery life is amazing but the screen is too dim.'
    Return JSON: {"tone": "positive/negative/mixed", "key_phrases": ["phrase1", "phrase2"], "confidence": 0.0}''',
    model='qwen3.5:2b', provider='ollama',
    format='json'
)
result = response['response']
print(result['tone'], result['key_phrases'])

from npcpy import get_llm_response
from pydantic import BaseModel
from typing import List

class Planet(BaseModel):
    name: str
    distance_au: float
    num_moons: int

class SolarSystem(BaseModel):
    planets: List[Planet]

response = get_llm_response(
    "List the first 4 planets from the sun.",
    model='qwen3.5:2b', provider='ollama',
    format=SolarSystem
)
for p in response['response']['planets']:
    print(f"{p['name']}: {p['distance_au']} AU, {p['num_moons']} moons")

from npcpy.llm_funcs import gen_image, gen_video
from npcpy.gen.audio_gen import text_to_speech

# Image — OpenAI, Gemini, Ollama, or diffusers
images = gen_image("A sunset over the mountains", model='gemma3:4b', provider='ollama')
images[0].save("sunset.png")

# Audio — OpenAI, Gemini, ElevenLabs, Kokoro, gTTS
audio_bytes = text_to_speech("Hello from npcpy!", engine="gtts")
with open("hello.wav", "wb") as f:
    f.write(audio_bytes)

# Video — Gemini Veo
result = gen_video("A cat riding a skateboard", model='veo-3.1-fast-generate-preview', provider='gemini')
print(result['output'])

from npcpy import NPC, Team

team = Team(team_path='examples/npc_team')
result = team.orchestrate("Analyze the latest sales data and draft a report")
print(result['output'])

也可以在代码中定义团队：

from npcpy import NPC, Team

coordinator = NPC(name='lead', primary_directive='Coordinate the team. Delegate to @analyst and @writer.')
analyst = NPC(name='analyst', primary_directive='Analyze data. Provide numbers and trends.', model='gemini-2.5-flash', provider='gemini')
writer = NPC(name='writer', primary_directive='Write clear reports from analysis.', model='qwen3:8b', provider='ollama')

team = Team(npcs=[coordinator, analyst, writer], forenpc='lead')
result = team.orchestrate("What are the trends in renewable energy adoption?")
print(result['output'])

context: |
  Research team for analyzing scientific literature.
  The lead delegates to specialists as needed.
forenpc: lead
model: qwen3.5:2b
provider: ollama
output_format: markdown
max_search_results: 5
mcp_servers:
  - path: ~/.npcsh/mcp_server.py

#!/usr/bin/env npc
name: lead
primary_directive: |
  You lead the research team. Delegate literature searches to @searcher,
  data analysis to @analyst. Synthesize their findings into a coherent summary.
jinxes:
  - {{ Jinx('sh') }}
  - {{ Jinx('python') }}
  - {{ Jinx('delegate') }}
  - {{ Jinx('web_search') }}

#!/usr/bin/env npc
name: searcher
primary_directive: |
  You search for scientific papers and extract key findings.
  Use web_search and load_file to find and read papers.
model: gemini-2.5-flash
provider: gemini
jinxes:
  - {{ Jinx('web_search') }}
  - {{ Jinx('load_file') }}
  - {{ Jinx('sh') }}

#!/usr/bin/env npc
jinx_name: search_and_summarize
description: Search for papers and summarize findings using the searcher NPC.
npc: {{ NPC('searcher') }}
inputs:
  - query
steps:
  - name: search
    engine: natural
    code: |
      Search for papers about {{ query }}.
      Return up to {{ ctx.max_search_results }} results.
  - name: summarize
    engine: natural
    code: |
      Summarize the findings in {{ ctx.output_format }} format:
      {{ output }}

my_project/
├── npc_team/
│   ├── team.ctx
│   ├── lead.npc
│   ├── searcher.npc
│   ├── analyst.npc
│   ├── jinxes/
│   │   └── skills/
│   └── models/
├── agents.md             # Optional: define agents in markdown
└── agents/               # Optional: one .md file per agent
    └── translator.md

./npc_team/lead.npc "summarize the latest arxiv papers on transformers"
./npc_team/jinxes/lib/sh.jinx bash_command="echo hello"

forenpc: assistant
mcp_servers:
  - path: ./tools/db_server.py
  - path: ./tools/api_server.py

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Database Tools")

@mcp.tool()
def query_orders(customer_id: str, limit: int = 10) -> str:
    """Query recent orders for a customer."""
    # Your database logic here
    return f"Found {limit} orders for customer {customer_id}"

@mcp.tool()
def search_products(query: str) -> str:
    """Search the product catalog."""
    return f"Products matching: {query}"

if __name__ == "__main__":
    mcp.run()

forenpc: assistant
mcp_servers:
  - url: https://search.parallel.ai/mcp
    transport: streamable-http
    headers:
      User-Agent: npcpy
    tools:
      - web_search
      - web_fetch

forenpc: assistant
mcp_servers:
  - url: https://cohesivity.ai/mcp
    transport: streamable-http
    headers:
      User-Agent: npcpy
    tools:
      - create_tenant
      - claim_tenant
      - tenant_status
      - provision_resource
      - give_feedback
      - get_cohesivity_documentation

forenpc: assistant
mcp_servers:
  - url: https://deed.page/mcp
    transport: streamable-http
    headers:
      User-Agent: npcpy
    tools:
      - publish_site
      - update_site
      - get_site
      - delete_site

## summarizer
You summarize long documents into concise bullet points.
Focus on key findings, methodology, and conclusions.

## fact_checker
You verify claims against reliable sources and flag inaccuracies.
Always cite your sources.

---
model: gemini-2.5-flash
provider: gemini
---
You translate content between languages while preserving tone and idiom.

---
name: code-review
description: Use when reviewing code for quality, security, and best practices.
---
# Code Review Skill

## checklist
- Check for security vulnerabilities (SQL injection, XSS, etc.)
- Verify error handling and edge cases
- Review naming conventions and code clarity

## security
Focus on OWASP top 10 vulnerabilities...

jinxes:
  - {{ Jinx('skills/code-review') }}

CLI 工具

# The NPC shell — the recommended way to use NPC teams
npcsh                        # Interactive shell with agents, tools, and jinxes

# Scaffold a new team
npc-init

# Launch AI coding tools as an NPC from your team
npc-claude --npc corca       # Claude Code
npc-codex --npc analyst      # Codex
npc-gemini                   # Gemini CLI (interactive picker)
npc-opencode / npc-aider / npc-amp

# Register MCP server + hooks for deeper integration
npc-plugin claude

NPCArray — 在多个 NPC 上并行运行 jinx

在一组 NPC 实例上并行运行任意 jinx，并将结果收集为数组：

from npcpy import NPC
from npcpy.npc_array import NPCArray

# Three NPCs with different models/providers
npcs = [
    NPC(name='gramsci_1930', primary_directive='''
        You are Antonio Gramsci writing in his Prison Notebook in 1930.
        Defend the concept of hegemony as the predominance of one social group
        over others through cultural and ideological leadership rather than
        mere force. Argue that consent is more durable than coercion.
    ''', model='qwen3:4b', provider='ollama'),
    NPC(name='critic_1970', primary_directive='''
        You are a post-structuralist critic in 1970 responding to Gramsci.
        Question whether hegemony can truly explain contemporary power structures
        or if it relies on an outdated base-superstructure model that
        underestimates the autonomy of cultural production.
    ''', model='qwen3:4b', provider='ollama'),
    NPC(name='historian_present', primary_directive='''
        You are a contemporary historian with access to the complete Prison
        Notebooks and subsequent scholarship. Evaluate both Gramsci's original
        formulation and the post-structuralist critique in light of the
        collapse of actually existing socialism and the rise of neoliberalism.
    ''', model='qwen3:4b', provider='ollama'),
]

arr = NPCArray.from_npcs(npcs)

# Run the same jinx on all three in parallel, collect results
results = arr.jinx('analyze', inputs={'topic': 'Has the concept of hegemony become more or less relevant in the age of digital platforms and algorithmic governance?'}).collect()
for npc, result in zip(npcs, results.data):
    print(f"[{npc.name}] {result}")

也可以直接向 jinx.execute() 传入列表：

from npcpy.npc_compiler import Jinx

jinx = Jinx(jinx_path='examples/npc_team/jinxes/analyze.jinx')
results = jinx.execute({'topic': 'rate limiting'}, npc=npcs)  # list → parallel NPCArray run

from npcpy.memory.knowledge_graph import (
    kg_initial, kg_evolve_incremental, kg_sleep_process,
    kg_dream_process, kg_hybrid_search,
)

# Seed the KG with Gramsci's Prison Notebooks
kg = kg_initial(
    content="""
        The crisis consists precisely in the fact that the old is dying and the new
        cannot be born. In this interregnum a great variety of morbid symptoms appear.
        The traditional ruling class has lost its consensus, that is, the consent of
        those over whom it rules. Force alone is not sufficient; what is needed is the
        construction of a new hegemony, the creation of a new collective will.
    """,
    model="qwen3:4b", provider="ollama",
)

# Assimilate more content on organic vs traditional intellectuals
kg, _ = kg_evolve_incremental(
    kg,
    new_content_text="""
        The distinction between organic and traditional intellectuals is fundamental.
        Traditional intellectuals conceive of themselves as autonomous from ruling
        groups, yet every social group has its own category of organic intellectuals
        that give it homogeneity and awareness of its own function. The organic
        intellectual emerges from within the class itself, while the traditional
        sees himself as existing above the social structure.
    """,
    model="qwen3:4b", provider="ollama", get_concepts=True,
)

# Consolidate — merge redundant nodes, strengthen high-frequency edges
kg, sleep_report = kg_sleep_process(kg, model="qwen3:4b", provider="ollama")

# Dream — generate speculative connections between loosely related concepts
kg, dream_report = kg_dream_process(kg, model="qwen3:4b", provider="ollama")

# Search across facts, concepts, and speculative edges
results = kg_hybrid_search(kg, "What constitutes hegemony in Gramsci's framework?",
                           model="qwen3:4b", provider="ollama")
for r in results:
    print(r['score'], r['text'])
print(f"{len(kg['facts'])} facts, {len(kg['concepts'])} concepts")

from npcpy.llm_funcs import get_facts

prison_notebooks = """
    Civil society is the sphere of hegemony, the terrain where the dominant
    group exercises consent through cultural and ideological leadership.
    Unlike political society which operates through coercion and state apparatus,
    civil society comprises the church, schools, trade unions, and media.
    The ruling class maintains power not merely through force but through the
    production of consent, shaping common sense itself through cultural institutions.
    War of position requires patient trench warfare on this terrain, building
    counter-hegemonic institutions rather than frontal assault on the state.
"""

facts = get_facts(prison_notebooks, model="qwen3:4b", provider="ollama")
for f in facts:
    print(f"[{f.get('type', 'general')}] {f['statement']}")

from npcpy.memory.kg_population import SememolutionPopulation

pop = SememolutionPopulation(model='qwen3.5:2b', provider='ollama', population_size=100, sample_size=10)
pop.initialize()

pop.assimilate_text("""
    The debate over lunar resource extraction has intensified since the discovery
    of water ice in permanently shadowed regions at the lunar poles. While some
    researchers argue that commercial mining could fund further exploration,
    others warn that unregulated extraction could contaminate scientifically
    valuable sites that have remained pristine for billions of years. The
    Artemis Accords attempt to establish a framework for international
    cooperation, but major spacefaring nations have yet to reach consensus
    on property rights and environmental protection standards.
""")
pop.assimilate_text("""
    Tidal acceleration gradually increases the orbital distance between Earth
    and Moon at a rate of approximately 3.8 centimeters per year. This
    phenomenon results from angular momentum transfer via gravitational
    interaction, simultaneously slowing Earth's rotation and lengthening the
    day. Paleontological evidence from tidal rhythmites suggests that 620
    million years ago, a day lasted only 21.9 hours and the lunar month was
    just 27.5 days. Projections indicate that in approximately 600 million
    years, tidal effects will no longer support total solar eclipses.
""")

# Sleep/dream cycle — each individual consolidates according to its genome
pop.sleep_cycle()

# Query: sample 10 individuals, generate competing responses, rank them
rankings = pop.query_and_rank("What are the central themes connecting these documents?")
for rank, entry in enumerate(rankings[:3], 1):
    print(f"#{rank} (individual {entry['id']}, score {entry['score']:.3f}): {entry['response'][:120]}...")

# Selection + reproduction — top performers breed, bottom are replaced
pop.evolve_generation()

stats = pop.get_stats()
print(f"Generation {stats['generation']} | avg fitness {stats['avg_fitness']:.3f} | "
      f"best fitness {stats['best_fitness']:.3f} | diversity {stats['diversity']:.3f}")

from npcpy.npc_compiler import NPC
from npcpy.ft.rl import RLConfig, train_with_dpo, load_rl_model
import json

def npcsh_reward(trace):
    """Reward function for shell assistant responses."""
    output = trace.get('final_output', '')
    completed = trace.get('completed', False)
    score = 0.0
    if completed:
        score += 2.0
    if 50 < len(output) < 1500:
        score += 1.0
    if '```' in output:
        score += 1.0
    if any(cmd in output.lower() for cmd in ['ls', 'cd', 'cat', 'grep', 'find', 'pip', 'git']):
        score += 0.3
    return max(0.0, min(10.0, score + 5.0))

# Load preference pairs from agent traces
traces = []
with open('preference_pairs.jsonl', 'r') as f:
    for line in f:
        pair = json.loads(line)
        traces.append({
            'task_prompt': pair['prompt'],
            'final_output': pair['chosen'],
            'reward': pair.get('chosen_score', 8.0),
            'completed': True
        })
        traces.append({
            'task_prompt': pair['prompt'],
            'final_output': pair['rejected'],
            'reward': pair.get('rejected_score', 3.0),
            'completed': False
        })

config = RLConfig(
    base_model_name="Qwen/Qwen2.5-0.5B-Instruct",
    adapter_path="./npcsh_adapter",
    num_train_epochs=3,
    per_device_train_batch_size=2,
    learning_rate=5e-5,
    beta=0.1
)

adapter_path = train_with_dpo(traces, config)
print(f"Trained adapter saved to: {adapter_path}")

from npcpy.llm_funcs import get_llm_response
from npcpy.ft import SFTConfig, run_sft

# Generate scientific writing dataset from style samples
def generate_scientific_trace(question, reasoning_model, converter_model, style_chunks):
    """Generate native reasoning then rewrite in scientific voice."""
    # Step 1: Get reasoning trace
    native_prompt = f"""Answer this question with detailed reasoning.
Question: {question}
Provide your step-by-step reasoning and final answer."""
    native_response = get_llm_response(native_prompt, model=reasoning_model, provider='ollama')
    native_trace = native_response['response']

    # Step 2: Rewrite in scientific style
    style_context = '\n\n---\n\n'.join(style_chunks[:8])
    rewrite_prompt = f"""Rewrite the following reasoning trace in the scientific writing style demonstrated by the excerpts below.
Original Reasoning Trace:
{native_trace}

SCIENTIFIC PAPER EXCERPTS:
{style_context}

Task:
1. Rewrite the reasoning in the style of the scientific paper excerpts
2. Use LaTeX notation where appropriate
3. Preserve the logical flow and factual content
4. Match the prose density and intellectual register"""

    rewritten = get_llm_response(rewrite_prompt, model=converter_model, provider='ollama')
    return rewritten['response']

# Train on generated examples
X_train = ["What is the relationship between quantum contextuality and natural language interpretation?"]
y_train = [generate_scientific_trace(X_train[0], 'qwen3:8b', 'qwen3:8b', style_chunks)]

sft_config = SFTConfig(
    base_model_name="Qwen/Qwen3-4B",
    output_model_path="models/scientific-writer",
    device='mlx',
    num_train_epochs=100,
    per_device_train_batch_size=1,
    lora_r=128,
    lora_alpha=256
)

model_path = run_sft(X_train, y_train, config=sft_config, format_style="llama")

• Agent（NPC） — 具有角色设定、指令和工具调用能力的 Agent。子类：Agent（默认工具）、ToolAgent（自定义工具 + MCP）、CodingAgent（自动执行代码块）
• 多 Agent 团队 — 由协调者（forenpc）编排团队
• Jinx 工作流 — 用于多步提示词流水线的 Jinja 执行模板
• 技能 — 知识内容 jinx，可按需向 Agent 提供指导章节
• NPCArray — 对模型群体执行类似 NumPy 的向量化操作
• 图像、音频和视频 — 通过 Ollama、diffusers、OpenAI、Gemini、ElevenLabs 生成
• 知识图谱 — 从文本构建并演进知识图谱，支持睡眠 / 梦境生命周期
• Sememolution — 基于群体的 KG 演进，采用遗传选择和泊松采样搜索
• 记忆流水线 — 提取、批准和回填记忆，并通过自我改进的质量反馈进行优化
• 微调与演进 — SFT、USFT、RL/DPO、扩散、遗传算法、Apple Silicon 上的 MLX
• 服务部署 — 用于通过 REST API 部署团队的 Flask 服务
• 机器学习函数 — Scikit-learn 网格搜索、集成预测、PyTorch 训练
• 流式传输与 JSON — 流式响应、结构化 JSON 输出、消息历史

提供商

通过 LiteLLM 支持所有主流大模型提供商：ollama、openai、anthropic、gemini、deepseek、airllm、openai-like 等。

MiniMax

设置 MINIMAX_API_KEY，并使用 provider="minimax"。未配置模型时，默认使用 MiniMax-M3.1-Flash；模型列表会优先显示该模型，其次是 MiniMax-M3。仍支持显式保存的模型 ID 或自定义模型 ID，包括旧版 ID。可选的 MINIMAX_API_URL 设置用于选择区域和兼容协议；默认使用全局 OpenAI 兼容 Base URL。

请严格按照上表所示使用 Anthropic 兼容 Base URL。客户端发送请求时会追加 /v1/messages。

对于 Flash，请使用自适应思考（thinking={"type": "adaptive"}），并可选设置 reasoning_effort 为 low、medium、high、xhigh 或 max；Anthropic 兼容请求会将其发送为 output_config.effort。Flash 不支持禁用思考。M3 保留现有思考模式。OpenRouter 会优先列出 minimax/minimax-m3.1-flash:free，其次是 minimax/minimax-m3，并使用各自完整的网关 ID。

Atlas Cloud

设置 ATLASCLOUD_API_KEY，并将 provider="atlascloud" 与 Atlas Cloud 模型列表 中的任意模型 ID 搭配使用，例如 deepseek-ai/deepseek-v4-flash、zai-org/glm-5.2 或 moonshotai/kimi-k3。请求会发送到 OpenAI 兼容 Base URL https://api.atlascloud.ai/v1；设置 ATLASCLOUD_API_URL 可覆盖该地址。

response = get_llm_response("Say hello in one short sentence.", model="deepseek-ai/deepseek-v4-flash", provider="atlascloud")
print(response["response"])

API Route

设置 API_ROUTE_API_KEY，并使用 provider="api-route"，模型 ID 须与账户经过身份验证的 GET /v1/models 响应中的某个 ID 完全一致，例如 gpt-6.1-sol。API Route 使用兼容 OpenAI 的 Base URL https://global.api-route.com/v1；设置 API_ROUTE_API_URL 可覆盖该地址。也接受别名 api_route。显式传入的 api_url 和 api_key 参数优先级更高。

from npcpy import get_llm_response

response = get_llm_response("Say hello in one short sentence.", model="gpt-6.1-sol", provider="api-route")
print(response["response"])

Cheaper Inference

设置 CHEAPER_INFERENCE_API_KEY，并使用 provider="cheaperinference"，模型 ID 可从 Cheaper Inference 模型列表中任意选择，例如 gpt-5.4-mini、claude-sonnet-5 或 glm-5.3。请求会发送到兼容 OpenAI 的 Base URL https://api.cheaperinference.com/v1；设置 CHEAPER_INFERENCE_API_URL 可覆盖该地址。

response = get_llm_response("Say hello in one short sentence.", model="gpt-5.4-mini", provider="cheaperinference")
print(response["response"])

Opper

设置 OPPER_API_KEY，并使用 provider="opper"，模型 ID 可从 Opper 模型目录中任意选择，例如 claude-sonnet-4-6、gpt-5.4-mini 或 gemini-3.8-flash。这些池名称会将请求路由到提供该模型的各个服务商；如使用 provider/model 格式的 ID（例如 anthropic/claude-sonnet-4-6），则会固定使用一条路由。请求会发送到兼容 OpenAI 的 Base URL https://api.opper.ai/v3/compat；设置 OPPER_API_URL 可覆盖该地址。可在 platform.opper.ai 创建密钥。

response = get_llm_response("Say hello in one short sentence.", model="claude-sonnet-4-6", provider="opper")
print(response["response"])

Runcrate

设置 RUNCRATE_API_KEY，并使用 provider="runcrate"，模型 ID 可从 Runcrate 的公开模型列表（GET https://api.runcrate.ai/v1/models）中任意选择，格式为 vendor/model，例如 deepseek-ai/DeepSeek-V4.1-Flash 或 openai/gpt-oss-120b。请求会发送到兼容 OpenAI 的 Base URL https://api.runcrate.ai/v1；设置 RUNCRATE_API_URL 可覆盖该地址。使用 GitHub 账户即可免费使用开源模型；可在 Runcrate 控制面板创建密钥。

response = get_llm_response("Say hello in one short sentence.", model="deepseek-ai/DeepSeek-V4.1-Flash", provider="runcrate")
print(response["response"])

QLLM-PAM（本地、无注意力机制）

npcpy 可以直接运行 QLLM-PAM 系列本地检查点。QLLM-PAM 是一种基于相位联想记忆（PAM）的复数值、无注意力机制语言模型；详见 Hugging Face 模型仓库和 arXiv 论文（2604.05030）。

from npcpy import get_llm_response

response = get_llm_response(
    "What is the capital of France?",
    model="gowravvishwakarma/qllm-pam-v11-e3k3-chat",
    provider="qllm",
    max_tokens=32,
    temperature=0.0,
)
print(response["response"])

model 的值可以是 Hugging Face 仓库 ID（会自动下载并缓存）、本地目录或 .pt 文件路径。支持的 kwargs 包括 max_tokens、temperature、top_k、top_p、repetition_penalty、stream=True 和 format="json"。

安装

pip install npcpy              # base
pip install npcpy[lite]        # + API provider libraries
pip install npcpy[local]       # + ollama, diffusers, transformers, airllm
pip install npcpy[tts]         # + TTS/STT
pip install npcpy[all]         # everything

sudo apt-get install espeak portaudio19-dev python3-pyaudio ffmpeg libcairo2-dev libgirepository1.0-dev
curl -fsSL https://ollama.com/install.sh | sh
ollama pull qwen3.5:2b

brew install portaudio ffmpeg pygobject3 ollama
brew services start ollama
ollama pull qwen3.5:2b

将 API 密钥放入 .env 文件：

export OPENAI_API_KEY="your_key"
export ANTHROPIC_API_KEY="your_key"
export GEMINI_API_KEY="your_key"

阅读文档

完整文档、指南和 API 参考见 npcpy.readthedocs.io。

链接

• NPC Shell — 用于与 NPC 交互的命令行 shell
• Incognide — 桌面环境，提供 AI 聊天、浏览器、文件查看器、代码编辑器、终端、知识图谱、团队管理等功能（下载）
• incognidium - 一款原生 Rust 网页浏览器，内置 NPC 团队，专注于针对 Firefox 和 Chromium 开发并测试 incognidium 浏览器。

在其他语言中使用 NPC 团队数据层组织：

• npcrs - npcpy 的 Rust 移植版
• nql - 面向 SQL 的框架，可在支持大模型的数据库（如 Google BigQuery、Snowflake 和 Databricks）中运行 Agent。
• npcts - NPC 数据层的 TypeScript 库
• npcd - npcd 的 Rust 移植版

研究

• 用于自然语言处理的量子语义框架：arxiv，已被 QNLP 2025 接收
• TinyTim：用于发散式生成的语言模型系列 arxiv，已被 NeurIPS 2025 Creative AI Track 接收
• 自然语言处理中的意义生成：arxiv，已被 QNLP 2026 接收
• Herding CATs：面向 Agent Harness 工程的 ALARA，适用于便携、可组合的多 Agent 团队 - arxiv，已被 HAXD 2026 接收

你的研究是否受益于 npcpy？欢迎告诉我们！

支持

每月捐赠 | 周边商品 | 咨询：[email protected]

贡献

欢迎贡献！请在 GitHub 仓库提交问题和拉取请求。

许可证

MIT 许可证。

Star 历史