0환경 설정
1
Ollama + Mistral —
ollama pull mistral (7B, ~4.1GB)2
패키지 —
pip install fastapi uvicorn ollama langchain-community chromadb sentence-transformers1Function Calling 에이전트
Mistral의 네이티브 Function Calling을 사용해 외부 API, DB, 파일 시스템과 연동하는 에이전트를 만듭니다.
function_agent.py — 날씨/계산기/파일 에이전트
자연어 명령을 받아 적절한 도구를 선택·실행하는 완전한 에이전트. Ollama의 tool_calls 기능 활용.
import json
from ollama import Client
ollama = Client()
# ── 도구 정의
TOOLS = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "특정 도시의 현재 날씨 정보를 가져옵니다.",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "도시 이름 (예: 서울, 부산)"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"], "default": "celsius"},
},
"required": ["city"],
},
},
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "수학 계산을 수행합니다.",
"parameters": {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "계산할 수식 (예: 2 + 3 * 4)"},
},
"required": ["expression"],
},
},
},
{
"type": "function",
"function": {
"name": "search_file",
"description": "로컬 파일에서 텍스트를 검색합니다.",
"parameters": {
"type": "object",
"properties": {
"filename": {"type": "string"},
"query": {"type": "string"},
},
"required": ["filename", "query"],
},
},
},
]
# ── 도구 실행 함수들
def get_weather(city: str, unit: str = "celsius") -> dict:
# 실제 날씨 API 연동 시 여기에 구현
return {"city": city, "temp": 22, "condition": "맑음", "unit": unit}
def calculate(expression: str) -> dict:
try:
result = eval(expression, {"__builtins__": {}}) # 안전한 eval
return {"expression": expression, "result": result}
except Exception as e:
return {"error": str(e)}
def search_file(filename: str, query: str) -> dict:
try:
with open(filename) as f:
lines = [l for l in f.readlines() if query.lower() in l.lower()]
return {"matches": lines[:5], "total": len(lines)}
except FileNotFoundError:
return {"error": f"{filename} 파일을 찾을 수 없습니다."}
TOOL_FUNCTIONS = {
"get_weather": get_weather,
"calculate": calculate,
"search_file": search_file,
}
# ── 에이전트 실행
def run_agent(user_message: str) -> str:
messages = [{"role": "user", "content": user_message}]
while True:
resp = ollama.chat(
model="mistral",
messages=messages,
tools=TOOLS,
)
msg = resp["message"]
messages.append(msg)
# 도구 호출 없으면 최종 응답
if not msg.get("tool_calls"):
return msg["content"]
# 도구 실행
for tool_call in msg["tool_calls"]:
fn_name = tool_call["function"]["name"]
fn_args = json.loads(tool_call["function"]["arguments"])
result = TOOL_FUNCTIONS[fn_name](**fn_args)
messages.append({
"role": "tool",
"content": json.dumps(result, ensure_ascii=False),
})
# 테스트
queries = [
"서울 날씨는 어때?",
"123 * 456 + 789 계산해줘",
"오늘 밥 먹었어?", # 도구 불필요한 일반 대화
]
for q in queries:
print(f"Q: {q}")
print(f"A: {run_agent(q)}\n")python2다중 도구 ReAct 에이전트
여러 도구를 순서대로 사용하며 복잡한 작업을 단계적으로 해결하는 에이전트입니다.
react_agent.py — 연구 보조 에이전트
from langchain_community.llms import Ollama
from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import tool
from langchain import hub
from langchain.prompts import PromptTemplate
llm = Ollama(model="mistral", temperature=0)
@tool
def web_search(query: str) -> str:
"""인터넷에서 정보를 검색합니다. 최신 정보나 모르는 사실을 찾을 때 사용합니다."""
# 실제로는 DuckDuckGo, SerpAPI 등 연동
return f"'{query}'에 대한 검색 결과: [데모 결과] Python 3.12가 2023년 10월 출시되었습니다."
@tool
def read_file(filepath: str) -> str:
"""로컬 파일의 내용을 읽습니다. filepath는 파일 경로입니다."""
try:
with open(filepath) as f:
return f.read()[:2000] # 처음 2000자
except Exception as e:
return f"오류: {e}"
@tool
def write_summary(content: str, filename: str = "summary.txt") -> str:
"""내용을 파일로 저장합니다."""
with open(filename, "w", encoding="utf-8") as f:
f.write(content)
return f"✅ {filename}에 저장되었습니다."
@tool
def math_calc(expression: str) -> str:
"""수학 계산을 수행합니다. 예: '2**10' → 1024"""
try:
return str(eval(expression, {"__builtins__": {}, "abs": abs, "round": round}))
except Exception as e:
return f"계산 오류: {e}"
tools = [web_search, read_file, write_summary, math_calc]
# ReAct 프롬프트 (한국어 커스텀)
react_prompt = PromptTemplate.from_template("""
당신은 주어진 도구를 활용해 문제를 해결하는 AI 에이전트입니다.
사용 가능한 도구:
{tools}
형식:
생각: [현재 상황 분석]
행동: [도구 이름]
행동 입력: [도구에 전달할 입력]
관찰: [도구 실행 결과]
... (반복)
최종 답변: [사용자에게 전달할 최종 답변]
질문: {input}
{agent_scratchpad}
""")
agent = create_react_agent(llm, tools, react_prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=5)
result = agent_executor.invoke({
"input": "Python 최신 버전을 검색하고, 2의 10승을 계산해서 정리해줘"
})
print(result["output"])python3RAG + 로컬 임베딩
mistral_rag.py — 사내 문서 Q&A
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.llms import Ollama
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory
# 무료 다국어 임베딩 모델
embeddings = HuggingFaceEmbeddings(
model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
)
# 문서 인덱싱
def index_documents(file_paths: list[str]) -> Chroma:
all_docs = []
splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=40)
for path in file_paths:
with open(path, encoding="utf-8") as f:
text = f.read()
from langchain.schema import Document
docs = splitter.split_documents([Document(page_content=text, metadata={"source": path})])
all_docs.extend(docs)
return Chroma.from_documents(all_docs, embeddings, persist_directory="./mistral_rag_db")
# 대화형 QA (대화 히스토리 유지)
def create_conversational_qa(vectorstore: Chroma):
llm = Ollama(model="mistral", temperature=0.3)
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True,
output_key="answer",
)
return ConversationalRetrievalChain.from_llm(
llm=llm,
retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
memory=memory,
return_source_documents=True,
)
# 사용 예시
if __name__ == "__main__":
# 샘플 문서 생성
with open("company_policy.txt", "w") as f:
f.write("우리 회사 휴가 정책: 연간 15일의 유급 휴가를 제공합니다.\n")
f.write("재택근무: 주 3일까지 재택 근무 가능합니다.\n")
f.write("점심 시간: 12시~13시입니다.\n")
vs = index_documents(["company_policy.txt"])
qa = create_conversational_qa(vs)
questions = ["휴가는 몇 일이야?", "재택 근무는 얼마나 할 수 있어?"]
for q in questions:
result = qa.invoke({"question": q})
print(f"Q: {q}")
print(f"A: {result['answer']}\n")python4텍스트 분류 API
classifier_api.py — 다중 레이블 분류
from fastapi import FastAPI
from pydantic import BaseModel
from ollama import Client
import json
app = FastAPI(title="Text Classifier API")
ollama = Client()
class ClassifyRequest(BaseModel):
text: str
categories: list[str]
multi_label: bool = False # True면 여러 카테고리 허용
class ClassifyResponse(BaseModel):
text: str
category: str | list[str]
confidence: float
reason: str
@app.post("/classify", response_model=ClassifyResponse)
async def classify(req: ClassifyRequest):
cats = ", ".join(req.categories)
mode = "하나 이상" if req.multi_label else "정확히 하나"
resp = ollama.chat(
model="mistral",
messages=[{
"role": "system",
"content": f"""텍스트를 분류해 JSON으로 반환하세요.
카테고리: {cats}
{mode}의 카테고리를 선택하세요.
JSON 형식:
{{"category": {"[배열]" if req.multi_label else '"단일 문자열"'},
"confidence": 0.0~1.0,
"reason": "선택 이유 한 문장"}}"""
}, {
"role": "user",
"content": req.text,
}],
format="json",
)
data = json.loads(resp["message"]["content"])
return ClassifyResponse(text=req.text, **data)
# 테스트: uvicorn classifier_api:app --reload
# curl -X POST http://localhost:8000/classify \
# -H "Content-Type: application/json" \
# -d '{"text":"내일 비가 온대", "categories":["날씨","음식","스포츠","기술"]}'python5번역 서비스
translation_api.py — 다국어 번역 API
from fastapi import FastAPI
from pydantic import BaseModel
from ollama import AsyncClient
import asyncio
app = FastAPI(title="Translation API")
ollama = AsyncClient()
LANGS = {"ko": "한국어", "en": "영어", "ja": "일본어", "zh": "중국어", "fr": "프랑스어", "de": "독일어"}
class TranslateRequest(BaseModel):
text: str
source: str = "auto" # 자동 감지
target: str = "en"
class TranslateResponse(BaseModel):
original: str
translated: str
source_lang: str
target_lang: str
@app.post("/translate", response_model=TranslateResponse)
async def translate(req: TranslateRequest):
target_name = LANGS.get(req.target, req.target)
if req.source == "auto":
system = f"다음 텍스트를 {target_name}로 번역하세요. 번역문만 출력하고 설명은 하지 마세요."
source_name = "자동 감지"
else:
source_name = LANGS.get(req.source, req.source)
system = f"{source_name}를 {target_name}로 번역하세요. 번역문만 출력하세요."
resp = await ollama.chat(
model="mistral",
messages=[
{"role": "system", "content": system},
{"role": "user", "content": req.text},
],
)
return TranslateResponse(
original=req.text,
translated=resp["message"]["content"].strip(),
source_lang=source_name,
target_lang=target_name,
)
# 병렬 번역 (여러 언어 동시)
@app.post("/translate/batch")
async def batch_translate(text: str, targets: list[str] = ["en", "ja", "zh"]):
tasks = [translate(TranslateRequest(text=text, target=t)) for t in targets]
results = await asyncio.gather(*tasks)
return {r.target_lang: r.translated for r in results}python