DuckDuckGo의 검색 API를 사용하여 대형 모델에게 네트워킹 기능에 대한 무료 액세스를 제공함으로써 뉴 빙과 같은 효과를 얻을 수 있습니다.

프로젝트 개요

통합을 통해 DuckDuckGo 공통의 대규모 언어 모델에 유사한 New Bing 실시간 웹 액세스 및 지능형 Q&A 기능을 제공합니다. 기존 대형 모델의 멀티플렉싱 함수 호출함수를 사용하여 효율적이고 유연한 쿼리 처리 및 답변 생성 프로세스를 지원합니다.

핵심 실현

이 시스템은 Function calling 쿼리 처리와 답변 생성의 원활한 통합을 위한 메커니즘입니다:

1. 쿼리 분석 및 검색: 다음을 사용하여 claude-3-haiku 어쩌면 gpt-3.5-turbo 더 간단한 모델은 최적화된 검색어를 생성하고 함수 호출을 통해 검색 API를 호출합니다.
2. 웹 검색: 실시간 정보 검색을 수행하기 위해 DuckDuckGo API를 호출합니다.
3. 지능형 답변 생성: 다음을 사용하여 claude-3-5-sonnnet 어쩌면 gpt-4o 이와 같은 보다 복잡한 모델은 검색 결과와 원래 질문을 결합하여 정확하고 포괄적인 답변을 생성합니다.

코드 구현

1. 환경 설정 및 기능 정의

import os
import json
from openai import OpenAI
from duckduckgo_search import DDGS
API_KEY = os.getenv("OPENAI_API_KEY")
BASE_URL = os.getenv("OPENAI_BASE_URL")
client = OpenAI(api_key=API_KEY, base_url=BASE_URL)
FUNCTIONS = [
{
"name": "search_duckduckgo",
"description": "使用DuckDuckGo搜索引擎查询信息。可以搜索最新新闻、文章、博客等内容。",
"parameters": {
"type": "object",
"properties": {
"keywords": {
"type": "array",
"items": {"type": "string"},
"description": "搜索的关键词列表。例如：['Python', '机器学习', '最新进展']。"
}
},
"required": ["keywords"]
}
}
]

2. 관련 보조 기능

def search_duckduckgo(keywords):
search_term = " ".join(keywords)
with DDGS() as ddgs:
return list(ddgs.text(keywords=search_term, region="cn-zh", safesearch="on", max_results=5))
def print_search_results(results):
for result in results:
print(
f"标题: {result['title']}\n链接: {result['href']}\n摘要: {result['body']}\n---")
def get_openai_response(messages, model="gpt-3.5-turbo", functions=None, function_call=None):
try:
response = client.chat.completions.create(
model=model,
messages=messages,
functions=functions,
function_call=function_call
)
return response.choices[0].message
except Exception as e:
print(f"调用OpenAI API时出错: {str(e)}")
return None
def process_function_call(response_message):
function_name = response_message.function_call.name
function_args = json.loads(response_message.function_call.arguments)
print(f"\n模型选择调用函数: {function_name}")
if function_name == "search_duckduckgo":
keywords = function_args.get('keywords', [])
if not keywords:
print("错误：模型没有提供搜索关键词")
return None
print(f"关键词: {', '.join(keywords)}")
function_response = search_duckduckgo(keywords)
print("\nDuckDuckGo搜索返回结果:")
print_search_results(function_response)
return function_response
else:
print(f"未知的函数名称: {function_name}")
return None

3. 주요 처리 기능

def main(question):
print(f"问题：{question}")
messages = [{"role": "user", "content": question}]
response_message = get_openai_response(
messages, functions=FUNCTIONS, function_call="auto")
if not response_message:
return
if response_message.function_call:
function_response = process_function_call(response_message)
if function_response:
messages.extend([
response_message.model_dump(),
{
"role": "function",
"name": response_message.function_call.name,
"content": json.dumps(function_response, ensure_ascii=False)
}
])
final_response = get_openai_response(messages, model="gpt-4o")
if final_response:
print("\n最终回答:")
print(final_response.content)
else:
print("\n模型直接回答:")
print(response_message.content)

실현

이러한 방식으로 예를 들어 다양한 질문에 답할 수 있는 대형 모델을 만들 수 있습니다:

1. 지리 정보: "PRD에 포산이 포함되나요?"
2. AI의 자체 기능에 대한 질문, "무엇을 할 수 있나요?"
3. 도메인별 질문, "식물 대 좀비 하이브리드의 저자는 누구인가요?"
4. 철학적 사색: "인생은 두 단어인데, 그 두 단어는 무엇일까요?"
5. "점핑 불평등이란 무엇인가요?" 관련 시사 이슈
6. 시간 관련 질문: "현재 베이징 시간이 어떻게 되나요?

전체 코드

import os
import json
from openai import OpenAI
from duckduckgo_search import DDGS

API_KEY = os.getenv("OPENAI_API_KEY")
BASE_URL = os.getenv("OPENAI_BASE_URL")
client = OpenAI(api_key=API_KEY, base_url=BASE_URL)

FUNCTIONS = [
{
"name": "search_duckduckgo",
"description": "使用DuckDuckGo搜索引擎查询信息。可以搜索最新新闻、文章、博客等内容。",
"parameters": {
"type": "object",
"properties": {
"keywords": {
"type": "array",
"items": {"type": "string"},
"description": "搜索的关键词列表。例如：['Python', '机器学习', '最新进展']。"
}
},
"required": ["keywords"]
}
}
]


def search_duckduckgo(keywords):
search_term = " ".join(keywords)
with DDGS() as ddgs:
return list(ddgs.text(keywords=search_term, region="cn-zh", safesearch="on", max_results=5))


def print_search_results(results):
for result in results:
print(
f"标题: {result['title']}\n链接: {result['href']}\n摘要: {result['body']}\n---")


def get_openai_response(messages, model="gpt-3.5-turbo", functions=None, function_call=None):
try:
response = client.chat.completions.create(
model=model,
messages=messages,
functions=functions,
function_call=function_call
)
return response.choices[0].message
except Exception as e:
print(f"调用OpenAI API时出错: {str(e)}")
return None


def process_function_call(response_message):
function_name = response_message.function_call.name
function_args = json.loads(response_message.function_call.arguments)

print(f"\n模型选择调用函数: {function_name}")

if function_name == "search_duckduckgo":
keywords = function_args.get('keywords', [])

if not keywords:
print("错误：模型没有提供搜索关键词")
return None

print(f"关键词: {', '.join(keywords)}")

function_response = search_duckduckgo(keywords)
print("\nDuckDuckGo搜索返回结果:")
print_search_results(function_response)

return function_response
else:
print(f"未知的函数名称: {function_name}")
return None


def main(question):
print(f"问题：{question}")

messages = [{"role": "user", "content": question}]
response_message = get_openai_response(
messages, functions=FUNCTIONS, function_call="auto")

if not response_message:
return

if response_message.function_call:
if not response_message.content:
response_message.content = ""
function_response = process_function_call(response_message)
if function_response:
messages.extend([
response_message.model_dump(),
{
"role": "function",
"name": response_message.function_call.name,
"content": json.dumps(function_response, ensure_ascii=False)
}
])

final_response = get_openai_response(messages, model="gpt-4o")
if final_response:
print("\n最终回答:")
print(final_response.content)
else:
print("\n模型直接回答:")
print(response_message.content)


if __name__ == "__main__":
main("植物大战僵尸杂交版的作者是谁？他是怎么想到做出来这个游戏的？")