agent_common

📦 PyPI 패키지 · 💻 GitHub 소스 및 매뉴얼

🇰🇷 한국어 설명 | 🇺🇸 English Description | 📖 매뉴얼 (Manuals)

🇰🇷 agent_common 패키지 (한국어)

중앙 에이전트 및 데이터 이관/생성 서비스를 위한 공통 로깅, 설정 로더, 인프라 클라이언트, 동적 도구(Tool) 파서 및 에러 처리 라이브러리 패키지입니다.

📌 주요 제공 기능

1. 설정 로더 및 불변 설정 객체 (agent_common.config_loader)

• 1.1. 계층적 YAML 설정 해석 및 병합 (Deep Merge): 패키지 기본 설정(agent_common/config/*.yml)과 개별 프로젝트 설정(config/*.yml) 동적 병합.
• 1.2. 불변 점 표기법 조회 (ReadOnlyConfig): config.ecs.endpoint_url, config.transfer.max_workers_int 형태로 직관적 속성 접근 및 런타임 변조 방지.
• 1.3. 타입 접미사 자동 형 변환 및 타입 보증 (Type Guarantee & Coercion - v0.4.14 / 범용화 v0.4.32):
  • _int: int 정수형 자동 형 변환 및 보증
  • _float: float 실수형 자동 형 변환 및 보증
  • _bool: bool 불리언형 자동 변환 및 타입 보증 (파이썬 bool 또는 대소문자 무관 "true", "false" 지원, 0, 1 등 비지원 값 유입 시 Fail-Fast 차단)
  • _str: str 문자열 변환 및 .strip() 공백 자동 정제
  • _list / _dict: 리스트 / 불변 딕셔너리(ReadOnlyConfig) 래핑 보증
  • 범용 함수 coerce_type_by_key_suffix 및 중첩 딕셔너리 일괄 변환 coerce_dict_by_key_suffix 제공으로 임의의 외부 설정 파일(rule.yml, mapping.yml 등) 및 데이터 파이프라인 완벽 지원
  • 타입 불일치 시 침묵하지 않고 상세 안내와 함께 즉시 조기 실패(Fail-Fast, ValueError/TypeError) 발생 보증
• 1.4. Fail-Fast 필수 설정 검증 (require_setting()): 프로그램 시작 시 필수 설정값 누락 시 상세 원인 출력 후 프로세스 즉시 종료.
• 1.5. 네트워크 프록시 제어 (_apply_no_proxy): proxy.no_proxy 설정의 NO_PROXY 환경변수 자동 반영.
• 1.6. 모든 상수의 설정 파일화 및 템플릿 보정 (ensure_config_file()): 코드 내 모든 상수의 설정 파일화(외부화), config.yml 자동 생성 및 누락 상수 강제 주입·보정.
• 1.7. 환경변수 템플릿 치환 (_interpolate_env_vars): YAML 파일 및 설정 딕셔너리 내 ${VAR_NAME:-default} 구문 자동 치환을 통한 선언적 환경변수 바인딩 지원 (v0.4.55).

2. 단일 행 로깅 포매터 및 로거 (agent_common.logger)

• 2.1. 단일 행 평탄화 포매터 및 예외 원천 추적 (SingleLineFlattenFormatter): 모든 로그 및 Traceback 예외 메시지를 1줄로 평탄화 및 [Origin: ...] 원천 위치 추출, 호출 스택 기반 호출자/클래스명(%(caller)s, %(className)s) 자동 분리 추출 및 프로그램 로거 이름(%(name)s) 통일 지원 (v0.4.36)
• 2.2. 로깅 환경 일괄 구성 및 핸들러 제어 (ProjectLogger.configure): 콘솔 및 파일 로그 핸들러 동적 생성, 일자별 폴더 분리, 실행 로그 레벨별 디렉터리 자동 분기({log_level} 기반 log_file 단일화) 및 서드파티 노이즈 억제
• 2.3. 다국어 로그 메시지 템플릿 사전 및 코드 기반 로깅 (logging_messages_*.yml): config.yml의 logging.language (KO 또는 EN) 설정에 따라 한국어/영문 메시지 사전 자동 연동, 런타임 동적 언어 전환 및 안전한 템플릿 치환, default_str 표준 매개변수 기반 기본 템플릿 치환 보증 (v0.4.73)
• 2.4. 작업 진행 통계 및 예외/제외 사유별 실시간 집계 (record_result): 성공, 실패, 제외(Skip) 3단계 상태 분류 및 인스턴스/클래스 전역 멀티스레드 에러 집계
• 2.5. 작업 결과 요약 리포트 자동 생성 (log_summary): '전체 = 성공 + 실패 + 제외' 정합성 보장, TableFormatter 기반 세로줄 자동 맞춤, 소요 시간, 처리 속도, 전송량이 포함된 표준 마크다운 표(Table) 자동 출력 (v0.4.63)

3. 스토리지 및 데이터베이스 클라이언트 (agent_common.clients)

• 3.1. AWS S3 및 Dell ECS 오브젝트 스토리지 클라이언트 (S3Client): AWS S3 및 Dell ECS(S3 호환) 저장소 접속, 초기화 즉시 head_bucket Fail-Fast 검증, 대용량 페이징 제너레이터(list_objects), 메타데이터 빠른 조회(get_object_size), 메모리 스트리밍 획득(get_object_stream), GCS 실시간 파일 전송 및 동일 파일 스마트 스킵(transfer_to_gcs - "UPLOADED", "SKIPPED", "FAILED" 반환 및 구간별 정밀 레이턴시 로깅 지원 - v0.4.72)
• 3.2. Google Cloud Storage 스트리밍 클라이언트 및 멀티 계층 인증 (GcsClient): 4단계 GCP 인증 우선순위(GOOGLE_APPLICATION_CREDENTIALS_JSON 인메모리 JSON -> GOOGLE_APPLICATION_CREDENTIALS 파일 -> credentials_path_str -> Google ADC) 지원, 연결 및 버킷 권한 조기 검증, 블롭 메타데이터 및 크기 조회(get_blob_size), 메모리 낭비 없는 청크 단위 스트림 직접 업로드(upload_stream - v0.4.56)
• 3.3. BigQuery 배치 및 스트리밍 적재 클라이언트 (BigQueryClient): 연결 및 테이블 스키마 사전 캐싱(get_table), JSON 배치 로드 Job(load_table_from_json_data) 및 중첩 에러(errors, location, reason) 상세 분해, 실시간 스트리밍 인서트(insert_rows_json_data), 범용 동기 SQL 쿼리(query), 중복 전송 방지용 기존 키 집합 추출(get_existing_keys), WHERE 조건식 기반 안전 일괄 삭제 DML(delete_rows - v0.4.80)
• 3.4. BigQuery 고성능 인라인 MERGE (Upsert) 쿼리 엔진 (merge_table_from_json_data): 스테이징 임시 테이블 생성 없이 직접 UNNEST(JSON_QUERY_ARRAY(@json_payload)) 기반 인라인 MERGE INTO 수행, 기본키(PK) 기준 자동 UPDATE/INSERT 분기, 생성일시 등 최초 값 보존(preserve_columns_list), 컬럼 데이터 타입 자동 추론 및 명시적 캐스팅(column_types_dict), 한글/특수문자/예약어 백틱(`) 완벽 보호, HTTP 413 페이로드 초과 방지 기본 100건 청크 자동 분할, 후속 연쇄 쿼리(post_queries_list) 지원
• 3.5. BigQuery 타임존 오프셋 변환 및 테이블 타임존 모드 검증·동기화 (convert_to_bigquery_timestamp): ISO 8601, 공백 구분, 14자리/8자리 숫자 등 다양한 원천 날짜 문자열의 BigQuery 표준 타임스탬프 정규화, 타임존 오프셋 우선순위(timezone_offset_str - v0.4.71), 한국 시각 숫자 보존 모드(kst_as_utc_timestamp_bool), 테이블 메타데이터(라벨 timestamp_mode, 테이블 및 컬럼 Description) 자동 동기화 및 기존 데이터 존재 시 불일치 차단(Fail-Fast)

4. 동적 도구 로더 및 템플릿 평가기 (agent_common.tool_parser) & 내장 도구 (agent_common.tool)

• 4.1. 이원화된 Tool 디렉터리 계층 탐색 및 동적 로딩 (ToolParser.load_tool_function):
  • 1순위 (내장 도구): agent_common/tool/ 하위 모듈 (전사 표준 내장 도구)
  • 2순위 (프로젝트 도구): config.yml의 transfer.tool_dir_str에 지정된 로컬 경로 (예: medallion/tool/)
  • 모듈 함수, 클래스.메서드, 클래스 내부 메서드 3단계 탐색, _tool_cache 인메모리 캐싱 및 scan_rules_for_tool_functions 사전 검증 지원
• 4.2. 선언적 템플릿 치환 및 표현식 평가 (ToolParser.eval):
  • 변수 네임스페이스 바인딩: {ecs.key}, {sys.today}, {json.title}
  • 동적 도구 함수 호출: "{code.date_check_to_code(contentInfo.enddate)}", "{path.get_json_name(ecs.key)}"
  • 파이프(|) 우선순위 폴백 및 기본값: "{meta.title|json.title|'기본제목'}"
  • inspect.signature 기반 유연한 파라미터 매핑 및 ctx 컨텍스트 자동 주입
• 4.3. 안전한 네임스페이스 탐색 (_SafeNamespace):
  • 점(.) 및 인덱스([]) 접근 통합, 대소문자 무관(Case-insensitive) 유연한 탐색
  • 누락된 필드에 대해 KeyError 없이 안전하게 빈 문자열("") 반환 및 중첩 딕셔너리/리스트 재귀적 래핑
• 4.4. 내장 공통 일시 도구 (DateTimeUtils):
  • 테이블 규칙 및 템플릿 평가 전용 날짜/시간 도구, 코어 타임존 해석은 TimeUtils에 위임하여 역할 분담 (v0.4.69 / v0.4.74)
  • parse_datetime(dt_input_any, default_tz_obj): 다양한 형식(datetime, ISO 문자열)의 일시를 timezone-aware datetime으로 정규화 변환 (v0.4.74)
  • get_now_timestamp(tz_obj): ISO 8601 표준 타임스탬프(YYYY-MM-DD HH:MM:SS+09:00 또는 +00:00) 반환 (BigQuery 적재 및 규칙 파일 표준)
  • get_now_no_tz(tz_obj): 타임존 없는 일시 문자열(YYYY-MM-DD HH:MM:SS) 반환 (로그 및 요약표 전용)
  • get_now_compact(tz_obj): YYYYMMDDHHMMSS 형식 14자리 압축 일시 반환 (예: 20260824110500)
  • get_today_yyyymmdd(tz_obj): YYYYMMDD 형식 8자리 일자 반환 (새벽 배치 날짜 역전 방지, 예: 20260824)

5. 진행률 트래커 및 공용 유틸리티 (agent_common.utils)

• 5.1. 호스트 시스템 타임존 감지, 전 세계 표준시 해석 및 일시 정규화 (TimeUtils): 호스트 시스템(배치 KST vs 파드 UTC) 타임존 동적 자동 감지, 전 세계 30여 개 주요 표준 타임존(UTC, KST, JST, EST, CET 등) 해석, ISO 8601 오프셋 계산, 일시 객체/문자열을 timezone-aware datetime으로 정규화하는 parse_datetime 지원 코어 시간 인프라 유틸리티 (v0.4.69 / v0.4.74)
• 5.2. 멀티스레드 실시간 진행률 추적 및 마일스톤 경고 (ProgressTracker): 멀티스레드 실시간 진행률 추적([N/Total] (P%)), 처리 속도(건/s, MB/s) 및 남은 시간(ETA) 예측, 일반 진행 INFO vs 10% 단위 마일스톤 WARNING 승격 로깅
• 5.3. 유니코드 전각 문자 폭 계산 및 마크다운/콘솔 테이블 칼맞춤 포매터 (TableFormatter): 유니코드 동아시아 문자 폭(unicodedata.east_asian_width) 정밀 계산 기반 한글/한자(2칸) vs 영문(1칸) 모노스페이스 콘솔 및 마크다운 테이블 세로줄 자동 맞춤 포매터 (v0.4.63)

7. 통합 LLM 클라이언트 및 추론 엔진 (agent_common.llm)

• 7.1. 모델 프로필 관리 및 텍스트 생성
• 7.2. 외부 채팅 API 및 Fabrix 연동
• 7.3. 로컬 GGUF 추론 및 모델 캐싱
• 7.4. 실행 모드 및 조건부 로컬 전환
• 7.5. 추론 결과 및 예외 처리
• 7.6. Groq 감독관 AI 및 Antigravity Stop 훅

🛠️ 사용 예시 (Usage Examples)

AI에게 프로그램 개발을 요청하는 방법

AI에게 만들려는 기능, 아래 PyPI 및 GitHub 링크, 이 README의 「주요 제공 기능」에서 참고할 기능 번호를 함께 전달하면 됩니다. “README의 몇 번, 몇 번 기능과 연결된 상세 매뉴얼을 참고해서 agent_common을 활용한 프로그램을 만들어줘”라고 요청하세요.

• PyPI: agent-common 패키지
• GitHub: agent_common 저장소 및 README

예를 들어, Dell ECS S3 스토리지에서 GCS로 파일을 복사하는 프로그램을 만들려면 3.1 (S3Client), 3.2 (GcsClient) 기능을 지정해 다음과 같이 요청하면 됩니다.

Dell ECS S3 스토리지에서 Google Cloud Storage(GCS)로 파일을 복사하는 Python 프로그램을 만들어줘.

아래 agent-common 패키지와 GitHub 문서를 참고해줘.
- https://pypi.org/project/agent-common/
- https://github.com/kampores/agent_common

주로 README의 「주요 제공 기능」 중 3.1(AWS S3 및 Dell ECS 오브젝트 스토리지 클라이언트)과
3.2(Google Cloud Storage 스트리밍 클라이언트 및 멀티 계층 인증),
그리고 각 항목에 연결된 상세 매뉴얼을 참고해서 agent_common을 활용해 구현해줘.

구현 전에 해당 기능의 상세 매뉴얼과 실제 API 사용법을 확인하고,
요구사항에서 빠진 정보는 임의로 가정하지 말고 질문해줘.
처음부터 agent_common의 모든 기능을 적용하려고 하지 말고,
먼저 파일 복사에 필요한 최소 기능을 구현하고 동작을 확인해줘.
그다음 필요한 부가기능을 변수형 보증(1.3), config 점 표기법(1.2) 등으로
나누어 하나씩 추가하고, 추가할 때마다 기존 파일 복사가 정상 동작하는지 확인해줘.

다른 프로그램도 같은 방식으로 만들려는 기능에 해당하는 번호를 골라 요청하면 됩니다. 복사 대상 버킷, 대상 파일 범위, 저장 경로 규칙 등 구체적인 요구사항을 함께 설명하면 원하는 동작을 더 명확하게 전달할 수 있습니다.

최소 기능부터 단계적으로 확장하기

처음부터 agent_common의 모든 기능을 한꺼번에 적용하도록 요청하지 말고, 핵심 기능 구현 → 동작 확인 → 필요한 부가기능 추가 순서로 진행하세요. 패키지가 이미 제공하는 기능은 재사용하고, 현재 목적에 필요한 기능부터 적용하면 됩니다.

1. 최소 기능 구현: ECS S3 → GCS 복사 프로그램이라면 접속·인증과 지정한 파일의 복사에 필요한 기능부터 구현합니다.
2. 핵심 동작 확인: 지정한 파일이 원하는 GCS 경로에 정상적으로 복사되는지 확인합니다.
3. 부가기능 단계적 추가: 필요에 따라 변수형 보증(1.3), config 점 표기법(1.2), 진행률 추적(5.2), 결과 요약(2.5) 등을 하나씩 적용하고 기존 동작을 확인합니다.

후속 요청도 “현재 파일 복사 동작을 유지하면서 README 1.3의 변수형 보증을 적용해줘”처럼 범위를 좁혀 전달하면 됩니다.

1. 전역 config 점 표기법 및 타입 보증 활용

from agent_common.config_loader import config

# 1) 타입 접미사에 따른 자동 형 변환 보증
max_workers: int = config.transfer.max_workers_int       # int 타입 보증
host: str = config.database.host_str                     # str 타입 및 .strip() 정제 보증
is_active: bool = config.transfer.is_active_bool         # bool 타입 보증

# 2) 계층적 속성 접근
api_url: str = config.services.api_endpoint_url
db_port: int = config.database.port_int

2. ToolParser를 통한 동적 룰 평가

from agent_common.tool_parser import ToolParser

# ToolParser 인스턴스 생성 (설정 파일 기반으로 내장/로컬 Tool 자동 탐색)
tool_parser = ToolParser()

# 컨텍스트 데이터 준비
context_dict = {
    "storage": {"key": "/data/incoming/20260804/sample_document.json"},
    "document": {"expiry_date": "2024-12-31"},
    "sys": tool_parser.build_sys_context(),
}

# 1) 도구 함수 호출 템플릿 평가
date_val = tool_parser.eval("{date.get_today_yyyymmdd()}", context_dict)
# -> "20260824"

# 2) 네임스페이스 및 내장 일시 템플릿 평가
today_val = tool_parser.eval("{sys.today}", context_dict)
# -> "20260824"

3. ProgressTracker 실시간 진행률 추적

from agent_common.utils import ProgressTracker
from agent_common.logger import ProjectLogger

logger = ProjectLogger("MyTask")
tracker = ProgressTracker(total_items_int=1000, logger_obj=logger, item_name_str="파일")

for file_info in file_list:
    try:
        # 처리 로직 수행
        tracker.increment_success(bytes_int=len(data))
    except Exception as e:
        tracker.increment_failure(error_msg_str=str(e))

# 최종 결과 요약 리포트 출력
tracker.log_summary()

4. LlmClient를 통한 통합 텍스트/SQL 생성

from agent_common.llm import LlmClient

# 1) 설정 풀에 정의된 모델명 또는 용도로 클라이언트 초기화
llm_client = LlmClient(purpose_str="sql_generator")

# 2) 선택한 프로필의 provider_str에 따라 텍스트 생성
prompt_str = "사용자 요청: 2026년 8월 일일 가입자 수 통계 쿼리를 작성해줘."
response_str = llm_client.generate(
    prompt_str=prompt_str,
    system_prompt_str="당신은 BigQuery 전문 SQL 생성 AI입니다."
)

print(f"생성된 결과 ({llm_client.last_generated_by_str}):\n{response_str}")

5. 스토리지 및 BigQuery 클라이언트 활용

S3/ECS 파일 순회 및 GCS 실시간 파이프라인 전송 (동일 파일 스마트 스킵)

from agent_common.clients import S3Client, GcsClient

# S3/ECS 및 GCS 클라이언트 초기화 (버킷 접근 권한 Fail-Fast 자동 검증)
s3_client = S3Client(
    endpoint_url_str="https://ecs.mycorp.internal:9021",
    access_key_str="MY_ACCESS_KEY",
    secret_key_str="MY_SECRET_KEY",
    bucket_name_str="source-lake",
)
gcs_client = GcsClient(bucket_name_str="target-lake")

# 대용량 객체 제너레이터 순회 및 실시간 스트리밍 전송
for obj_dict in s3_client.list_objects(prefix_str="raw/events/20260824/"):
    s3_key_str: str = obj_dict["Key"]
    size_int: int = obj_dict["Size"]
    gcs_blob_str: str = f"lake/{s3_key_str.lstrip('/')}"
    
    # 동일 파일 존재 시 스킵("SKIPPED"), 신규 전송 시 "UPLOADED"
    status_str = s3_client.transfer_to_gcs(
        gcs_client_obj=gcs_client,
        s3_key_str=s3_key_str,
        gcs_blob_name_str=gcs_blob_str,
        size_int=size_int,
    )
    print(f"[{status_str}] {s3_key_str} -> {gcs_blob_str} ({size_int:,} bytes)")

BigQuery 고성능 인라인 MERGE (Upsert) 실행

from agent_common.clients import BigQueryClient

bq_client = BigQueryClient(
    project_id_str="my-gcp-project",
    dataset_id_str="analytics_dw",
    table_id_str="tb_member_profile",
)

members_list = [
    {
        "member_id": "M001",
        "name": "홍길동",
        "login_count": 42,
        "is_vip": True,
        "created_at": "2026-01-01 00:00:00+09:00",
        "last_login_at": "2026-08-24 15:30:00+09:00",
    }
]

# 임시 테이블 없이 단일 SQL 쿼리로 직접 MERGE 수행 (created_at 최초값 보존)
bq_client.merge_table_from_json_data(
    json_data_any=members_list,
    pk_key_str="member_id",
    preserve_columns_list=["created_at"],
    chunk_size_int=100,
)

🚀 설치 및 빌드 방법

📦 Wheel 패키지 빌드 (.whl 생성)

새로운 버전으로 패키징하여 .whl 파일을 빌드할 경우 scripts/build_agent_common_whl.py 또는 agent_common 디렉터리 내에서 아래 명령을 실행합니다.

1. 사내 폐쇄망 환경 (인터넷 차단, 완전히 오프라인 빌드)

외부 PyPI 접속을 완전히 차단하기 위해 --no-index, --no-build-isolation, --no-deps 옵션을 지정합니다.

# 루트 디렉터리에서 자동 빌드 스크립트 실행 (권장)
python scripts/build_agent_common_whl.py

# 또는 pip wheel 직접 실행
pip wheel ./agent_common --no-index --no-build-isolation --no-deps -w whls/

2. 인터넷 연동망 환경 (온라인 빌드)

# pip wheel 이용
pip wheel ./agent_common --no-deps -w whls/

# 또는 build 모듈 이용
python -m build agent_common --wheel -o whls/

Wheel 패키지 설치

# 개발 환경 (Editable 모드 - 기본 경량 설치)
pip install -e agent_common

# 개발 환경 (클라우드 클라이언트 extras 포함)
pip install -e "agent_common[clients]"

# 배포 환경 (Wheel 패키지 설치)
pip install dist/agent_common-0.4.78-py3-none-any.whl

🌐 PyPI 공식 배포 (관리자 전용)

# 1. 빌드 도구 최신화
pip install build twine

# 2. 패키지 빌드 (sdist 및 wheel 동시 생성)
python -m build

# 3. 배포 아카이브 검증
python -m twine check dist/*

# 4. PyPI 업로드
python -m twine upload dist/agent_common-0.4.78*

📖 상세 기능 매뉴얼 (User Manuals)

모듈별 상세 아키텍처 및 실전 코드 예시는 아래 상세 매뉴얼 문서를 참고하세요:

📋 버전 변경 이력 (Changelog)

v0.4.78 (2026-09-21)

• AI에게 요구사항·패키지 링크·기능 번호를 전달하는 방법과 최소 기능부터 단계적으로 확장하는 가이드를 추가했습니다.
• 통합 LLM 클라이언트의 한국어·영어 매뉴얼을 추가하고, 모델 설정·외부 API·로컬 GGUF·환경변수·예외 처리를 설명했습니다.
• Groq 감독관 AI와 Antigravity Stop 훅의 AGENTS.md 검사 사례 및 실제 검사 범위·생략 조건을 정리했습니다.
• LLM auto 모드의 조건부 전환 설명을 실제 구현에 맞추고, 개발 중인 에러 핸들러 소개는 제외했습니다.
• 패키지 버전을 0.4.78로 올렸습니다. 런타임 동작 변경은 없습니다.

자세한 버전 변경 이력은 GitHub CHANGELOG.md 파일을 참고하세요.

🇺🇸 agent_common Package (English)

A comprehensive Python common library providing unified logging, hierarchical configuration loaders, cloud and database infrastructure clients, dynamic tool parsers, and centralized error handling for enterprise agent services and data migration pipelines.

📌 Key Features

1. Configuration Loader & Immutable Config Object (agent_common.config_loader)

• 1.1. Hierarchical YAML Parsing & Deep Merge: Dynamically merges base package configurations (agent_common/config/*.yml) with project-specific configurations (config/*.yml).
• 1.2. Immutable Dot-Notation Access (ReadOnlyConfig): Intuitive attribute-based lookup (config.ecs.endpoint_url, config.transfer.max_workers_int) while preventing unintended runtime mutations.
• 1.3. Type Guarantee & Automatic Coercion via Type Suffixes (v0.4.14 / Generalized v0.4.32):
  • _int: Automatic integer conversion and type guarantee.
  • _float: Automatic floating-point conversion and type guarantee.
  • _bool: Strict boolean conversion and type guarantee (Python bool or case-insensitive "true", "false"; unsupported values like 0, 1 fail fast).
  • _str: Automatic string conversion and .strip() whitespace trimming.
  • _list / _dict: Guaranteed list / immutable dictionary (ReadOnlyConfig) wrapping.
  • Standalone functions coerce_type_by_key_suffix and recursive batch coercion coerce_dict_by_key_suffix for arbitrary external configuration files (rule.yml, mapping.yml) and data mappings.
  • Strict Fail-Fast guarantee: type-suffix mismatches immediately raise diagnostic exceptions (ValueError/TypeError) instead of silently falling back to raw values.
• 1.4. Fail-Fast Required Setting Validation (require_setting()): Immediate process termination with diagnostic output if required settings are missing during startup.
• 1.5. Network Proxy Control (_apply_no_proxy): Automatic synchronization of NO_PROXY environment variable from proxy.no_proxy configuration.
• 1.6. Externalizing All Constants & Self-Healing Templates (ensure_config_file()): Materializing all in-code constants to configuration files, automatic scaffolding, and in-place missing key injection.

2. Single-Line Log Formatter & Project Logger (agent_common.logger)

• 2.1. Single-Line Flatten Formatter & Origin Tracking (SingleLineFlattenFormatter): Flattens log records, extracts [Origin: ...] caller frames, and optimizes for centralized log aggregators (Logstash, Fluentd, CloudWatch).
• 2.2. Batch Logging Configuration & Handler Control (ProjectLogger.configure): Dynamic console/file handler initialization, date-based directories, level-based directory creation (log_file with {log_level}), and third-party noise suppression.
• 2.3. Multilingual Message Catalog & Code-Based Logging (logging_messages_*.yml): Dynamic bilingual dictionary loading (KO/EN), runtime language switching, and safe template parameter substitution.
• 2.4. Real-Time Metric Tracking & Error/Exclusion Classification (record_result): Three-tier outcome model (Success, Failure, Excluded/Skip) and dual instance/class-global multithreaded telemetry.
• 2.5. Automatic Summary Report Generation (log_summary): Emits structured 80-column execution summary reports with duration, throughput (items/s), transfer rate (MB/s), and decoded error diagnostics.

3. Storage and Database Infrastructure Clients (agent_common.clients)

• 3.1. AWS S3 & Dell ECS Object Storage Client (S3Client): Connects to AWS S3 and Dell ECS (S3-compatible) storage, early head_bucket Fail-Fast verification, high-throughput paginated iterator (list_objects), fast header metadata lookup (get_object_size), in-memory streaming body extraction (get_object_stream), real-time streaming pipeline upload to GCS with smart duplicate skipping (transfer_to_gcs - returning "UPLOADED", "SKIPPED", or "FAILED" with sub-stage latency telemetry - v0.4.72).
• 3.2. Google Cloud Storage Streaming Client & Multi-Tier Auth (GcsClient): Four-tier GCP credential resolution hierarchy (GOOGLE_APPLICATION_CREDENTIALS_JSON in-memory JSON -> GOOGLE_APPLICATION_CREDENTIALS file -> credentials_path_str -> Google ADC), instant bucket reachability validation, blob metadata lookup (get_blob_size), zero-disk chunked streaming uploads (upload_stream - v0.4.56).
• 3.3. BigQuery Batch Loading & Streaming Ingestion (BigQueryClient): Fail-fast schema caching (get_table), JSON batch load jobs (load_table_from_json_data) with unpacked nested error diagnostics (errors, location, reason), real-time streaming ingestion (insert_rows_json_data), general SQL query execution (query), unique key deduplication set lookup (get_existing_keys).
• 3.4. BigQuery High-Performance Inline MERGE (Upsert) Engine (merge_table_from_json_data): Executes direct inline MERGE INTO via UNNEST(JSON_QUERY_ARRAY(@json_payload)) without temporary staging tables, automatic primary key routing, original column preservation (preserve_columns_list), schema type inference and casting (column_types_dict), reserved keyword and Unicode column backtick escaping, HTTP 413 payload limit protection via default 100-row chunking, post-processing query execution (post_queries_list).
• 3.5. BigQuery Timestamp Conversion & Timezone Mode Synchronization (convert_to_bigquery_timestamp): Normalizes ISO 8601, whitespace, and 14-digit/8-digit timestamps to standard BigQuery formats, timezone offset precedence (timezone_offset_str - v0.4.71), display convenience mode (kst_as_utc_timestamp_bool), automated table metadata synchronization (labels, table/column descriptions) with fail-fast mismatch blocking.

4. Dynamic Tool Loader & Template Evaluator (agent_common.tool_parser) & Built-in Tools (agent_common.tool)

• 4.1. Dual Tool Hierarchy Discovery & Dynamic Loading (ToolParser.load_tool_function):
  • Priority 1 (Built-in Tools): Modules under agent_common/tool/ (standard enterprise tools).
  • Priority 2 (Project Tools): Local path configured in config.yml under transfer.tool_dir_str (e.g., medallion/tool/).
  • 3-step function introspection, _tool_cache in-memory caching, and scan_rules_for_tool_functions pre-flight validation.
• 4.2. Declarative Template Evaluation & Expression Resolution (ToolParser.eval):
  • Variable namespace binding: {ecs.key}, {sys.today}, {json.title}.
  • Dynamic tool function invocation: "{code.date_check_to_code(contentInfo.enddate)}", "{path.get_json_name(ecs.key)}".
  • Pipe (|) fallback chains and default values: "{meta.title|json.title|'Default Title'}".
  • inspect.signature-based parameter mapping and automatic ctx context injection.
• 4.3. Safe Namespace Lookup & Case-Insensitive Access (_SafeNamespace):
  • Unified dot-notation and bracket indexing, resilient case-insensitive key resolution.
  • Returns empty string ("") on missing keys without raising KeyError; recursive wrapping of nested dictionaries and lists.
• 4.4. Built-in DateTime Tool (DateTimeUtils):
  • Business rule and template formatting tool delegating core timezone calculations to TimeUtils (v0.4.69 / v0.4.74).
  • parse_datetime(dt_input_any, default_tz_obj): Normalizes various datetime formats and ISO strings into timezone-aware datetimes (v0.4.74).
  • get_now_timestamp(tz_obj): Returns standard ISO 8601 timestamp string (YYYY-MM-DD HH:MM:SS+09:00 or +00:00).
  • get_now_no_tz(tz_obj): Returns clean datetime string without timezone (YYYY-MM-DD HH:MM:SS).
  • get_now_compact(tz_obj): Returns 14-digit compact timestamp string (YYYYMMDDHHMMSS).
  • get_today_yyyymmdd(tz_obj): Returns 8-digit date string (YYYYMMDD), preventing overnight batch date rollback.

5. Progress Tracker & Common Utilities (agent_common.utils)

• 5.1. System Timezone Detection, Global Timezone Resolution & Datetime Normalization (TimeUtils): Dynamic host OS system timezone detection, 30+ world timezone parsing, ISO 8601 offset calculation, and parse_datetime normalization core utility (v0.4.69 / v0.4.74).
• 5.2. Multithreaded Progress Tracking & Milestone Telemetry (ProgressTracker): Real-time multithreaded progress tracking ([N/Total] (P%)), throughput/ETA calculation, and tiered logging (standard INFO vs 10% milestone WARNING level elevation).
• 5.3. Unicode East Asian Width Alignment & Table Formatter (TableFormatter): Precision terminal and Markdown table column width alignment utility calculating Unicode East Asian character display widths (unicodedata.east_asian_width) (v0.4.63).

7. Unified LLM Client & Inference Engine (agent_common.llm)

• 7.1. Model Profiles and Text Generation
• 7.2. External Chat APIs and Fabrix
• 7.3. Local GGUF Inference and Model Caching
• 7.4. Provider Selection and Conditional Local Fallback
• 7.5. Inference Results and Error Handling
• 7.6. Groq Supervisor and Antigravity Stop Hook

🛠️ Usage Examples

1. Dot-Notation Global config & Type Guarantees

from agent_common.config_loader import config

# 1) Guaranteed type coercion via type suffixes
max_workers: int = config.transfer.max_workers_int       # Guaranteed int
host: str = config.database.host_str                     # Guaranteed str with .strip()
is_active: bool = config.transfer.is_active_bool         # Guaranteed bool

# 2) Hierarchical attribute access
api_url: str = config.services.api_endpoint_url
db_port: int = config.database.port_int

2. Dynamic Rule Evaluation via ToolParser

from agent_common.tool_parser import ToolParser

# Initialize ToolParser (auto-discovers built-in and project tools)
tool_parser = ToolParser()

# Prepare context dictionary
context_dict = {
    "storage": {"key": "/data/incoming/20260804/sample_document.json"},
    "document": {"expiry_date": "2024-12-31"},
    "sys": tool_parser.build_sys_context(),
}

# 1) Evaluate tool function call template
date_val = tool_parser.eval("{date.get_today_yyyymmdd()}", context_dict)
# -> "20260824"

# 2) Evaluate system namespace & date templates
today_val = tool_parser.eval("{sys.today}", context_dict)
# -> "20260824"

3. Real-time Progress Tracking with ProgressTracker

from agent_common.utils import ProgressTracker
from agent_common.logger import ProjectLogger

logger = ProjectLogger("MyTask")
tracker = ProgressTracker(total_items_int=1000, logger_obj=logger, item_name_str="file")

for file_info in file_list:
    try:
        # Processing logic
        tracker.increment_success(bytes_int=len(data))
    except Exception as e:
        tracker.increment_failure(error_msg_str=str(e))

# Output final execution summary report
tracker.log_summary()

4. Unified Text/SQL Generation with LlmClient

from agent_common.llm import LlmClient

# 1) Initialize client with configured purpose or model name
llm_client = LlmClient(purpose_str="sql_generator")

# 2) Generate text using the selected profile's provider_str
prompt_str = "User request: Generate daily subscriber statistics SQL for August 2026."
response_str = llm_client.generate(
    prompt_str=prompt_str,
    system_prompt_str="You are an expert AI for BigQuery SQL generation."
)

print(f"Generated result ({llm_client.last_generated_by_str}):\n{response_str}")

5. Storage and BigQuery Client Usage

S3/ECS Object Traversal & Direct GCS Streaming (Smart Skipping)

from agent_common.clients import S3Client, GcsClient

# Initialize S3/ECS and GCS clients (fail-fast bucket validation)
s3_client = S3Client(
    endpoint_url_str="https://ecs.mycorp.internal:9021",
    access_key_str="MY_ACCESS_KEY",
    secret_key_str="MY_SECRET_KEY",
    bucket_name_str="source-lake",
)
gcs_client = GcsClient(bucket_name_str="target-lake")

# Iterate over large object prefixes and stream into GCS directly
for obj_dict in s3_client.list_objects(prefix_str="raw/events/20260824/"):
    s3_key_str: str = obj_dict["Key"]
    size_int: int = obj_dict["Size"]
    gcs_blob_str: str = f"lake/{s3_key_str.lstrip('/')}"
    
    # "SKIPPED" if identical file exists, "UPLOADED" on successful stream
    status_str = s3_client.transfer_to_gcs(
        gcs_client_obj=gcs_client,
        s3_key_str=s3_key_str,
        gcs_blob_name_str=gcs_blob_str,
        size_int=size_int,
    )
    print(f"[{status_str}] {s3_key_str} -> {gcs_blob_str} ({size_int:,} bytes)")

BigQuery High-Performance Inline MERGE (Upsert)

from agent_common.clients import BigQueryClient

bq_client = BigQueryClient(
    project_id_str="my-gcp-project",
    dataset_id_str="analytics_dw",
    table_id_str="tb_member_profile",
)

members_list = [
    {
        "member_id": "M001",
        "name": "Alex",
        "login_count": 42,
        "is_vip": True,
        "created_at": "2026-01-01 00:00:00+09:00",
        "last_login_at": "2026-08-24 15:30:00+09:00",
    }
]

# Direct parameterized MERGE without staging tables (preserves created_at)
bq_client.merge_table_from_json_data(
    json_data_any=members_list,
    pk_key_str="member_id",
    preserve_columns_list=["created_at"],
    chunk_size_int=100,
)

🚀 Installation and Build Guide

📦 Wheel Package Build (.whl)

Run the following commands within the agent_common directory or using scripts/build_agent_common_whl.py:

1. Air-gapped / Offline Environment

Use --no-index, --no-build-isolation, and --no-deps to build offline without external PyPI access:

# Recommended: Run build script from root
python scripts/build_agent_common_whl.py

# Or build wheel directly
pip wheel ./agent_common --no-index --no-build-isolation --no-deps -w whls/

2. Online Environment

# Using pip wheel
pip wheel ./agent_common --no-deps -w whls/

# Or using build module
python -m build agent_common --wheel -o whls/

Installing the Wheel Package

# Development (Editable mode - lightweight core)
pip install -e agent_common

# Development (With cloud client extras)
pip install -e "agent_common[clients]"

# Production (Wheel package)
pip install dist/agent_common-0.4.78-py3-none-any.whl

🌐 Official PyPI Distribution (Maintainers Only)

# 1. Update build tools
pip install build twine

# 2. Build distributions (sdist & wheel)
python -m build

# 3. Validate distribution archives
python -m twine check dist/*

# 4. Upload to PyPI
python -m twine upload dist/agent_common-0.4.78*

📖 Detailed Feature Manuals

For comprehensive architecture details and practical code examples for each module:

📋 Version History (Changelog)

v0.4.78 (2026-09-21)

• Added AI request examples using requirements, package links, and feature numbers, with incremental implementation guidance.
• Added Korean/English LLM manuals covering configuration, external APIs, local GGUF, environment overrides, and failure handling.
• Documented the Groq supervisor and Antigravity Stop hook example, including review scope and skipped-review conditions.
• Corrected conditional fallback documentation and omitted the unfinished error-handler introduction.
• Bumped the package version to 0.4.78 with no runtime behavior changes.

For detailed version history, please refer to GitHub CHANGELOG_EN.md.