如何快速實現REST API集成以優化業務流程
我們需要創建3個核心文件：
project/
├── main.py # 主程序
├── actions.py # 功能函數
├── prompts.py # 提示詞模板
└── .env # 環境變量OPENAI_API_KEY=sk-your-api-key-here先在actions.py中定義Agent可以執行的操作：
def get_website_info(url)：
“”“模擬獲取網站信息的函數”“”
info = {
“google.com”： {“response_time”： 0.3， “status”： “online”}，
“github.com”： {“response_time”： 0.5， “status”： “online”}
}
return info.get(url， {“response_time”： 1.0， “status”： “unknown”})
def search_knowledge(query)：
“”“模擬知識庫搜索”“”
knowledge_base = {
“python”： “Python是一種高級編程語言”，
“openai”： “OpenAI是一家AI研究公司”
}
return knowledge_base.get(query， “未找到相關信息”)在prompts.py中設置ReAct提示詞模板：
SYSTEM_PROMPT = “”“你是一個智能AI助手，運行在Thought(思考) -> Action(行動) -> Response(響應)的循環中。
可用的操作有：
1. get_website_info： 獲取網站信息
2. search_knowledge： 搜索知識庫
示例格式：
Thought： 我需要了解網站狀態
Action： {”name“： ”get_website_info“， ”args“： {”url“： ”google.com“}}
Response： 根據獲得的信息進行回答
每次行動后請說”PAUSE“等待響應。
”“”在main.py中實現Agent的核心邏輯：
from openai import OpenAI
import json
import os
from dotenv import load_dotenv
from actions import *
from prompts import SYSTEM_PROMPT
load_dotenv()
client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
def run_conversation(user_input)：
messages = [
{“role”： “system”， “content”：SYSTEM_PROMPT}，
{“role”： “user”， “content”：user_input}
]
for _ in range(5)： # 最多5輪對話
response = client.chat.completions.create(
model=“gpt-3.5-turbo”，
messages=messages
)
assistant_message = response.choices[0].message.content
if “PAUSE” not in assistant_message：
return assistant_message
# 解析Action并執行
try：
action = json.loads(assistant_message.split(“Action： ”)[1].split(“PAUSE”)[0])
if action[“name”] == “get_website_info”：
result = get_website_info(**action[“args”
elif action[“name”] == “search_knowledge”：
result = search_knowledge(**action[“args”])
messages.append({“role”： “assistant”， “content”：assistant_message})
messages.append({“role”： “user”， “content”：f“Action result： {result}”})
except Exception as e：
messages.append({“role”： “user”， “content”：f“Error： {str(e)}”})
return “抱歉，我沒能完成這個任務”
# 測試運行
if __name__ == “__main__”：
test_input = “google.com的響應時間是多少？”
print(run_conversation(test_input)) ## 6. 添加記憶模塊
為了讓我們的Agent能夠記住對話歷史，我們新建一個memory.py：
```python
class ConversationMemory：
def __init__(self， max_tokens=1000)：
self.conversations = []
self.max_tokens = max_tokens
def add_memory(self， role， content)：
self.conversations.append({
“role”：role，
“content”：content
})
# 簡單的記憶管理，超過最大限制就移除最早的記憶
while self._estimate_tokens() > self.max_tokens：
self.conversations.pop(0)
def _estimate_tokens(self)：
# 粗略估算token數量
return sum(len(conv[“content”].split()) * 1.3 for conv in self.conversations)
def get_relevant_memory(self， query)：
# 簡單的相關性搜索
relevant = []
for conv in self.conversations：
if any(word in conv[“content”].lower()
for word in query.lower().split())：
relevant.append(conv)
return relevant
創建utils.py添加輔助功能：
import logging
import time
from functools import wraps
# 配置日志
logging.basicConfig(
level=logging.INFO，
format='%(asctime)s - %(levelname)s - %(message)s'，
filename='agent.log'
)
def retry_on_error(max_retries=3， delay=1)：
“”“錯誤重試裝飾器”“”
def decorator(func)：
@wraps(func)
def wrapper(*args， **kwargs)：
for i in range(max_retries)：
try：
return func(*args， **kwargs)
except Exception as e：
logging.error(f“嘗試 {i+1}/{max_retries} 失?。?{str(e)}”)
if i < max_retries - 1：
time.sleep(delay)
raise Exception(f“在{max_retries}次嘗試后失敗”)
return wrapper
return decorator擴展actions.py with更多功能：
import pandas as pd
import requests
from bs4 import BeautifulSoup
from utils import retry_on_error
class AgentActions：
@retry_on_error(max_retries=3)
def web_search(self， query)：
“”“模擬網絡搜索”“”
# 實際項目中可以接入搜索API
return f“搜索結果： {query}”
def analyze_data(self， data_str)：
“”“簡單的數據分析”“”
try：
# 將字符串轉換為DataFrame
data = pd.read_json(data_str)
analysis = {
“rows”：len(data)，
“columns”：list(data.columns)，
“summary”： data.describe().to_dict()
}
return analysis
except Exception as e：
return f“數據分析錯誤： {str(e)}”
@retry_on_error()
def fetch_webpage(self， url)：
“”“獲取網頁內容”“”
response = requests.get(url)
soup = BeautifulSoup(response.text， 'html.parser')
return {
“title”：soup.title.string if soup.title else “No title”，
“text”：soup.get_text()[：500] # 只返回前500字符
}更新main.py使用新功能：
from memory import ConversationMemory
from actions import AgentActions
import logging
from utils import retry_on_error
class Agent：
def __init__(self)：
self.memory = ConversationMemory()
self.actions = AgentActions()
self.client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
@retry_on_error()
def get_response(self， user_input)：
# 獲取相關記憶
relevant_memory = self.memory.get_relevant_memory(user_input)
# 構建消息
messages = [
{“role”： “system”， “content”：SYSTEM_PROMPT}，
*relevant_memory，
{“role”： “user”， “content”：user_input}
]
try：
response = self.client.chat.completions.create(
model=“gpt-3.5-turbo”，
messages=messages
)
response_content = response.choices[0].message.content
self.memory.add_memory(“assistant”， response_content)
return response_content
except Exception as e：
logging.error(f“獲取響應失?。?{str(e)}”)
return “抱歉，我遇到了一些問題，請稍后再試”
# 使用示例
if __name__ == “__main__”：
agent = Agent()
# 測試對話
conversations = [
“Python是什么編程語言？”，
“它有哪些主要特點？”，
“給我看一個簡單的Python代碼示例”
]
for conv in conversations：
print(f“\n用戶： {conv}”)
response = agent.get_response(conv)
print(f“AI： {response}”)
試試實現這些功能：
本文章轉載微信公眾號@樓談
