Agency Agents: 125K+ Star Open-Source AI Agency Framework

# Clone the repository
git clone https://github.com/msitarzewski/agency-agents.git
cd agency-agents

# 종속성 설치
pip 설치 -r 요구사항.txt

# AI API 키 구성
AI_API_KEY="your-api-key-here" 내보내기
내보내기 AI_MODEL="claude-sonnet-4-20250514"

from codebase_memory import Indexer

# Initialize indexer
indexer = Indexer(
    codebase_path="./my-project",
    embedding_model="sentence-transformers/all-MiniLM-L6-v2",
    storage_backend="chroma"
)

# Index the entire codebase
results = indexer.index()
print(f"Indexed {results['files']} files, {results['tokens']} tokens")
# Output: Indexed 342 files, 1,247,832 tokens

# Get semantic similarity for a query
query = "How does the authentication flow work?"
similar = indexer.search(query, top_k=5)
for doc in similar:
    print(f"[{doc['score']:.2f}] {doc['path']}: {doc['snippet'][:100]}")

{
  "mcpServers": {
    "codebase-memory": {
      "command": "npx",
      "args": [
        "-y",
        "@deusdata/codebase-memory-mcp"
      ],
      "env": {
        "INDEX_PATH": "/path/to/your/codebase",
        "VECTOR_STORE": "chroma",
        "EMBEDDING_MODEL": "all-MiniLM-L6-v2"
      }
    }
  }
}

{
  "mcpServers": {
    "codebase-memory": {
      "command": "python",
      "args": ["-m", "codebase_memory.server"],
      "env": {
        "INDEX_PATH": "~/projects/my-app",
        "PERSIST": "true"
      }
    }
  }
}

// Rust incremental indexer
pub struct IncrementalIndexer {
    file_hashes: HashMap<PathBuf, String>,
    vector_store: ChromaStore,
}

impl IncrementalIndexer {
    pub fn index_changed(&mut self, codebase_path: &Path) -> IndexResult {
        let mut changed_files = Vec::new();
        let mut deleted_files = Vec::new();
        
        for entry in walk_dir(codebase_path)? {
            let current_hash = compute_hash(&entry.path)?;
            
            match self.file_hashes.get(&entry.path) {
                Some(stored_hash) if stored_hash != &current_hash => {
                    changed_files.push(entry.path);
                }
                None => {
                    changed_files.push(entry.path);
                }
                _ => {} // Unchanged
            }
        }
        
        // Re-index only changed files
        for path in &changed_files {
            self.vector_store.update(path)?;
        }
        
        Ok(IndexResult {
            indexed: changed_files.len(),
            skipped: 0,
            duration_ms: elapsed.as_millis() as u64,
        })
    }
}

class SearchPipeline:
    def __init__(self, vector_store, reranker=None):
        self.store = vector_store
        self.reranker = reranker
    
    def search(self, query: str, top_k: int = 10) -> List[Document]:
        # Step 1: Embed the query
        query_embedding = self._embed(query)
        
        # Step 2: Retrieve candidate documents
        candidates = self.store.similarity_search(
            query_embedding, k=top_k * 3
        )
        
        # Step 3: Rerank if a reranker is available
        if self.reranker:
            candidates = self.reranker.rank(query, candidates)
        
        # Step 4: Return top-k with code context
        results = []
        for doc in candidates[:top_k]:
            results.append({
                'path': doc.path,
                'snippet': doc.extract_context(window=5),
                'score': doc.score,
                'language': doc.language,
            })
        
        return results

from codebase_memory.parsers import BaseParser, register_parser

@register_parser("mylang")
class MyLangParser(BaseParser):
    def parse(self, file_path):
        with open(file_path) as f:
            content = f.read()
        segments = []
        for match in re.finditer(r"(def|class|module)\s+(\w+)", content):
            segments.append({
                "type": match.group(1),
                "name": match.group(2),
                "content": content[match.start():match.end()+200],
                "line": content[:match.start()].count("\n") + 1,
            })
        return segments

indexer = Indexer(
    codebase_path="./project",
    exclude_patterns=[
        "**/node_modules/**",
        "**/__pycache__/**",
        "**/*.lock",
        "**/test/fixtures/**",
    ],
    include_patterns=[
        "**/*.py",
        "**/*.ts",
        "**/*.go",
        "**/src/**",
    ]
)

from sentence_transformers import SentenceTransformer

code_model = SentenceTransformer("Salesforce/codet5p-220m-paraphrase")

indexer = Indexer(
    codebase_path="./project",
    embedding_model=code_model,
    embedding_dimension=220,
)

repositories = [
    "/home/user/project-alpha",
    "/home/user/project-beta",
    "/home/user/shared-libraries",
]

multi_indexer = MultiRepoIndexer(
    repositories=repositories,
    shared_embeddings=True,
    cross_reference_resolution=True,
)

results = multi_indexer.search("authentication flow")

# Use the Frontend Designer agent
python agents/frontend_designer.py --task "Create a landing page for a SaaS product"

# 백엔드 개발자 에이전트 사용
python Agents/backend_dev.py --task "인증을 사용하여 REST API 구축"

# DevOps 엔지니어 에이전트를 사용합니다.
python Agents/devops.py --task "Docker 및 GitHub Actions를 사용하여 CI/CD 파이프라인 설정"

# Run the complete agency workflow
python agency.py --project "Build a task management app" --agents all

# 특정 에이전트와 함께 실행
python Agency.py --project "작업 관리 앱 구축" \
--에이전트 프런트엔드, 백엔드, DevOps, qa

# 대화형 모드로 실행
Python Agency.py --interactive

# Create a new project directory
mkdir my-ai-project
cd my-ai-project

# 에이전시 작업공간 초기화
python -m Agency_agents init --project "내 SaaS 대시보드"

# 그러면 다음과 같은 구조가 생성됩니다:
# my-ai-프로젝트/
# ├── 에이전트/
# │ ├── config.yaml
# │ └── 작업.yaml
# ├── 출력/
# ├── 로그/
# └── README.md

team:
  frontend:
    model: claude-sonnet-4-20250514
    temperature: 0.3
    max_tokens: 4096
  backend:
    model: claude-sonnet-4-20250514
    temperature: 0.2
    max_tokens: 4096
  devops:
    model: claude-sonnet-4-20250514
    temperature: 0.1
    max_tokens: 2048
  qa:
    model: claude-sonnet-4-20250514
    temperature: 0.2
    max_tokens: 2048

project:
  name: "SaaS Dashboard"
  description: "A real-time analytics dashboard for e-commerce"
  tech_stack:
    - React
    - Node.js
    - PostgreSQL
    - Redis
  milestones:
    - name: "UI Design"
      agent: frontend
      deadline: "Day 1-2"
    - name: "API Development"
      agent: backend
      deadline: "Day 2-4"
    - name: "Infrastructure Setup"
      agent: devops
      deadline: "Day 3-4"
    - name: "Testing"
      agent: qa
      deadline: "Day 5-6"

# Run the full agency pipeline
python -m agency_agents run --tasks tasks.yaml --config config.yaml

# 진행상황을 실시간으로 모니터링
python -m Agency_agents 모니터 --follow

# 개별 에이전트 출력 보기
python -m Agency_agents 출력 --agent frontend --latest

# Check the output directory
tree output/

# QA 보고서 보기
고양이 출력/qa-report.md

# 자동화된 테스트 실행
CD 출력 && npm 테스트

class TaskRouter:
    def __init__(self, agents):
        self.agents = agents
        self.keywords = self._build_keyword_index()

def _build_keyword_index(self):
반품 {
'프런트엔드': ['ui', 'css', 'html', '반응', 'vue', '컴포넌트'],
'백엔드': ['api', '데이터베이스', '서버', '경로', '엔드포인트'],
'devops': ['docker', 'ci-cd', '배포', '파이프라인', 'kubernetes'],
'qa': ['테스트', '사양', '주장', '적용 범위'],
# ... 추가 매핑
}

def 경로(자기, 작업_설명):
점수 = {}
Agent_name의 경우 self.keywords.items()의 키워드:
점수 = 합계(task_description.lower()의 kw인 경우 키워드의 kw에 대해 1)
점수[agent_name] = 점수

최대값(점수, 키=scores.get)을 반환합니다.

from queue import Queue
import threading

클래스 AgentQueue:
def __init__(self):
self.tasks = 대기열()
self.results = {}

def add_task(self, task, 에이전트_유형, 우선순위=0):
self.tasks.put({
'작업': 작업,
'에이전트': 에이전트_유형,
'우선순위': 우선순위,
'타임스탬프': datetime.now()
})

def get_next_task(self):
return self.tasks.get(block=False)

def quality_check(agent_output, task_requirements):
    checks = [
        ('syntax', check_syntax(agent_output)),
        ('completeness', check_completeness(agent_output, task_requirements)),
        ('security', check_security(agent_output)),
        ('performance', check_performance(agent_output)),
    ]

Passed = all(체크인 수표의 경우 check[1])
반품 {
'통과': 통과,
'수표': 수표,
'점수': sum(c[1] for c in checks) / len(checks)
}

docker-compose up -d
docker exec -it agency-agents python agency.py --project "Build a web app"