> 백엔드 개발 > 파이썬 튜토리얼 > 엔터프라이즈 에이전트 시스템 구축: 핵심 구성 요소 설계 및 최적화

엔터프라이즈 에이전트 시스템 구축: 핵심 구성 요소 설계 및 최적화

Patricia Arquette
풀어 주다: 2024-11-23 13:46:14
원래의
821명이 탐색했습니다.

Building Enterprise Agent Systems: Core Component Design and Optimization

소개

엔터프라이즈급 AI 에이전트를 구축하려면 구성요소 설계, 시스템 아키텍처, 엔지니어링 방식을 신중하게 고려해야 합니다. 이 문서에서는 강력하고 확장 가능한 에이전트 시스템을 구축하기 위한 주요 구성 요소와 모범 사례를 살펴봅니다.

1. 프롬프트 템플릿 엔지니어링

1.1 템플릿 디자인 패턴

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

from typing import Protocol, Dict

from jinja2 import Template

 

class PromptTemplate(Protocol):

    def render(self, **kwargs) -> str:

        pass

 

class JinjaPromptTemplate:

    def __init__(self, template_string: str):

        self.template = Template(template_string)

 

    def render(self, **kwargs) -> str:

        return self.template.render(**kwargs)

 

class PromptLibrary:

    def __init__(self):

        self.templates: Dict[str, PromptTemplate] = {}

 

    def register_template(self, name: str, template: PromptTemplate):

        self.templates[name] = template

 

    def get_template(self, name: str) -> PromptTemplate:

        return self.templates[name]

로그인 후 복사

1.2 버전 관리 및 테스트

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

class PromptVersion:

    def __init__(self, version: str, template: str, metadata: dict):

        self.version = version

        self.template = template

        self.metadata = metadata

        self.test_cases = []

 

    def add_test_case(self, inputs: dict, expected_output: str):

        self.test_cases.append((inputs, expected_output))

 

    def validate(self) -> bool:

        template = JinjaPromptTemplate(self.template)

        for inputs, expected in self.test_cases:

            result = template.render(**inputs)

            if not self._validate_output(result, expected):

                return False

        return True

로그인 후 복사

2. 계층적 메모리 시스템

2.1 메모리 아키텍처

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

from typing import Any, List

from datetime import datetime

 

class MemoryEntry:

    def __init__(self, content: Any, importance: float):

        self.content = content

        self.importance = importance

        self.timestamp = datetime.now()

        self.access_count = 0

 

class MemoryLayer:

    def __init__(self, capacity: int):

        self.capacity = capacity

        self.memories: List[MemoryEntry] = []

 

    def add(self, entry: MemoryEntry):

        if len(self.memories) >= self.capacity:

            self._evict()

        self.memories.append(entry)

 

    def _evict(self):

        # Implement memory eviction strategy

        self.memories.sort(key=lambda x: x.importance * x.access_count)

        self.memories.pop(0)

 

class HierarchicalMemory:

    def __init__(self):

        self.working_memory = MemoryLayer(capacity=5)

        self.short_term = MemoryLayer(capacity=50)

        self.long_term = MemoryLayer(capacity=1000)

 

    def store(self, content: Any, importance: float):

        entry = MemoryEntry(content, importance)

 

        if importance > 0.8:

            self.working_memory.add(entry)

        elif importance > 0.5:

            self.short_term.add(entry)

        else:

            self.long_term.add(entry)

로그인 후 복사

2.2 메모리 검색 및 인덱싱

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

from typing import List, Tuple

import numpy as np

from sklearn.metrics.pairwise import cosine_similarity

 

class MemoryIndex:

    def __init__(self, embedding_model):

        self.embedding_model = embedding_model

        self.embeddings = []

        self.memories = []

 

    def add(self, memory: MemoryEntry):

        embedding = self.embedding_model.embed(memory.content)

        self.embeddings.append(embedding)

        self.memories.append(memory)

 

    def search(self, query: str, k: int = 5) -> List[Tuple[MemoryEntry, float]]:

        query_embedding = self.embedding_model.embed(query)

        similarities = cosine_similarity(

            [query_embedding],

            self.embeddings

        )[0]

 

        top_k_indices = np.argsort(similarities)[-k:]

 

        return [

            (self.memories[i], similarities[i])

            for i in top_k_indices

        ]

로그인 후 복사

3. 관찰 가능한 추론 체인

3.1 사슬 구조

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

from typing import List, Optional

from dataclasses import dataclass

import uuid

 

@dataclass

class ThoughtNode:

    content: str

    confidence: float

    supporting_evidence: List[str]

 

class ReasoningChain:

    def __init__(self):

        self.chain_id = str(uuid.uuid4())

        self.nodes: List[ThoughtNode] = []

        self.metadata = {}

 

    def add_thought(self, thought: ThoughtNode):

        self.nodes.append(thought)

 

    def get_path(self) -> List[str]:

        return [node.content for node in self.nodes]

 

    def get_confidence(self) -> float:

        if not self.nodes:

            return 0.0

        return sum(n.confidence for n in self.nodes) / len(self.nodes)

로그인 후 복사

3.2 체인 모니터링 및 분석

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

import logging

from opentelemetry import trace

from prometheus_client import Histogram

 

reasoning_time = Histogram(

    'reasoning_chain_duration_seconds',

    'Time spent in reasoning chain'

)

 

class ChainMonitor:

    def __init__(self):

        self.tracer = trace.get_tracer(__name__)

 

    def monitor_chain(self, chain: ReasoningChain):

        with self.tracer.start_as_current_span("reasoning_chain") as span:

            span.set_attribute("chain_id", chain.chain_id)

 

            with reasoning_time.time():

                for node in chain.nodes:

                    with self.tracer.start_span("thought") as thought_span:

                        thought_span.set_attribute(

                            "confidence",

                            node.confidence

                        )

                        logging.info(

                            f"Thought: {node.content} "

                            f"(confidence: {node.confidence})"

                        )

로그인 후 복사

4. 구성 요소 분리 및 재사용

4.1 인터페이스 디자인

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

from abc import ABC, abstractmethod

from typing import Generic, TypeVar

 

T = TypeVar('T')

 

class Component(ABC, Generic[T]):

    @abstractmethod

    def process(self, input_data: T) -> T:

        pass

 

class Pipeline:

    def __init__(self):

        self.components: List[Component] = []

 

    def add_component(self, component: Component):

        self.components.append(component)

 

    def process(self, input_data: Any) -> Any:

        result = input_data

        for component in self.components:

            result = component.process(result)

        return result

로그인 후 복사

4.2 구성요소 레지스트리

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

class ComponentRegistry:

    _instance = None

 

    def __new__(cls):

        if cls._instance is None:

            cls._instance = super().__new__(cls)

            cls._instance.components = {}

        return cls._instance

 

    def register(self, name: str, component: Component):

        self.components[name] = component

 

    def get(self, name: str) -> Optional[Component]:

        return self.components.get(name)

 

    def create_pipeline(self, component_names: List[str]) -> Pipeline:

        pipeline = Pipeline()

        for name in component_names:

            component = self.get(name)

            if component:

                pipeline.add_component(component)

        return pipeline

로그인 후 복사

5. 성능 모니터링 및 최적화

5.1 성능 지표

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

from dataclasses import dataclass

from typing import Dict

import time

 

@dataclass

class PerformanceMetrics:

    latency: float

    memory_usage: float

    token_count: int

    success_rate: float

 

class PerformanceMonitor:

    def __init__(self):

        self.metrics: Dict[str, List[PerformanceMetrics]] = {}

 

    def record_operation(

        self,

        operation_name: str,

        metrics: PerformanceMetrics

    ):

        if operation_name not in self.metrics:

            self.metrics[operation_name] = []

        self.metrics[operation_name].append(metrics)

 

    def get_average_metrics(

        self,

        operation_name: str

    ) -> Optional[PerformanceMetrics]:

        if operation_name not in self.metrics:

            return None

 

        metrics_list = self.metrics[operation_name]

        return PerformanceMetrics(

            latency=sum(m.latency for m in metrics_list) / len(metrics_list),

            memory_usage=sum(m.memory_usage for m in metrics_list) / len(metrics_list),

            token_count=sum(m.token_count for m in metrics_list) / len(metrics_list),

            success_rate=sum(m.success_rate for m in metrics_list) / len(metrics_list)

        )

로그인 후 복사

5.2 최적화 전략

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

class PerformanceOptimizer:

    def __init__(self, monitor: PerformanceMonitor):

        self.monitor = monitor

        self.thresholds = {

            'latency': 1.0,  # seconds

            'memory_usage': 512,  # MB

            'token_count': 1000,

            'success_rate': 0.95

        }

 

    def analyze_performance(self, operation_name: str) -> List[str]:

        metrics = self.monitor.get_average_metrics(operation_name)

        if not metrics:

            return []

 

        recommendations = []

 

        if metrics.latency > self.thresholds['latency']:

            recommendations.append(

                "Consider implementing caching or parallel processing"

            )

 

        if metrics.memory_usage > self.thresholds['memory_usage']:

            recommendations.append(

                "Optimize memory usage through batch processing"

            )

 

        if metrics.token_count > self.thresholds['token_count']:

            recommendations.append(

                "Implement prompt optimization to reduce token usage"

            )

 

        if metrics.success_rate < self.thresholds['success_rate']:

            recommendations.append(

                "Review error handling and implement retry mechanisms"

            )

 

        return recommendations

로그인 후 복사

결론

엔터프라이즈급 에이전트 시스템을 구축하려면 다음 사항에 세심한 주의가 필요합니다.

  • 구조화된 프롬프트 관리 및 버전 관리
  • 효율적이고 확장 가능한 메모리 시스템
  • 관찰 가능하고 추적 가능한 추론 프로세스
  • 재사용 가능한 모듈식 구성 요소 디자인
  • 종합적인 성능 모니터링 및 최적화

위 내용은 엔터프라이즈 에이전트 시스템 구축: 핵심 구성 요소 설계 및 최적화의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!

원천:dev.to
본 웹사이트의 성명
본 글의 내용은 네티즌들의 자발적인 기여로 작성되었으며, 저작권은 원저작자에게 있습니다. 본 사이트는 이에 상응하는 법적 책임을 지지 않습니다. 표절이나 침해가 의심되는 콘텐츠를 발견한 경우 admin@php.cn으로 문의하세요.
저자별 최신 기사
인기 튜토리얼
더>
최신 다운로드
더>
웹 효과
웹사이트 소스 코드
웹사이트 자료
프론트엔드 템플릿