Agent 프레임워크와 엔지니어링

import json, re
from openai import OpenAI
from datetime import datetime

client = OpenAI()

# ========== 1. 도구 정의 ==========
def search_web(query):
    """웹 검색 시뮬레이션"""
    fake_results = {
        "Python": "Python은 고급 프로그래밍 언어이며, 최신 버전은 3.12입니다.",
        "RAG": "RAG는 검색 증강 생성으로, 2024-2025년 가장 인기 있는 AI 기술 중 하나입니다.",
    }
    for key, val in fake_results.items():
        if key.lower() in query.lower():
            return val
    return f"'{query}'에 대한 검색 결과를 찾을 수 없습니다"

def calculator(expression):
    """수학 표현식 안전 계산"""
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except:
        return "계산 오류"

def get_current_time():
    """현재 시간 가져오기"""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

TOOLS = {
    "search_web": {"func": search_web, "desc": "웹 정보 검색, 매개변수: query"},
    "calculator": {"func": calculator, "desc": "수학 표현식 계산, 매개변수: expression"},
    "get_current_time": {"func": get_current_time, "desc": "현재 시간 가져오기, 매개변수 없음"},
}

# ========== 2. Agent 시스템 Prompt ==========
AGENT_PROMPT = """당신은 도구를 사용하여 작업을 완수할 수 있는 지능형 어시스턴트입니다.

사용 가능한 도구:
{tool_descriptions}

작업 흐름:
1. 사용자 요구 분석
2. 필요시 도구를 호출하여 정보 획득
3. 결과를 바탕으로 질문에 답변

도구 호출 시, 다음 JSON 형식을 엄격히 사용하세요 (한 줄로 작성):
TOOL_CALL: {{"tool": "도구명", "args": {{"매개변수명": "매개변수값"}}}}

도구가 필요 없으면 바로 답변하세요.
작업이 완료되면 FINAL_ANSWER: 로 시작하는 최종 답변을 제공하세요."""

# ========== 3. Agent 루프 ==========
class SimpleAgent:
    def __init__(self, tools, max_steps=5):
        self.tools = tools
        self.max_steps = max_steps

        tool_desc = "\n".join(
            [f"- {name}: {t['desc']}" for name, t in tools.items()]
        )
        self.system_prompt = AGENT_PROMPT.format(tool_descriptions=tool_desc)

    def run(self, task):
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": task}
        ]

        for step in range(self.max_steps):
            print(f"\n--- Step {step + 1} ---")

            response = client.chat.completions.create(
                model="gpt-4o", messages=messages, temperature=0.3
            )
            reply = response.choices[0].message.content
            print(f"AI: {reply}")
            messages.append({"role": "assistant", "content": reply})

            # 최종 답변 확인
            if "FINAL_ANSWER:" in reply:
                return reply.split("FINAL_ANSWER:")[1].strip()

            # 도구 호출 확인
            if "TOOL_CALL:" in reply:
                tool_json = reply.split("TOOL_CALL:")[1].strip()
                try:
                    call = json.loads(tool_json)
                    tool_name = call["tool"]
                    tool_args = call.get("args", {})

                    print(f"🔧 도구 호출: {tool_name}({tool_args})")
                    result = self.tools[tool_name]["func"](**tool_args)
                    print(f"📋 결과: {result}")

                    messages.append({
                        "role": "user",
                        "content": f"도구 {tool_name}의 실행 결과: {result}"
                    })
                except Exception as e:
                    messages.append({
                        "role": "user", "content": f"도구 호출 실패: {e}"
                    })

        return "최대 단계 수에 도달, 작업 미완료"

# ========== 사용 ==========
agent = SimpleAgent(TOOLS)
answer = agent.run("지금 몇 시야? 그리고 1024 * 768이 얼마인지 계산해줘")
print(f"\n최종 답변: {answer}")

pip install langchain langchain-openai langchain-community duckduckgo-search

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool

# ---- 1. 도구 정의 (데코레이터 사용, 매우 간단) ----
@tool
def calculator(expression: str) -> str:
    """수학 표현식을 계산합니다. '2 + 3 * 4'와 같은 수학 표현식 문자열을 입력합니다"""
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except:
        return "계산 오류"

@tool
def get_word_count(text: str) -> str:
    """텍스트의 단어 수와 문자 수를 집계합니다"""
    words = len(text.split())
    chars = len(text)
    return f"단어 수: {words}, 문자 수: {chars}"

tools = [calculator, get_word_count]

# ---- 2. Agent 생성 ----
llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 도구를 사용하여 작업을 완수할 수 있는 유능한 AI 어시스턴트입니다."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# ---- 3. 실행 ----
result = executor.invoke({"input": "(125 * 37) + (89 * 43)의 결과를 계산해주세요"})
print(result["output"])

@tool
def search_knowledge_base(query: str) -> str:
    """기업 지식 베이스에서 관련 정보를 검색합니다. 사용자가 제품, 정책, 프로세스 관련 질문을 할 때 사용합니다."""
    # Phase 4의 검색 로직 재사용
    emb = get_embeddings([query])[0]
    results = collection.query(query_embeddings=[emb], n_results=3)
    if results["documents"][0]:
        return "\n---\n".join(results["documents"][0])
    return "지식 베이스에서 관련 정보를 찾을 수 없습니다"

# Agent가 동시에 보유: 지식 베이스 검색 + 계산기 + 기타 도구
tools = [search_knowledge_base, calculator, get_word_count]
# AI가 자동 판단: 사용자가 제품 질문 → 지식 베이스 조회, 수학 질문 → 계산기 사용

pip install streamlit

import streamlit as st
from openai import OpenAI

st.set_page_config(page_title="AI 知识库问答", page_icon="🧠")
st.title("🧠 企业知识库问答系统")

# ---- 사이드바: 파일 업로드 ----
with st.sidebar:
    st.header("📁 文档管理")
    uploaded = st.file_uploader("문서 업로드", type=["pdf", "txt"], accept_multiple_files=True)
    if uploaded:
        st.success(f"{len(uploaded)}개 파일 업로드 완료")
        # 여기서 입고 로직 호출...

# ---- 채팅 인터페이스 ----
if "messages" not in st.session_state:
    st.session_state.messages = []

# 이전 메시지 표시
for msg in st.session_state.messages:
    with st.chat_message(msg["role"]):
        st.markdown(msg["content"])

# 입력창
if question := st.chat_input("질문을 입력하세요..."):
    # 사용자 메시지 표시
    st.session_state.messages.append({"role": "user", "content": question})
    with st.chat_message("user"):
        st.markdown(question)

    # AI 답변 표시
    with st.chat_message("assistant"):
        with st.spinner("생각 중..."):
            # RAG Q&A 함수 호출
            result = rag_query(question)  # Phase 4의 함수
            st.markdown(result["answer"])
            if result["sources"]:
                st.caption(f"📎 출처: {', '.join(result['sources'])}")

    st.session_state.messages.append({"role": "assistant", "content": result["answer"]})

streamlit run app.py    # 브라우저가 자동으로 http://localhost:8501 열기

import gradio as gr

def chat_fn(message, history):
    result = rag_query(message)
    return result["answer"]

demo = gr.ChatInterface(fn=chat_fn, title="AI 知识库问答")
demo.launch()   # → http://localhost:7860

pip install fastapi uvicorn

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List

app = FastAPI(title="AI 知识库 API")

# ---- 요청/응답 모델 ----
class QueryRequest(BaseModel):
    question: str
    top_k: int = 3

class QueryResponse(BaseModel):
    answer: str
    sources: List[str]
    tokens: int

# ---- API 라우트 ----
@app.post("/ask", response_model=QueryResponse)
async def ask_question(req: QueryRequest):
    try:
        result = rag_query(req.question, top_k=req.top_k)
        return QueryResponse(
            answer=result["answer"],
            sources=result["sources"],
            tokens=result.get("tokens", 0)
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health():
    return {"status": "ok"}

# 실행: uvicorn api_server:app --reload
# 자동 문서: http://localhost:8000/docs

import pandas as pd

# ---- 수동으로 평가 데이터셋 구축 ----
eval_data = [
    {"question": "반품 절차가 어떻게 되나요?",
     "expected_answer": "7일 이내 반품 가능, 반품 신청서 제출 필요",
     "expected_source": "반품교환정책.pdf"},
    {"question": "제품 보증 기간은 얼마나 되나요?",
     "expected_answer": "표준 보증 1년",
     "expected_source": "제품매뉴얼.pdf"},
    # ... 최소 20-50개
]
eval_df = pd.DataFrame(eval_data)

# ---- 일괄 평가 실행 ----
results = []
for _, row in eval_df.iterrows():
    result = rag_query(row["question"])
    results.append({
        "question": row["question"],
        "expected": row["expected_answer"],
        "actual": result["answer"],
        "sources": result["sources"],
        "source_hit": row["expected_source"] in result["sources"]
    })

results_df = pd.DataFrame(results)
print(f"출처 적중률: {results_df['source_hit'].mean():.1%}")

JUDGE_PROMPT = """당신은 AI 답변 품질 평가관입니다. 아래 답변의 품질을 평가해 주세요.

질문: {question}
참고 답변: {expected}
실제 답변: {actual}

다음 차원에서 점수를 매겨주세요 (1-5):
- 정확성: 답변이 올바른가
- 완전성: 핵심 정보를 포함하고 있는가
- 간결성: 간결하고 명확한가

JSON 출력: {{"accuracy": X, "completeness": X, "conciseness": X, "total": X, "comment": "..."}}"""

def judge_answer(question, expected, actual):
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": JUDGE_PROMPT.format(
            question=question, expected=expected, actual=actual
        )}],
        response_format={"type": "json_object"},
        temperature=0
    )
    return json.loads(resp.choices[0].message.content)

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "api_server:app", "--host", "0.0.0.0", "--port", "8000"]

docker build -t my-ai-app .           # 이미지 빌드
docker run -p 8000:8000 my-ai-app     # 컨테이너 실행

my-ai-project/
├── app/
│   ├── main.py            # FastAPI 진입점
│   ├── rag.py             # RAG 핵심 로직
│   ├── agent.py           # Agent 로직
│   ├── tools.py           # 도구 정의
│   └── prompts.py         # Prompt 템플릿
├── data/
│   └── docs/              # 지식 베이스 문서
├── eval/
│   ├── dataset.json       # 평가 데이터
│   └── run_eval.py        # 평가 스크립트
├── streamlit_app.py       # Web 인터페이스
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
├── .env                   # API Key (Git에 커밋하지 않음)
├── .gitignore
└── README.md

class ResearchAgent:
    """AI 리서치 어시스턴트 — RAG + Agent + 보고서 생성 통합"""

    def __init__(self):
        self.kb = KnowledgeBaseQA()       # Phase 4의 지식 베이스
        self.tools = self._init_tools()    # Agent 도구 모음
        self.cost_tracker = CostTracker()  # Phase 3의 비용 추적

    def _init_tools(self):
        return {
            "search_kb": self.kb.ask,           # 지식 베이스 검색
            "web_search": web_search,           # 웹 검색
            "calculator": calculator,            # 계산기
            "generate_report": self.gen_report, # 보고서 생성
        }

    def chat(self, user_input):
        """Agent 기반 채팅"""
        # Agent 루프: 사고 → 행동 → 관찰 → 결정
        ...

    def gen_report(self, topic):
        """연구 보고서 자동 생성"""
        # 1. 지식 베이스에서 관련 정보 검색
        # 2. 웹 검색으로 보충 정보 획득
        # 3. LLM으로 구조화된 보고서 통합
        # 4. Markdown 형식 보고서 반환
        ...

    def upload_docs(self, files):
        """문서 일괄 업로드"""
        for f in files:
            self.kb.add_document(f)

    def get_stats(self):
        """시스템 통계 정보"""
        return {
            "kb_chunks": self.kb.stats()["total_chunks"],
            "total_cost": self.cost_tracker.total_cost,
            "total_calls": self.cost_tracker.call_count,
        }