# 不同模型的统一接口示例from langchain.llms import OpenAI, Anthropicfrom langchain.chat_models import ChatOpenAI# 普通补全模型llm = OpenAI(model_name="text-davinci-003")# 聊天模型chat_model = ChatOpenAI(model_name="gpt-3.5-turbo")# 尽管底层实现不同，但提供类似的调用接口result1 = llm("请解释量子计算")result2 = chat_model.predict("请解释量子计算")

from langchain.prompts import PromptTemplate, FewShotPromptTemplate# 基础提示模板template = """你是一个专业的{domain}专家。请用{style}风格回答以下问题：
问题：{question}
回答："""prompt = PromptTemplate.from_template(template)# 小样本学习提示examples = [
    {"input": "太阳是什么？", "output": "太阳是太阳系的中心恒星。"},
    {"input": "月亮是什么？", "output": "地球的天然卫星。"}]example_prompt = PromptTemplate(
    input_variables=["input", "output"],
    template="输入：{input}\n输出：{output}")few_shot_prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,m.pzekj.com    
    prefix="请根据以下示例回答问题：",
    suffix="问题：{input}\n回答：",wap.pzekj.com    
    input_variables=["input"])m.urdkj.com

from langchain.chains import LLMChain, SequentialChain# 简单链simple_chain = LLMChain(llm=llm, prompt=prompt)# 复杂顺序链chain1 = LLMChain(llm=llm, prompt=prompt1, output_key="summary")chain2 = LLMChain(llm=llm, prompt=prompt2, output_key="analysis")chain3 = LLMChain(llm=llm, prompt=prompt3, output_key="recommendation")overall_chain = SequentialChain(
    chains=[chain1, chain2, chain3],zuqiu.xtokj.com    
    input_variables=["input_text"],m.xtokj.com    
    output_variables=["summary", "analysis", "recommendation"])www.xtokj.com

from langchain.agents import initialize_agent, Toolfrom langchain.tools import BaseTooldef search_api(query):
    # 调用搜索API的实现
    return resultsdef calculator(expression):
    # 计算表达式的实现
    return result
tools = [
    Tool(
        name="Web搜索",vip.xtokj.com    
        func=search_api,
        description="用于搜索最新信息"
    ),
    Tool(
        name="计算器",
        func=calculator,
        description="用于数学计算"
    )]agent = initialize_agent(
    tools, 
    llm, 
    agent="zero-shot-react-description", 
    verbose=True)result = agent.run("请计算圆周率的前五位然后搜索其应用场景")

from langchain.memory import ConversationBufferMemory, EntityMemory# 简单对话内存simple_memory = ConversationBufferMemory()# 实体记忆entity_memory = EntityMemory(llm=llm)# 在链中使用内存conversation_chain = ConversationChain(
    llm=llm,
    memory=ConversationBufferMemory(),
    verbose=True)

# 组件的独立测试def test_prompt_template():
    prompt = PromptTemplate.from_template("Hello {name}!")
    result = prompt.format(name="World")
    assert result == "Hello World!"

# 可重用的专业组件class LegalAnalyzer:
    def __init__(self, llm):
        self.chain = LLMChain(
            llm=llm,
            prompt=load_prompt("legal_analysis.yaml")
        )
    
    def analyze(self, document_text):
        return self.chain.run(document_text)

# 环境特定的配置管理import osfrom langchain.llms import OpenAIclass LLMConfig:
    @staticmethod
    def get_llm():
        model_name = os.getenv("LLM_MODEL", "gpt-3.5-turbo")
        api_key = os.getenv("OPENAI_API_KEY")
        temperature = float(os.getenv("LLM_TEMPERATURE", "0.7"))
        
        return OpenAI(
            model_name=model_name,
            openai_api_key=api_key,
            temperature=temperature        )# 配置化提示管理from langchain.prompts import load_prompt
prompt = load_prompt("prompts/customer_service.yaml")

# 异步支持提高吞吐量async def async_invoke_chain(chain, inputs):
    return await chain.arun(inputs)# 批量处理优化async def process_batch(chain, input_list):
    tasks = [chain.arun(input) for input in input_list]
    return await asyncio.gather(*tasks)# 缓存机制减少重复调用from langchain.cache import InMemoryCache
langchain.llm_cache = InMemoryCache()

# 集成监控工具from prometheus_client import Counter, Histogram
LLM_CALL_COUNT = Counter('llm_calls_total', 'Total LLM calls')LLM_CALL_DURATION = Histogram('llm_call_duration_seconds', 'LLM call duration')def monitored_llm_call(llm, prompt):
    start_time = time.time()
    LLM_CALL_COUNT.inc()
    
    try:
        result = llm(prompt)
        duration = time.time() - start_time
        LLM_CALL_DURATION.observe(duration)
        return result    except Exception as e:
        LOGGER.error(f"LLM call failed: {e}")
        raise

# 健壮的错误处理from tenacity import retry, stop_after_attempt, wait_exponential@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10))def robust_llm_call(llm, prompt):
    try:
        return llm(prompt)
    except RateLimitError:
        LOGGER.warning("Rate limit exceeded, retrying...")
        raise
    except TimeoutError:
        LOGGER.warning("Timeout occurred, retrying...")
        raise
    except Exception as e:
        LOGGER.error(f"Unexpected error: {e}")
        raise