Kurumsal çoklu ajan
sistem mimarisi.

Orkestratör, yetki kapısı, uzman ajanlar, devir ve rapor düğümü.

"""Read-only corporate multi-agent graph demonstration.

All figures and SOP entries below are synthetic. No LLM, ERP, email, or
production database is called. CLI roles are demonstrations, not login.
"""

from __future__ import annotations

import argparse
import json
import operator
import re
from typing import Annotated, Literal, TypedDict

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel, Field


Category = Literal["auto", "finance", "sop_audit", "sales"]


class TaskRequest(BaseModel):
    query: str = Field(min_length=3, max_length=600)
    category: Category = "auto"
    user_id: str = Field(min_length=1)
    scopes: frozenset[str]


class CorporateState(TypedDict, total=False):
    messages: Annotated[list[BaseMessage], operator.add]
    query: str
    user_id: str
    scopes: frozenset[str]
    task_category: str
    sop_segment: str | None
    active_agent: str
    needs_sop: bool
    findings: list[str]
    evidence: list[str]
    output_report: str
    compliance_status: bool | None
    review_required: bool
    audit_log: Annotated[list[str], operator.add]


DEMO_ROLES: dict[str, frozenset[str]] = {
    "finance": frozenset({"finance:read", "sop:finance"}),
    "sop": frozenset({"sop:audit", "sop:finance", "sop:sales"}),
    "sales": frozenset({"sales:read", "sop:sales"}),
}

SYNTHETIC_BUDGET = {"period": "2026-09", "budget": 100_000, "actual": 112_000}
SYNTHETIC_SOPS = [
    {"id": "SOP-FIN-01", "scope": "finance", "text": "Bütçe sapması %5'i aşarsa CFO incelemesi gerekir."},
    {"id": "SOP-SAT-01", "scope": "sales", "text": "Teklif, yetki limitine göre satış direktörü onayına sunulur."},
]


def _tokens(value: str) -> set[str]:
    return set(re.findall(r"\w+", value.casefold()))


def retrieve_sops(query: str, scopes: frozenset[str], segment: str) -> list[dict[str, str]]:
    """Small local retrieval example; scope is checked before ranking."""
    required = f"sop:{segment}"
    if required not in scopes:
        raise PermissionError(f"Gerekli SOP yetkisi yok: {required}")
    candidates = [doc for doc in SYNTHETIC_SOPS if doc["scope"] == segment]
    ranked = sorted(candidates, key=lambda doc: len(_tokens(query) & _tokens(doc["text"])), reverse=True)
    return ranked[:3]


def orchestrator_agent(state: CorporateState) -> CorporateState:
    query = state["query"].casefold()
    category = state.get("task_category", "auto")
    if category == "auto":
        if any(word in query for word in ("bütçe", "nakit", "finans", "sapma")):
            category = "finance"
        elif any(word in query for word in ("sop", "prosedür", "uyum")):
            category = "sop_audit"
        elif any(word in query for word in ("satış", "müşteri", "teklif")):
            category = "sales"
        else:
            raise ValueError("Görev sınıflandırılamadı; kategori açıkça seçilmeli.")
    if category not in ("finance", "sop_audit", "sales"):
        raise ValueError(f"Bilinmeyen görev kategorisi: {category}")
    needs_sop = category == "finance" and any(word in query for word in ("sop", "prosedür", "uyum"))
    sop_segment = "finance" if category == "finance" or any(word in query for word in ("finans", "bütçe", "nakit")) else "sales" if any(word in query for word in ("satış", "müşteri", "teklif")) else None
    if category == "sop_audit" and sop_segment is None:
        raise ValueError("SOP denetimi için finans veya satış kapsamını belirtin.")
    return {"task_category": category, "needs_sop": needs_sop, "sop_segment": sop_segment,
            "active_agent": "orchestrator", "audit_log": [f"route:{category}"]}


def authorize_task(state: CorporateState) -> CorporateState:
    required = {"finance": "finance:read", "sop_audit": "sop:audit", "sales": "sales:read"}[state["task_category"]]
    if required not in state["scopes"]:
        raise PermissionError(f"Görev yetkisi yok: {required}")
    if state.get("needs_sop") and "sop:finance" not in state["scopes"]:
        raise PermissionError("Finans SOP yetkisi yok: sop:finance")
    if state["task_category"] == "sop_audit" and f"sop:{state['sop_segment']}" not in state["scopes"]:
        raise PermissionError(f"SOP erişim yetkisi yok: sop:{state['sop_segment']}")
    return {"audit_log": [f"authorized:{state['user_id']}:{required}"]}


def route_task(state: CorporateState) -> str:
    return {"finance": "finance_agent", "sop_audit": "sop_agent", "sales": "sales_agent"}[state["task_category"]]


def finance_specialist_agent(state: CorporateState) -> CorporateState:
    row = SYNTHETIC_BUDGET
    variance = (row["actual"] - row["budget"]) / row["budget"] * 100
    finding = f"Sentetik {row['period']} bütçe sapması: %{variance:.1f} (bütçe {row['budget']:,}, gerçekleşen {row['actual']:,})."
    return {"active_agent": "finance", "findings": [finding],
            "evidence": ["demo:synthetic_budget:2026-09"], "audit_log": ["finance:read-only"]}


def route_after_finance(state: CorporateState) -> str:
    return "sop_agent" if state.get("needs_sop") else "report"


def sop_audit_agent(state: CorporateState) -> CorporateState:
    segment = state["sop_segment"]
    docs = retrieve_sops(state["query"], state["scopes"], segment)
    findings = list(state.get("findings", []))
    findings += [f"İlgili SOP alıntısı ({doc['id']}): {doc['text']}" for doc in docs]
    evidence = list(state.get("evidence", [])) + [f"demo:{doc['id']}" for doc in docs]
    return {"active_agent": "sop_audit", "findings": findings, "evidence": evidence,
            "audit_log": [f"sop:read:{segment}"]}


def sales_specialist_agent(state: CorporateState) -> CorporateState:
    return {"active_agent": "sales", "findings": ["Sentetik satış talebi sınıflandırıldı; teklif taslağı insan incelemesine hazır."],
            "evidence": ["demo:synthetic_sales_request"], "audit_log": ["sales:draft-only"]}


def report_agent(state: CorporateState) -> CorporateState:
    lines = [f"Görev: {state['query']}", f"Yönlendirme: {state['task_category']}"]
    lines += state.get("findings", [])
    lines += ["Kanıtlar: " + ", ".join(state.get("evidence", [])),
              "Durum: Demo çıktısı; SOP uyumu doğrulanmadı. İnsan incelemesi gerekli."]
    report = "\n".join(lines)
    return {"active_agent": "report", "output_report": report, "compliance_status": None,
            "review_required": True, "messages": [AIMessage(content=report)],
            "audit_log": ["report:review-required"]}


def build_graph():
    graph = StateGraph(CorporateState)
    graph.add_node("orchestrator", orchestrator_agent)
    graph.add_node("authorize", authorize_task)
    graph.add_node("finance_agent", finance_specialist_agent)
    graph.add_node("sop_agent", sop_audit_agent)
    graph.add_node("sales_agent", sales_specialist_agent)
    graph.add_node("report", report_agent)
    graph.add_edge(START, "orchestrator")
    graph.add_edge("orchestrator", "authorize")
    graph.add_conditional_edges("authorize", route_task, {
        "finance_agent": "finance_agent", "sop_agent": "sop_agent", "sales_agent": "sales_agent"
    })
    graph.add_conditional_edges("finance_agent", route_after_finance, {
        "sop_agent": "sop_agent", "report": "report"
    })
    graph.add_edge("sop_agent", "report")
    graph.add_edge("sales_agent", "report")
    graph.add_edge("report", END)
    return graph.compile()


def run(request: TaskRequest) -> CorporateState:
    initial: CorporateState = {
        "messages": [HumanMessage(content=request.query)], "query": request.query,
        "user_id": request.user_id, "scopes": request.scopes,
        "task_category": request.category, "findings": [], "evidence": [], "audit_log": []
    }
    return build_graph().invoke(initial)


def main() -> None:
    parser = argparse.ArgumentParser(description="Sentetik, salt-okunur LangGraph çoklu ajan örneği")
    parser.add_argument("query", nargs="?", default="Geçen ayın bütçe sapmalarını analiz et ve SOP kurallarına göre raporla")
    parser.add_argument("--role", choices=DEMO_ROLES, default="finance", help="Yalnızca demo rolü; gerçek kimlik doğrulama değildir")
    parser.add_argument("--category", choices=("auto", "finance", "sop_audit", "sales"), default="auto")
    args = parser.parse_args()
    request = TaskRequest(query=args.query, category=args.category, user_id=f"demo-{args.role}", scopes=DEMO_ROLES[args.role])
    result = run(request)
    print(json.dumps({"report": result["output_report"], "audit_log": result["audit_log"],
                      "review_required": result["review_required"], "compliance_status": result["compliance_status"]}, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

Kullanıcı kapsamı kontrolünden sonra RPC çağrısı yapan isteğe bağlı bağlayıcı.

"""Optional Supabase bridge. This demo does not call it by default.

Supply an authenticated, user-scoped Supabase client and a local embedding
function. Never use a service-role client for end-user retrieval.
"""

from collections.abc import Callable
from typing import Any


def query_private_sop_knowledge(
    query_text: str,
    *,
    scopes: frozenset[str],
    requested_scope: str,
    client: Any,
    embed_locally: Callable[[str], list[float]],
    limit: int = 3,
) -> list[dict[str, Any]]:
    if requested_scope not in ("finance", "sales"):
        raise ValueError("Bilinmeyen SOP kapsamı")
    if f"sop:{requested_scope}" not in scopes:
        raise PermissionError("SOP okuma yetkisi yok")
    if not 1 <= limit <= 10:
        raise ValueError("limit 1–10 arasında olmalı")
    vector = embed_locally(query_text)
    if len(vector) != 384:
        raise ValueError("Örnek şema 384 boyutlu yerel embedding bekler")
    response = client.rpc("match_sop_documents", {
        "query_embedding": vector,
        "requested_scope": requested_scope,
        "match_count": limit,
    }).execute()
    return list(response.data or [])

Kullanıcı ve SOP kapsamını vektör sıralamasından önce filtreleyen örnek SQL.

-- Optional reference schema; not deployed by this package.
-- Run only in a reviewed Supabase project with an ingestion plan.
create schema if not exists extensions;
create extension if not exists vector with schema extensions;

create table if not exists public.corporate_sop_embeddings (
    id bigint generated always as identity primary key,
    owner_id uuid not null,
    document_id text not null,
    scope text not null check (scope in ('finance', 'sales')),
    content text not null,
    embedding extensions.vector(384) not null
);

alter table public.corporate_sop_embeddings enable row level security;
revoke all on public.corporate_sop_embeddings from anon;
grant select on public.corporate_sop_embeddings to authenticated;

create policy "owner can read SOP embeddings"
on public.corporate_sop_embeddings
for select to authenticated
using (owner_id = (select auth.uid()));

create or replace function public.match_sop_documents(
    query_embedding extensions.vector(384),
    requested_scope text,
    match_count integer default 3
)
returns table (document_id text, content text, distance double precision)
language sql stable security invoker
set search_path = ''
as $$
    select d.document_id, d.content,
           (d.embedding <=> query_embedding)::double precision as distance
    from public.corporate_sop_embeddings as d
    where d.owner_id = (select auth.uid())
      and d.scope = requested_scope
    order by d.embedding <=> query_embedding
    limit least(greatest(match_count, 1), 10);
$$;

revoke all on function public.match_sop_documents(extensions.vector, text, integer) from public, anon;
grant execute on function public.match_sop_documents(extensions.vector, text, integer) to authenticated;

Devir, yetki reddi, belirsiz görev ve RAG parametreleri için testler.

import unittest
from types import SimpleNamespace

from main import DEMO_ROLES, TaskRequest, retrieve_sops, run
from supabase_rag import query_private_sop_knowledge


class PrototypeTests(unittest.TestCase):
    def test_finance_to_sop_handoff(self):
        result = run(TaskRequest(query="Bütçe sapmasını analiz et ve finans SOP uyumunu raporla",
                                 user_id="demo-finance", scopes=DEMO_ROLES["finance"]))
        self.assertIn("route:finance", result["audit_log"])
        self.assertIn("sop:read:finance", result["audit_log"])
        self.assertIn("%12.0", result["output_report"])
        self.assertIn("SOP-FIN-01", result["output_report"])
        self.assertIsNone(result["compliance_status"])
        self.assertTrue(result["review_required"])

    def test_sales_draft(self):
        result = run(TaskRequest(query="Satış teklifi hazırla", user_id="demo-sales",
                                 scopes=DEMO_ROLES["sales"]))
        self.assertIn("sales:draft-only", result["audit_log"])

    def test_finance_denied_for_sales_role(self):
        with self.assertRaises(PermissionError):
            run(TaskRequest(query="Bütçe sapması", user_id="demo-sales",
                            scopes=DEMO_ROLES["sales"]))

    def test_unknown_task_not_sent_to_sales(self):
        with self.assertRaises(ValueError):
            run(TaskRequest(query="Genel bilgi isteği", user_id="demo-sales",
                            scopes=DEMO_ROLES["sales"]))

    def test_sop_scope_checked_before_retrieval(self):
        with self.assertRaises(PermissionError):
            retrieve_sops("bütçe", DEMO_ROLES["sales"], "finance")

    def test_direct_sop_requires_named_segment(self):
        with self.assertRaises(ValueError):
            run(TaskRequest(query="SOP kurallarını denetle", user_id="demo-sop",
                            scopes=DEMO_ROLES["sop"]))

    def test_supabase_adapter_does_not_query_when_unauthorized(self):
        class FakeClient:
            called = False
            def rpc(self, *_args):
                self.called = True
                return SimpleNamespace(execute=lambda: SimpleNamespace(data=[]))
        client = FakeClient()
        with self.assertRaises(PermissionError):
            query_private_sop_knowledge("bütçe", scopes=DEMO_ROLES["sales"],
                                        requested_scope="finance", client=client,
                                        embed_locally=lambda _: [0.0] * 384)
        self.assertFalse(client.called)

    def test_supabase_adapter_passes_scope_to_rpc(self):
        class FakeClient:
            def rpc(self, name, params):
                self.name, self.params = name, params
                return SimpleNamespace(execute=lambda: SimpleNamespace(data=[{"document_id": "SOP-FIN-01"}]))
        client = FakeClient()
        rows = query_private_sop_knowledge("bütçe", scopes=DEMO_ROLES["finance"],
                                           requested_scope="finance", client=client,
                                           embed_locally=lambda _: [0.0] * 384)
        self.assertEqual(client.name, "match_sop_documents")
        self.assertEqual(client.params["requested_scope"], "finance")
        self.assertEqual(rows[0]["document_id"], "SOP-FIN-01")


if __name__ == "__main__":
    unittest.main()

Çalıştırma adımları ve üretime taşıma öncesi gereklilikler.

# Kurumsal Çoklu Ajan Prototipi

ProJection için **sentetik veriyle çalışan, salt okunur** Python/LangGraph örneği.
Orkestratör görevleri finans, SOP denetimi ve satış düğümlerine yönlendirir.
Finans sorusu SOP de içeriyorsa iki uzman arasında devir yapılır. Rapor, kanıt
kimliklerini ve insan incelemesi gereğini açıkça gösterir.

## Çalıştırma

Python 3.12 ile:

```sh
cd corporate_multi_agent
python3 -m venv .venv
. .venv/bin/activate
python -m pip install -r requirements.lock
python main.py
python -m unittest -v test_prototype.py
```

Örnekler:

```sh
python main.py "Satış teklifi hazırla" --role sales
python main.py "Finans SOP kurallarını denetle" --role sop --category sop_audit
```

Demo `--role` seçimi kimlik doğrulama değildir. Gerçek kullanıcı kimliği ve yetki
bilgisi güvenilir bir sunucu katmanından alınmalıdır.

## Dosyalar

- `main.py`: Çalıştırılabilir LangGraph akışı ve sentetik kayıtlar.
- `supabase_rag.py`: İsteğe bağlı, bağımlılıkları enjekte edilen Supabase RPC köprüsü.
- `schema.sql`: pgvector, RLS ve kapsam filtresi içeren örnek veritabanı şeması.
- `test_prototype.py`: Yönlendirme, devir, yetki reddi ve RAG parametre testleri.
- `requirements.in` / `requirements.lock`: Doğrudan ve çözümlenmiş bağımlılıklar.

## Entegrasyon sınırı

Varsayılan akış LLM, embedding API, ERP, CRM veya üretim veritabanı çağırmaz.
`supabase_rag.py` varsayılan grafiğe bağlı değildir. Bağlamak için yerel bir
384 boyutlu embedding işlevi ve **kullanıcı JWT'si ile çalışan** Supabase
istemcisi sağlanmalıdır. Son kullanıcı sorgularında service-role anahtarı
kullanılmamalıdır; bu anahtar RLS'yi aşabilir. SQL işlevi kapsam filtresini
ve kullanıcı filtresini vektör sıralamasından önce uygular. Uygulamaya alma
öncesinde tenant izolasyonu, kimlik doğrulama, kayıt saklama, insan onayı,
hata yönetimi ve güvenlik incelemesi ayrıca tasarlanmalıdır.

Kaynaklar: [LangGraph Graph API](https://docs.langchain.com/oss/python/langgraph/graph-api),
[Supabase semantic search](https://supabase.com/docs/guides/ai/semantic-search),
[Supabase pgvector](https://supabase.com/docs/guides/database/extensions/pgvector).

LangGraph, Pydantic ve isteğe bağlı Supabase paketleri.

langgraph==1.2.18
pydantic==2.14.1
supabase==2.33.0

Test edilen ortamın çözümlenmiş paket sürümleri.

annotated-types==0.8.0
anyio==4.15.1
certifi==2026.10.23
cffi==2.1.1
charset-normalizer==3.5.2
cryptography==50.0.2
deprecation==2.1.0
distro==1.9.0
h11==0.16.0
h2==4.4.1
hpack==4.2.0
httpcore==1.0.9
httpcore2==2.16.0
httpx==0.28.1
httpx2==2.16.0
hyperframe==6.1.0
idna==3.20
jsonpatch==1.38
jsonpointer==3.6
langchain-core==1.7.1
langchain-protocol==0.0.20
langgraph==1.2.18
langgraph-checkpoint==4.3.1
langgraph-prebuilt==1.1.1
langgraph-sdk==0.4.10
langsmith==0.14.14
multidict==7.2.1
orjson==3.13.0
ormsgpack==1.13.0
packaging==26.3
postgrest==2.33.0
propcache==0.5.4
pycparser==3.12
pydantic==2.14.1
pydantic_core==2.50.1
PyJWT==2.15.1
PyYAML==6.0.3
realtime==2.33.0
requests==2.34.2
requests-toolbelt==1.0.0
sniffio==1.3.1
storage3==2.33.0
StrEnum==0.4.15
supabase==2.33.0
supabase-auth==2.33.0
supabase-functions==2.33.0
tenacity==9.2.1
truststore==0.10.5
typing-inspection==0.4.4
typing_extensions==4.16.0
urllib3==2.8.0
uuid_utils==0.19.0
websockets==15.0.1
xxhash==4.0.2
yarl==1.25.1
zstandard==0.25.0