深入探索 Model Context Protocol：從理論到實踐 原創 精華
在當今快速發展的技術領域，人工智能和機器學習的創新不斷涌現。其中，Model Context Protocol（MCP）作為一種新興的開源協議，正逐漸成為研究和應用的熱點。雖然目前人們更多地關注 MCP 在工具集成方面的表現，但其在上下文管理方面的強大能力，才是真正值得我們深入探討的核心價值。
一、什么是 Model Context Protocol？
Model Context Protocol（MCP）是由 Anthropic 開發的一種開源協議，旨在解決大型語言模型（LLMs）在處理上下文信息時所面臨的諸多挑戰。它不僅關注工具的集成，更注重如何高效地管理和優化上下文信息，從而提升模型在實際應用中的表現。
二、MCP 的上下文管理能力：解決 LLMs 的核心痛點
（一）上下文窗口限制
大型語言模型的上下文窗口是有限的，這意味著它們在處理長篇對話或復雜任務時，可能會因為信息量過大而無法有效處理。MCP 提供了一種標準化的方法，能夠對對話歷史進行管理、優先級排序和壓縮，從而最大化地利用有限的上下文空間。這就好比在一個擁擠的圖書館里，通過合理安排書架和書籍，讓有限的空間能夠容納更多的知識。
（二）有狀態的對話
傳統的語言模型對話往往是無狀態的，即每次交互都是獨立的，無法記住之前的對話內容。而 MCP 通過在模型之外維護對話狀態，使得對話能夠跨越單次交互，實現更長、更連貫的交流。想象一下，當你與朋友聊天時，你們可以隨時回憶起之前的對話內容，這種連續性讓交流更加自然和深入。
（三）記憶管理
在處理大量信息時，MCP 能夠選擇性地保留重要信息，同時丟棄不相關的內容，為 AI 系統創建一個更高效的“工作記憶”。這就像人類的大腦，能夠自動篩選出重要的記憶，而忽略那些無關緊要的細節，從而提高信息處理的效率。
（四）跨會話的上下文共享
通過合理的實現，MCP 可以在不同的會話甚至不同的模型之間保持上下文的連續性。這意味著用戶在不同設備或不同場景下使用 AI 服務時，依然能夠享受到無縫的體驗。這就好比你在不同的手機或電腦上登錄同一個賬號，依然能夠看到之前的瀏覽記錄和設置。
（五）結構化的知識表示
MCP 不再將上下文視為簡單的字符序列，而是能夠以更結構化的方式表示知識。這使得 AI 系統能夠更高效地理解和處理信息，就像將雜亂無章的文件整理成有序的文件夾，方便查找和使用。
（六）檢索增強
MCP 提供了一個框架，能夠動態地從外部源檢索相關信息，并將其整合到上下文中。這就好比在做研究時，你可以隨時從圖書館或互聯網上獲取最新的資料，豐富你的知識庫。
三、為什么上下文管理如此重要？
盡管工具集成在 MCP 中備受關注，因為它能夠讓語言模型“行動”起來，但上下文管理能力同樣具有革命性。它解決了語言模型在時間維度上與信息交互的根本限制。實際上，有效的上下文管理是工具使用的基礎——模型需要理解之前使用了哪些工具、這些工具返回了什么信息，以及這些信息與當前對話狀態的關系。
四、實踐中的 MCP：一個簡單的服務器與客戶端示例
為了更好地理解 MCP 的實際應用，我們可以通過一個簡單的服務器和客戶端示例來展示其上下文管理能力。
（一）搭建 MCP 服務器
首先，我們需要創建一個本地的 MCP 服務器。在你的 MacBook 上，打開終端窗口，創建一個名為 ??mcp_context_server.py?? 的文件，并將以下代碼復制到文件中并保存。
# mcp_context_server.py
import json
import uuid
import time
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import parse_qs, urlparse
# In-memory storage for context/sessions
context_store = {}
# Simple weather database
WEATHER_DATA = {
"New York": {"temperature": 72, "condition": "Sunny", "humidity": 65},
"London": {"temperature": 60, "condition": "Rainy", "humidity": 80},
"Tokyo": {"temperature": 75, "condition": "Partly Cloudy", "humidity": 70},
"Sydney": {"temperature": 80, "condition": "Clear", "humidity": 55},
"Paris": {"temperature": 68, "condition": "Cloudy", "humidity": 75}
}
class SessionContext:
"""Represents a user session with context data"""
def __init__(self, session_id):
self.session_id = session_id
self.created_at = time.time()
self.last_accessed = time.time()
self.data = {
"recent_searches": [],
"preferred_unit": "celsius",
"visits": 0
}
def update_access(self):
self.last_accessed = time.time()
self.data["visits"] += 1
def add_search(self, city):
# Keep only the 5 most recent searches
if city notin self.data["recent_searches"]:
self.data["recent_searches"].insert(0, city)
self.data["recent_searches"] = self.data["recent_searches"][:5]
def set_preference(self, key, value):
self.data[key] = value
def to_dict(self):
return {
"session_id": self.session_id,
"created_at": self.created_at,
"last_accessed": self.last_accessed,
"data": self.data
}
class MCPRequestHandler(BaseHTTPRequestHandler):
def _set_headers(self, content_type='application/json'):
self.send_response(200)
self.send_header('Content-type', content_type)
# Now set the cookie if needed
if hasattr(self, 'should_set_cookie') and self.should_set_cookie:
self.send_header('Set-Cookie', f'session_id={self.new_session_id}; Path=/')
self.end_headers()
def _get_or_create_session(self):
# Check if client sent a session cookie
session_id = None
if'Cookie'in self.headers:
cookies = self.headers['Cookie'].split('; ')
for cookie in cookies:
if cookie.startswith('session_id='):
session_id = cookie.split('=')[1]
break
# Create new session if none exists or if it's expired
ifnot session_id or session_id notin context_store:
session_id = str(uuid.uuid4())
context_store[session_id] = SessionContext(session_id)
# Don't set the cookie header here - it's too early!
self.should_set_cookie = True# Flag to set cookie when sending headers
self.new_session_id = session_id
else:
self.should_set_cookie = False
# Update last accessed time
context_store[session_id].update_access()
return context_store[session_id]
def _clean_expired_sessions(self, max_age=3600):# 1 hour
"""Remove sessions that haven't been used for max_age seconds"""
current_time = time.time()
expired_keys = []
for key, session in context_store.items():
if current_time - session.last_accessed > max_age:
expired_keys.append(key)
for key in expired_keys:
del context_store[key]
def do_GET(self):
self._clean_expired_sessions()
parsed_url = urlparse(self.path)
path = parsed_url.path
query = parse_qs(parsed_url.query)
session = self._get_or_create_session()
# Get user's temperature preference
unit = query.get('unit', [session.data.get('preferred_unit')])[0]
# Set temperature unit preference if specified
if'unit'in query:
session.set_preference('preferred_unit', unit)
if path == '/api/weather':
self._set_headers()
# Get city from query parameters
if'city'in query:
city = query['city'][0]
session.add_search(city)
if city in WEATHER_DATA:
data = WEATHER_DATA[city].copy()
# Convert temperature based on user preference
if unit == 'fahrenheit'and'temperature'in data:
# Data is stored in Fahrenheit
pass
elif unit == 'celsius'and'temperature'in data:
# Convert Fahrenheit to Celsius
data['temperature'] = round((data['temperature'] - 32) * 5/9, 1)
response = {
"city": city,
"weather": data,
"unit": unit
}
else:
response = {"error": f"City '{city}' not found"}
else:
response = {"cities": list(WEATHER_DATA.keys())}
# Include session context data in response
response["context"] = session.to_dict()
self.wfile.write(json.dumps(response).encode())
elif path == '/api/context':
self._set_headers()
response = {"context": session.to_dict()}
self.wfile.write(json.dumps(response).encode())
else:
# Serve a simple HTML interface for manual testing
self._set_headers('text/html')
html = f"""
<!DOCTYPE html>
<html>
<head>
<title>MCP Weather Service</title>
<style>
body {{ font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; }}
.card {{ border: 1px solid #ddd; border-radius: 8px; padding: 15px; margin-bottom: 15px; }}
.recent {{ color: #666; font-size: 0.9em; }}
</style>
</head>
<body>
<h1>MCP Weather Service</h1>
<div class="card">
<h2>Your Session</h2>
<p>Session ID: {session.session_id}</p>
<p>Number of visits: {session.data["visits"]}</p>
<p>Temperature unit preference: {session.data["preferred_unit"]}</p>
<form>
<label>
Change temperature unit:
<select name="unit" notallow="this.form.submit()">
<option value="celsius" {"selected" if session.data["preferred_unit"] == "celsius" else ""}>Celsius</option>
<option value="fahrenheit" {"selected" if session.data["preferred_unit"] == "fahrenheit" else ""}>Fahrenheit</option>
</select>
</label>
</form>
</div>
<div class="card">
<h2>Get Weather</h2>
<form actinotallow="/" method="get">
<select name="city">
{' '.join([f'<option value="{city}">{city}</option>' for city in WEATHER_DATA.keys()])}
</select>
<button type="submit">Get Weather</button>
</form>
</div>
{"<div class='card'><h2>Your Recent Searches</h2><ul>" +
''.join([f'<li><a href="/?city={city}">{city}</a></li>' for city in session.data["recent_searches"]]) +
"</ul></div>" if session.data["recent_searches"] else ""}
{"<div class='card'><h2>Weather Result</h2>" +
f"<h3>Weather in {query['city'][0]}</h3>" +
f"<p>Temperature: {WEATHER_DATA[query['city'][0]]['temperature']}°F " +
f"({round((WEATHER_DATA[query['city'][0]]['temperature'] - 32) * 5/9, 1)}°C)</p>" +
f"<p>Condition: {WEATHER_DATA[query['city'][0]]['condition']}</p>" +
f"<p>Humidity: {WEATHER_DATA[query['city'][0]]['humidity']}%</p></div>"
if 'city' in query and query['city'][0] in WEATHER_DATA else ""}
</body>
</html>
"""
self.wfile.write(html.encode())
def do_POST(self):
self._clean_expired_sessions()
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length)
request_json = json.loads(post_data.decode('utf-8'))
session = self._get_or_create_session()
if self.path == '/api/preferences':
# Update user preferences
for key, value in request_json.items():
session.set_preference(key, value)
self._set_headers()
response = {
"status": "success",
"message": "Preferences updated",
"context": session.to_dict()
}
self.wfile.write(json.dumps(response).encode())
else:
self.send_response(404)
self.end_headers()
self.wfile.write(json.dumps({"error": "Not found"}).encode())
def run_server(port=8000):
server_address = ('', port)
httpd = HTTPServer(server_address, MCPRequestHandler)
print(f"Starting MCP context-aware server on port {port}...")
httpd.serve_forever()
if __name__ == "__main__":
run_server()_set_headers() response = {"context": session.to_dict()} self.wfile.write(json.dumps(response).encode())
else:
# Serve a simple HTML interface for manual testing
self._set_headers('text/html')
html = f"""
<!DOCTYPE html>
<html>
<head>
<title>MCP Weather Service</title>
<style>
body {{ font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; }}
.card {{ border: 1px solid #ddd; border-radius: 8px; padding: 15px; margin-bottom: 15px; }}
.recent {{ color: #666; font-size: 0.9em; }}
</style>
</head>
<body>
<h1>MCP Weather Service</h1>
<div class="card">
<h2>Your Session</h2>
<p>Session ID: {session.session_id}</p>
<p>Number of visits: {session.data["visits"]}</p>
<p>Temperature unit preference: {session.data["preferred_unit"]}</p>
<form>
<label>
Change temperature unit:
<select name="unit" notallow="this.form.submit()">
<option value="celsius" {"selected" if session.data["preferred_unit"] == "celsius" else ""}>Celsius</option>
<option value="fahrenheit" {"selected" if session.data["preferred_unit"] == "fahrenheit" else ""}>Fahrenheit</option>
</select>
</label>
</form>
</div>
<div class="card">
<h2>Get Weather</h2>
<form actinotallow="/" method="get">
<select name="city">
{' '.join([f'<option value="{city}">{city}</option>' for city in WEATHER_DATA.keys()])}
</select>
<button type="submit">Get Weather</button>
</form>
</div>
{"<div class='card'><h2>Your Recent Searches</h2><ul>" +
''.join([f'<li><a href="/?city={city}">{city}</a></li>' for city in session.data["recent_searches"]]) +
"</ul></div>" if session.data["recent_searches"] else ""}
{"<div class='card'><h2>Weather Result</h2>" +
f"<h3>Weather in {query['city'][0]}</h3>" +
f"<p>Temperature: {WEATHER_DATA[query['city'][0]]['temperature']}°F " +
f"({round((WEATHER_DATA[query['city'][0]]['temperature'] - 32) * 5/9, 1)}°C)</p>" +
f"<p>Condition: {WEATHER_DATA[query['city'][0]]['condition']}</p>" +
f"<p>Humidity: {WEATHER_DATA[query['city'][0]]['humidity']}%</p></div>"
if 'city' in query and query['city'][0] in WEATHER_DATA else ""}
</body>
</html>
"""
self.wfile.write(html.encode())
def do_POST(self):
self._clean_expired_sessions()
content_length = int(self.headers['Content-Length'])
post_data = self.rfile.read(content_length)
request_json = json.loads(post_data.decode('utf-8'))
session = self._get_or_create_session()
if self.path == '/api/preferences':
# Update user preferences
for key, value in request_json.items():
session.set_preference(key, value)
self._set_headers()
response = {
"status": "success",
"message": "Preferences updated",
"context": session.to_dict()
}
self.wfile.write(json.dumps(response).encode())
else:
self.send_response(404)
self.end_headers()
self.wfile.write(json.dumps({"error": "Not found"}).encode())def run_server(port=8000): server_address = ('', port) httpd = HTTPServer(server_address, MCPRequestHandler) print(f"Starting MCP context-aware server on port {port}...") httpd.serve_forever()
if name == "main": run_server()
運行服務器的命令如下：
```bash
python3 mcp_context_server.py
當服務器運行時，終端窗口會顯示服務器正在運行的狀態，命令提示符不會返回，這表明服務器已經成功啟動。
（二）創建 MCP 客戶端
接下來，我們需要創建一個客戶端來與服務器交互。創建一個名為 ??mcp_context_client.py?? 的文件，并將以下代碼復制到文件中。
# mcp_context_client.py
import json
import requests
import sys
import os
class WeatherClient:
def __init__(self, server_url="http://localhost:8000"):
self.server_url = server_url
self.session = requests.Session() # Use a session to maintain cookies
self.context = None
# Try to load saved session ID from file
self.session_file = "session.json"
if os.path.exists(self.session_file):
try:
with open(self.session_file, "r") as f:
saved_data = json.load(f)
if"session_id"in saved_data:
self.session.cookies.set("session_id", saved_data["session_id"])
if"context"in saved_data:
self.context = saved_data["context"]
print("Restored previous session")
except Exception as e:
print(f"Error loading session: {e}")
def save_session(self):
"""Save the session ID and context to a file"""
if"session_id"in self.session.cookies:
session_data = {
"session_id": self.session.cookies.get("session_id"),
"context": self.context
}
with open(self.session_file, "w") as f:
json.dump(session_data, f)
def get_cities(self):
"""Get list of available cities"""
response = self.session.get(f"{self.server_url}/api/weather")
data = response.json()
# Update context if available
if"context"in data:
self.context = data["context"]
self.save_session()
return data["cities"]
def get_weather(self, city, unit=None):
"""Get weather for a specific city"""
url = f"{self.server_url}/api/weather?city={city}"
# Use stored preference or provided unit
if unit:
url += f"&unit={unit}"
elif self.context and"data"in self.context and"preferred_unit"in self.context["data"]:
url += f"&unit={self.context['data']['preferred_unit']}"
response = self.session.get(url)
data = response.json()
# Update context if available
if"context"in data:
self.context = data["context"]
self.save_session()
return data
def update_preferences(self, preferences):
"""Update user preferences"""
response = self.session.post(
f"{self.server_url}/api/preferences",
jsnotallow=preferences
)
data = response.json()
# Update context if available
if"context"in data:
self.context = data["context"]
self.save_session()
return data
def get_context(self):
"""Get current session context"""
ifnot self.context:
response = self.session.get(f"{self.server_url}/api/context")
data = response.json()
self.context = data["context"]
self.save_session()
return self.context
def display_weather(self, weather_data):
"""Display weather information nicely"""
if"error"in weather_data:
print(f"\nError: {weather_data['error']}")
return
city = weather_data["city"]
weather = weather_data["weather"]
unit = weather_data.get("unit", "celsius")
unit_symbol = "°F"if unit == "fahrenheit"else"°C"
print(f"\nWeather for {city}:")
print(f"Temperature: {weather['temperature']}{unit_symbol}")
print(f"Condition: {weather['condition']}")
print(f"Humidity: {weather['humidity']}%")
def display_context(self):
"""Display current context information"""
context = self.get_context()
ifnot context:
print("\nNo context available")
return
print("\n=== Your Session Context ===")
print(f"Session ID: {context['session_id']}")
print(f"Created: {context['created_at']}")
print(f"Last accessed: {context['last_accessed']}")
print("\nPreferences:")
print(f"Temperature unit: {context['data']['preferred_unit']}")
print(f"Total visits: {context['data']['visits']}")
if context['data']['recent_searches']:
print("\nRecent searches:")
for i, city in enumerate(context['data']['recent_searches'], 1):
print(f"{i}. {city}")
def run_client():
client = WeatherClient()
whileTrue:
print("\n--- MCP Weather Client with Context ---")
print("1. List all cities")
print("2. Get weather for a city")
print("3. Change temperature unit preference")
print("4. View session context")
print("5. Exit")
choice = input("Enter your choice (1-5): ")
if choice == "1":
cities = client.get_cities()
print("\nAvailable cities:")
for city in cities:
print(f"- {city}")
elif choice == "2":
city = input("Enter city name: ")
try:
weather_data = client.get_weather(city)
client.display_weather(weather_data)
except Exception as e:
print(f"Error getting weather: {e}")
elif choice == "3":
unit = input("Choose temperature unit (celsius/fahrenheit): ").lower()
if unit in ["celsius", "fahrenheit"]:
try:
result = client.update_preferences({"preferred_unit": unit})
print(f"Temperature unit updated to {unit}")
except Exception as e:
print(f"Error updating preferences: {e}")
else:
print("Invalid unit. Please enter 'celsius' or 'fahrenheit'.")
elif choice == "4":
client.display_context()
elif choice == "5":
print("Exiting weather client.")
sys.exit(0)
else:
print("Invalid choice. Please try again.")
if __name__ == "__main__":
run_client()
運行客戶端的命令如下：
```bash
python3 mcp_context_client.py
通過這個簡單的客戶端，你可以與服務器進行交互，查詢天氣信息、更改溫度單位偏好設置，以及查看當前會話的上下文信息。這不僅展示了 MCP 在上下文管理方面的強大功能，還體現了其在實際應用中的靈活性和實用性。
五、MCP 上下文管理的關鍵特性
通過上述服務器和客戶端的實現，我們可以總結出 MCP 在上下文管理方面的幾個關鍵特性：
（一）為每個客戶端創建唯一的會話 ID
服務器會為每個連接的客戶端生成一個唯一的會話 ID，并通過 cookie 機制將其傳遞給客戶端。這個會話 ID 是上下文管理的核心，它使得服務器能夠識別和跟蹤每個用戶的會話狀態。
（二）使用 cookie 維護會話
通過 cookie，客戶端在每次請求時都會攜帶會話 ID，服務器可以根據這個 ID 恢復用戶的上下文信息。這種方式不僅簡單高效，還能確保會話信息在多次請求之間的連續性。
（三）上下文數據跨多次請求持久化
在 MCP 的上下文管理中，用戶的偏好設置（如溫度單位）、使用歷史（如最近搜索的城市）以及會話統計信息（如訪問次數、會話創建時間）等數據都會被持久化。這意味著即使用戶關閉客戶端后重新連接，服務器依然能夠恢復之前的上下文狀態。
（四）上下文數據元素
- 用戶偏好設置：例如溫度單位的選擇（攝氏或華氏），這些偏好設置會根據用戶的操作進行更新，并在后續的交互中被應用。
- 使用歷史：記錄用戶的最近搜索記錄，方便用戶快速查找之前查詢過的信息。
- 會話統計信息：包括會話的創建時間、最后訪問時間以及訪問次數等，這些信息有助于服務器進行會話管理和優化。
六、MCP 的實際應用場景與未來展望
（一）實際應用場景
MCP 的上下文管理能力使其在多種實際應用場景中具有巨大的潛力。例如：
- 智能客服系統：通過維護用戶的對話歷史和偏好設置，MCP 可以讓智能客服系統提供更加個性化和連貫的服務。用戶無需重復說明問題，系統能夠根據上下文直接提供解決方案。
- 智能語音助手：在語音交互場景中，MCP 可以幫助語音助手更好地理解用戶的意圖，并根據上下文提供更準確的回答。例如，用戶可以連續提問，而助手能夠根據之前的對話內容進行回答。
- 多模態應用：在結合文本、圖像和語音的多模態應用中，MCP 的上下文管理能力可以確保不同模態之間的信息能夠無縫銜接，提升用戶體驗。
（二）未來展望
隨著人工智能技術的不斷發展，MCP 的上下文管理能力將為語言模型的應用帶來更多的可能性。未來，我們可以期待：
- 更高效的上下文壓縮技術：隨著模型規模的增大，上下文窗口的限制將更加明顯。MCP 將不斷優化上下文壓縮技術，以更好地利用有限的上下文空間。
- 跨平臺的上下文共享：MCP 將進一步擴展其跨平臺能力，使得用戶在不同的設備和應用之間能夠無縫切換，同時保持一致的上下文體驗。
- 與更多工具的集成：MCP 的工具連接能力將與上下文管理能力深度融合，為用戶提供更加智能化和自動化的解決方案。
七、總結
Model Context Protocol（MCP）不僅僅是一個工具集成的協議，更是一個強大的上下文管理框架。它通過解決語言模型在上下文窗口限制、有狀態對話、記憶管理、跨會話共享、結構化知識表示和檢索增強等方面的痛點，為人工智能的應用帶來了全新的可能性。
通過實際的服務器和客戶端示例，我們看到了 MCP 在上下文管理方面的強大功能。它不僅能夠為用戶提供更加個性化和連貫的交互體驗，還能在多種實際應用場景中發揮重要作用。
隨著技術的不斷進步，MCP 的未來充滿了無限可能。它將不斷優化和擴展其功能，為人工智能的發展提供更加堅實的基礎。讓我們拭目以待，看看 MCP 將如何改變我們與人工智能的交互方式。
如果你對這篇文章感興趣，不妨在 Medium 上給我點贊，或者在 LinkedIn 和 X 上關注我，獲取更多關于 MCP 和人工智能的最新資訊。
本文轉載自公眾號Halo咯咯 作者：基咯咯
原文鏈接：??https://mp.weixin.qq.com/s/f2v0M5Dt53MXtEh_zRrl0w??
?著作權歸作者所有，如需轉載，請注明出處，否則將追究法律責任
已于2025-5-6 09:45:25修改
贊
收藏
回復
分享
微博
QQ
微信
舉報
回復
相關推薦
